- odl_process = None
- honeynode_process1 = None
- honeynode_process2 = None
- honeynode_process3 = None
- honeynode_process4 = None
- restconf_baseurl = "http://localhost:8181/restconf"
- WAITING = 20 #nominal value is 300
-
-#START_IGNORE_XTESTING
-
- @classmethod
- def setUpClass(cls):
- print ("starting honeynode1...")
- cls.honeynode_process1 = test_utils.start_xpdra_honeynode()
- time.sleep(20)
-
- print ("starting honeynode2...")
- cls.honeynode_process2 = test_utils.start_roadma_honeynode()
- time.sleep(20)
-
- print ("starting honeynode3...")
- cls.honeynode_process3 = test_utils.start_roadmc_honeynode()
- time.sleep(20)
-
- print ("starting honeynode4...")
- cls.honeynode_process4 = test_utils.start_xpdrc_honeynode()
- time.sleep(20)
- print ("all honeynodes started")
-
- print ("starting opendaylight...")
- cls.odl_process = test_utils.start_tpce()
- time.sleep(80)
- print ("opendaylight started")
-
- @classmethod
- def tearDownClass(cls):
- for child in psutil.Process(cls.odl_process.pid).children():
- child.send_signal(signal.SIGINT)
- child.wait()
- cls.odl_process.send_signal(signal.SIGINT)
- cls.odl_process.wait()
- for child in psutil.Process(cls.honeynode_process1.pid).children():
- child.send_signal(signal.SIGINT)
- child.wait()
- cls.honeynode_process1.send_signal(signal.SIGINT)
- cls.honeynode_process1.wait()
- for child in psutil.Process(cls.honeynode_process2.pid).children():
- child.send_signal(signal.SIGINT)
- child.wait()
- cls.honeynode_process2.send_signal(signal.SIGINT)
- cls.honeynode_process2.wait()
- for child in psutil.Process(cls.honeynode_process3.pid).children():
- child.send_signal(signal.SIGINT)
- child.wait()
- cls.honeynode_process3.send_signal(signal.SIGINT)
- cls.honeynode_process3.wait()
- for child in psutil.Process(cls.honeynode_process4.pid).children():
- child.send_signal(signal.SIGINT)
- child.wait()
- cls.honeynode_process4.send_signal(signal.SIGINT)
- cls.honeynode_process4.wait()
- print ("all processes killed")
-
- def setUp(self): # instruction executed before each test method
- print ("execution of {}".format(self.id().split(".")[-1]))
-
-#END_IGNORE_XTESTING
-
-# connect netconf devices
- def test_01_connect_xpdrA(self):
- url = ("{}/config/network-topology:"
- "network-topology/topology/topology-netconf/node/XPDR-A1"
- .format(self.restconf_baseurl))
- data = {"node": [{
- "node-id": "XPDR-A1",
- "netconf-node-topology:username": "admin",
- "netconf-node-topology:password": "admin",
- "netconf-node-topology:host": "127.0.0.1",
- "netconf-node-topology:port": "17840",
- "netconf-node-topology:tcp-only": "false",
- "netconf-node-topology:pass-through": {}}]}
- headers = {'content-type': 'application/json'}
- response = requests.request(
- "PUT", url, data=json.dumps(data), headers=headers,
- auth=('admin', 'admin'))
- self.assertEqual(response.status_code, requests.codes.created)
- time.sleep(20)
-
- def test_02_connect_xpdrC(self):
- url = ("{}/config/network-topology:"
- "network-topology/topology/topology-netconf/node/XPDR-C1"
- .format(self.restconf_baseurl))
- data = {"node": [{
- "node-id": "XPDR-C1",
- "netconf-node-topology:username": "admin",
- "netconf-node-topology:password": "admin",
- "netconf-node-topology:host": "127.0.0.1",
- "netconf-node-topology:port": "17844",
- "netconf-node-topology:tcp-only": "false",
- "netconf-node-topology:pass-through": {}}]}
- headers = {'content-type': 'application/json'}
- response = requests.request(
- "PUT", url, data=json.dumps(data), headers=headers,
- auth=('admin', 'admin'))
- self.assertEqual(response.status_code, requests.codes.created)
- time.sleep(20)
-
- def test_03_connect_rdmA(self):
- url = ("{}/config/network-topology:"
- "network-topology/topology/topology-netconf/node/ROADM-A1"
- .format(self.restconf_baseurl))
- data = {"node": [{
- "node-id": "ROADM-A1",
- "netconf-node-topology:username": "admin",
- "netconf-node-topology:password": "admin",
- "netconf-node-topology:host": "127.0.0.1",
- "netconf-node-topology:port": "17841",
- "netconf-node-topology:tcp-only": "false",
- "netconf-node-topology:pass-through": {}}]}
- headers = {'content-type': 'application/json'}
- response = requests.request(
- "PUT", url, data=json.dumps(data), headers=headers,
- auth=('admin', 'admin'))
- self.assertEqual(response.status_code, requests.codes.created)
- time.sleep(20)
-
- def test_04_connect_rdmC(self):
- url = ("{}/config/network-topology:"
- "network-topology/topology/topology-netconf/node/ROADM-C1"
- .format(self.restconf_baseurl))
- data = {"node": [{
- "node-id": "ROADM-C1",
- "netconf-node-topology:username": "admin",
- "netconf-node-topology:password": "admin",
- "netconf-node-topology:host": "127.0.0.1",
- "netconf-node-topology:port": "17843",
- "netconf-node-topology:tcp-only": "false",
- "netconf-node-topology:pass-through": {}}]}
- headers = {'content-type': 'application/json'}
- response = requests.request(
- "PUT", url, data=json.dumps(data), headers=headers,
- auth=('admin', 'admin'))
- self.assertEqual(response.status_code, requests.codes.created)
- time.sleep(20)
-
- def test_05_connect_xprdA_N1_to_roadmA_PP1(self):
- url = "{}/operations/transportpce-networkutils:init-xpdr-rdm-links".format(self.restconf_baseurl)
- data = {
- "networkutils:input": {
- "networkutils:links-input": {
- "networkutils:xpdr-node": "XPDR-A1",
- "networkutils:xpdr-num": "1",
- "networkutils:network-num": "1",
- "networkutils:rdm-node": "ROADM-A1",
- "networkutils:srg-num": "1",
- "networkutils:termination-point-num": "SRG1-PP1-TXRX"
- }
- }
- }
- headers = {'content-type': 'application/json'}
- response = requests.request(
- "POST", url, data=json.dumps(data),
- headers=headers, auth=('admin', 'admin'))
- self.assertEqual(response.status_code, requests.codes.ok)
- res = response.json()
- self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"])
- time.sleep(2)
-
- def test_06_connect_roadmA_PP1_to_xpdrA_N1(self):
- url = "{}/operations/transportpce-networkutils:init-rdm-xpdr-links".format(self.restconf_baseurl)
- data = {
- "networkutils:input": {
- "networkutils:links-input": {
- "networkutils:xpdr-node": "XPDR-A1",
- "networkutils:xpdr-num": "1",
- "networkutils:network-num": "1",
- "networkutils:rdm-node": "ROADM-A1",
- "networkutils:srg-num": "1",
- "networkutils:termination-point-num": "SRG1-PP1-TXRX"
- }
- }
- }
- headers = {'content-type': 'application/json'}
- response = requests.request(
- "POST", url, data=json.dumps(data),
- headers=headers, auth=('admin', 'admin'))
- self.assertEqual(response.status_code, requests.codes.ok)
- res = response.json()
- self.assertIn('Roadm Xponder links created successfully', res["output"]["result"])
- time.sleep(2)
-
- def test_07_connect_xprdC_N1_to_roadmC_PP1(self):
- url = "{}/operations/transportpce-networkutils:init-xpdr-rdm-links".format(self.restconf_baseurl)
- data = {
- "networkutils:input": {
- "networkutils:links-input": {
- "networkutils:xpdr-node": "XPDR-C1",
- "networkutils:xpdr-num": "1",
- "networkutils:network-num": "1",
- "networkutils:rdm-node": "ROADM-C1",
- "networkutils:srg-num": "1",
- "networkutils:termination-point-num": "SRG1-PP1-TXRX"
- }
- }
- }
- headers = {'content-type': 'application/json'}
- response = requests.request(
- "POST", url, data=json.dumps(data),
- headers=headers, auth=('admin', 'admin'))
- self.assertEqual(response.status_code, requests.codes.ok)
- res = response.json()
- self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"])
- time.sleep(2)
-
- def test_08_connect_roadmC_PP1_to_xpdrC_N1(self):
- url = "{}/operations/transportpce-networkutils:init-rdm-xpdr-links".format(self.restconf_baseurl)
- data = {
- "networkutils:input": {
- "networkutils:links-input": {
- "networkutils:xpdr-node": "XPDR-C1",
- "networkutils:xpdr-num": "1",
- "networkutils:network-num": "1",
- "networkutils:rdm-node": "ROADM-C1",
- "networkutils:srg-num": "1",
- "networkutils:termination-point-num": "SRG1-PP1-TXRX"
- }
- }
- }
- headers = {'content-type': 'application/json'}
- response = requests.request(
- "POST", url, data=json.dumps(data),
- headers=headers, auth=('admin', 'admin'))
- self.assertEqual(response.status_code, requests.codes.ok)
- res = response.json()
- self.assertIn('Roadm Xponder links created successfully', res["output"]["result"])
- time.sleep(2)
-
- def test_09_add_omsAttributes_ROADMA_ROADMC(self):
- # Config ROADMA-ROADMC oms-attributes
- url = ("{}/config/ietf-network:networks/network/openroadm-topology/ietf-network-topology:"
- "link/ROADM-A1-DEG2-DEG2-TTP-TXRXtoROADM-C1-DEG1-DEG1-TTP-TXRX/org-openroadm-network-topology:"
- "OMS-attributes/span"
- .format(self.restconf_baseurl))
- data = {"span": {
- "auto-spanloss": "true",
- "spanloss-base": 11.4,
- "spanloss-current": 12,
- "engineered-spanloss": 12.2,
- "link-concatenation": [{
- "SRLG-Id": 0,
- "fiber-type": "smf",
- "SRLG-length": 100000,
- "pmd": 0.5}]}}
- headers = {'content-type': 'application/json'}
- response = requests.request(
- "PUT", url, data=json.dumps(data), headers=headers,
- auth=('admin', 'admin'))
- self.assertEqual(response.status_code, requests.codes.created)
-
- def test_10_add_omsAttributes_ROADMC_ROADMA(self):
- # Config ROADMC-ROADMA oms-attributes
- url = ("{}/config/ietf-network:networks/network/openroadm-topology/ietf-network-topology:"
- "link/ROADM-C1-DEG1-DEG1-TTP-TXRXtoROADM-A1-DEG2-DEG2-TTP-TXRX/org-openroadm-network-topology:"
- "OMS-attributes/span"
- .format(self.restconf_baseurl))
- data = {"span": {
- "auto-spanloss": "true",
- "spanloss-base": 11.4,
- "spanloss-current": 12,
- "engineered-spanloss": 12.2,
- "link-concatenation": [{
- "SRLG-Id": 0,
- "fiber-type": "smf",
- "SRLG-length": 100000,
- "pmd": 0.5}]}}
- headers = {'content-type': 'application/json'}
- response = requests.request(
- "PUT", url, data=json.dumps(data), headers=headers,
- auth=('admin', 'admin'))
- self.assertEqual(response.status_code, requests.codes.created)
-
-#test service-create for Eth service from xpdr to xpdr
- def test_11_create_eth_service1(self):
- url = ("{}/operations/org-openroadm-service:service-create"
- .format(self.restconf_baseurl))
- data = {"input": {