self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
def test_05_connect_xprdA_N1_to_roadmA_PP1(self):
- url = "{}/operations/transportpce-networkutils:init-xpdr-rdm-links".\
- format(test_utils.RESTCONF_BASE_URL)
- data = {
- "networkutils:input": {
- "networkutils:links-input": {
- "networkutils:xpdr-node": "XPDRA01",
- "networkutils:xpdr-num": "1",
- "networkutils:network-num": "1",
- "networkutils:rdm-node": "ROADMA01",
- "networkutils:srg-num": "1",
- "networkutils:termination-point-num": "SRG1-PP1-TXRX"
- }
- }
- }
- response = requests.request(
- "POST", url, data=json.dumps(data),
- headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.connect_xpdr_to_rdm_request("XPDRA01", "1", "1",
+ "ROADMA01", "1", "SRG1-PP1-TXRX")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Xponder Roadm Link created successfully',
time.sleep(2)
def test_06_connect_roadmA_PP1_to_xpdrA_N1(self):
- url = "{}/operations/transportpce-networkutils:init-rdm-xpdr-links".\
- format(test_utils.RESTCONF_BASE_URL)
+ url = "{}/operations/transportpce-networkutils:init-rdm-xpdr-links"
data = {
"networkutils:input": {
"networkutils:links-input": {
}
}
}
- response = requests.request(
- "POST", url, data=json.dumps(data),
- headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.post_request(url, data)
+ response = test_utils.connect_rdm_to_xpdr_request("XPDRA01", "1", "1",
+ "ROADMA01", "1", "SRG1-PP1-TXRX")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Roadm Xponder links created successfully',
time.sleep(2)
def test_07_connect_xprdC_N1_to_roadmC_PP1(self):
- url = "{}/operations/transportpce-networkutils:init-xpdr-rdm-links".\
- format(test_utils.RESTCONF_BASE_URL)
- data = {
- "networkutils:input": {
- "networkutils:links-input": {
- "networkutils:xpdr-node": "XPDRC01",
- "networkutils:xpdr-num": "1",
- "networkutils:network-num": "1",
- "networkutils:rdm-node": "ROADMC01",
- "networkutils:srg-num": "1",
- "networkutils:termination-point-num": "SRG1-PP1-TXRX"
- }
- }
- }
- response = requests.request(
- "POST", url, data=json.dumps(data),
- headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.connect_xpdr_to_rdm_request("XPDRC01", "1", "1",
+ "ROADMC01", "1", "SRG1-PP1-TXRX")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Xponder Roadm Link created successfully',
time.sleep(2)
def test_08_connect_roadmC_PP1_to_xpdrC_N1(self):
- url = "{}/operations/transportpce-networkutils:init-rdm-xpdr-links".\
- format(test_utils.RESTCONF_BASE_URL)
- data = {
- "networkutils:input": {
- "networkutils:links-input": {
- "networkutils:xpdr-node": "XPDRC01",
- "networkutils:xpdr-num": "1",
- "networkutils:network-num": "1",
- "networkutils:rdm-node": "ROADMC01",
- "networkutils:srg-num": "1",
- "networkutils:termination-point-num": "SRG1-PP1-TXRX"
- }
- }
- }
- response = requests.request(
- "POST", url, data=json.dumps(data),
- headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.connect_rdm_to_xpdr_request("XPDRC01", "1", "1",
+ "ROADMC01", "1", "SRG1-PP1-TXRX")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Roadm Xponder links created successfully',
"link/ROADMA01-DEG1-DEG1-TTP-TXRXtoROADMC01-DEG2-DEG2-TTP-TXRX/"
"org-openroadm-network-topology:"
"OMS-attributes/span"
- .format(test_utils.RESTCONF_BASE_URL))
+ )
data = {"span": {
"clfi": "fiber1",
"auto-spanloss": "true",
"fiber-type": "smf",
"SRLG-length": 100000,
"pmd": 0.5}]}}
- response = requests.request(
- "PUT", url, data=json.dumps(data), headers=test_utils.TYPE_APPLICATION_JSON,
- auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.put_request(url, data)
self.assertEqual(response.status_code, requests.codes.created)
def test_10_add_omsAttributes_ROADMC_ROADMA(self):
"link/ROADMC01-DEG2-DEG2-TTP-TXRXtoROADMA01-DEG1-DEG1-TTP-TXRX/"
"org-openroadm-network-topology:"
"OMS-attributes/span"
- .format(test_utils.RESTCONF_BASE_URL))
+ )
data = {"span": {
"clfi": "fiber1",
"auto-spanloss": "true",
"fiber-type": "smf",
"SRLG-length": 100000,
"pmd": 0.5}]}}
- response = requests.request(
- "PUT", url, data=json.dumps(data), headers=test_utils.TYPE_APPLICATION_JSON,
- auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.put_request(url, data)
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(test_utils.RESTCONF_BASE_URL))
+ url = "{}/operations/org-openroadm-service:service-create"
data = {
"input": {
"sdnc-request-header": {
"operator-contact": "pw1234"
}
}
- response = requests.request(
- "POST", url, data=json.dumps(data), headers=test_utils.TYPE_APPLICATION_JSON,
- auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.post_request(url, data)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('PCE calculation in progress',
time.sleep(self.WAITING)
def test_12_get_eth_service1(self):
- url = ("{}/operational/org-openroadm-service:service-list/services/"
- "service1".format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/operational/org-openroadm-service:service-list/services/service1"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual(
"node/ROADMA01/yang-ext:"
"mount/org-openroadm-device:org-openroadm-device/"
"roadm-connections/SRG1-PP1-TXRX-DEG1-TTP-TXRX-1"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ )
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
# the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
"node/ROADMC01/yang-ext:mount/org-openroadm-device:"
"org-openroadm-device/"
"roadm-connections/SRG1-PP1-TXRX-DEG2-TTP-TXRX-1"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ )
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
# the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
time.sleep(5)
def test_15_check_topo_XPDRA(self):
- url1 = (
- "{}/config/ietf-network:"
- "networks/network/openroadm-topology/node/XPDRA01-XPDR1"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url1, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/ietf-network:networks/network/openroadm-topology/node/XPDRA01-XPDR1"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
liste_tp = res['node'][0]['ietf-network-topology:termination-point']
time.sleep(3)
def test_16_check_topo_ROADMA_SRG1(self):
- url1 = (
- "{}/config/ietf-network:"
- "networks/network/openroadm-topology/node/ROADMA01-SRG1"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url1, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/ietf-network:networks/network/openroadm-topology/node/ROADMA01-SRG1"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertNotIn({u'index': 1},
time.sleep(3)
def test_17_check_topo_ROADMA_DEG1(self):
- url1 = (
- "{}/config/ietf-network:"
- "networks/network/openroadm-topology/node/ROADMA01-DEG1"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url1, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/ietf-network:networks/network/openroadm-topology/node/ROADMA01-DEG1"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertNotIn({u'index': 1},
time.sleep(3)
def test_18_connect_xprdA_N2_to_roadmA_PP2(self):
- url = "{}/operations/transportpce-networkutils:init-xpdr-rdm-links".\
- format(test_utils.RESTCONF_BASE_URL)
- data = {
- "networkutils:input": {
- "networkutils:links-input": {
- "networkutils:xpdr-node": "XPDRA01",
- "networkutils:xpdr-num": "1",
- "networkutils:network-num": "2",
- "networkutils:rdm-node": "ROADMA01",
- "networkutils:srg-num": "1",
- "networkutils:termination-point-num": "SRG1-PP2-TXRX"
- }
- }
- }
- response = requests.request(
- "POST", url, data=json.dumps(data),
- headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.connect_xpdr_to_rdm_request("XPDRA01", "1", "2",
+ "ROADMA01", "1", "SRG1-PP2-TXRX")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Xponder Roadm Link created successfully',
time.sleep(2)
def test_19_connect_roadmA_PP2_to_xpdrA_N2(self):
- url = "{}/operations/transportpce-networkutils:init-rdm-xpdr-links".\
- format(test_utils.RESTCONF_BASE_URL)
- data = {
- "networkutils:input": {
- "networkutils:links-input": {
- "networkutils:xpdr-node": "XPDRA01",
- "networkutils:xpdr-num": "1",
- "networkutils:network-num": "2",
- "networkutils:rdm-node": "ROADMA01",
- "networkutils:srg-num": "1",
- "networkutils:termination-point-num": "SRG1-PP2-TXRX"
- }
- }
- }
- response = requests.request(
- "POST", url, data=json.dumps(data),
- headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.connect_rdm_to_xpdr_request("XPDRA01", "1", "2",
+ "ROADMA01", "1", "SRG1-PP2-TXRX")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Roadm Xponder links created successfully',
time.sleep(2)
def test_20_connect_xprdC_N2_to_roadmC_PP2(self):
- url = "{}/operations/transportpce-networkutils:init-xpdr-rdm-links".\
- format(test_utils.RESTCONF_BASE_URL)
- data = {
- "networkutils:input": {
- "networkutils:links-input": {
- "networkutils:xpdr-node": "XPDRC01",
- "networkutils:xpdr-num": "1",
- "networkutils:network-num": "2",
- "networkutils:rdm-node": "ROADMC01",
- "networkutils:srg-num": "1",
- "networkutils:termination-point-num": "SRG1-PP2-TXRX"
- }
- }
- }
- response = requests.request(
- "POST", url, data=json.dumps(data),
- headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.connect_xpdr_to_rdm_request("XPDRC01", "1", "2",
+ "ROADMC01", "1", "SRG1-PP2-TXRX")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Xponder Roadm Link created successfully',
time.sleep(2)
def test_21_connect_roadmC_PP2_to_xpdrC_N2(self):
- url = "{}/operations/transportpce-networkutils:init-rdm-xpdr-links".\
- format(test_utils.RESTCONF_BASE_URL)
- data = {
- "networkutils:input": {
- "networkutils:links-input": {
- "networkutils:xpdr-node": "XPDRC01",
- "networkutils:xpdr-num": "1",
- "networkutils:network-num": "2",
- "networkutils:rdm-node": "ROADMC01",
- "networkutils:srg-num": "1",
- "networkutils:termination-point-num": "SRG1-PP2-TXRX"
- }
- }
- }
- response = requests.request(
- "POST", url, data=json.dumps(data),
- headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.connect_rdm_to_xpdr_request("XPDRC01", "1", "2",
+ "ROADMC01", "1", "SRG1-PP2-TXRX")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Roadm Xponder links created successfully',
time.sleep(2)
def test_22_create_eth_service2(self):
- url = ("{}/operations/org-openroadm-service:service-create"
- .format(test_utils.RESTCONF_BASE_URL))
+ url = "{}/operations/org-openroadm-service:service-create"
data = {
"input": {
"sdnc-request-header": {
"operator-contact": "pw1234"
}
}
- response = requests.request(
- "POST", url, data=json.dumps(data), headers=test_utils.TYPE_APPLICATION_JSON,
- auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.post_request(url, data)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('PCE calculation in progress',
time.sleep(self.WAITING)
def test_23_get_eth_service2(self):
- url = ("{}/operational/org-openroadm-service:"
- "service-list/services/service2"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/operational/org-openroadm-service:service-list/services/service2"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual(
"node/ROADMA01/yang-ext:"
"mount/org-openroadm-device:org-openroadm-device/"
"roadm-connections/DEG1-TTP-TXRX-SRG1-PP2-TXRX-2"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ )
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
# the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
res['roadm-connections'][0]['destination'])
def test_25_check_topo_XPDRA(self):
- url1 = (
- "{}/config/ietf-network:"
- "networks/network/openroadm-topology/node/XPDRA01-XPDR1"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url1, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/ietf-network:networks/network/openroadm-topology/node/XPDRA01-XPDR1"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
liste_tp = res['node'][0]['ietf-network-topology:termination-point']
time.sleep(10)
def test_26_check_topo_ROADMA_SRG1(self):
- url1 = (
- "{}/config/ietf-network:"
- "networks/network/openroadm-topology/node/ROADMA01-SRG1"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url1, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/ietf-network:networks/network/openroadm-topology/node/ROADMA01-SRG1"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertNotIn({u'index': 1}, res['node'][0][
time.sleep(10)
def test_27_check_topo_ROADMA_DEG1(self):
- url1 = (
- "{}/config/ietf-network:"
- "networks/network/openroadm-topology/node/ROADMA01-DEG1"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url1, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/ietf-network:networks/network/openroadm-topology/node/ROADMA01-DEG1"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertNotIn({u'index': 1}, res['node'][0][
# creation service test on a non-available resource
def test_28_create_eth_service3(self):
- url = ("{}/operations/org-openroadm-service:service-create"
- .format(test_utils.RESTCONF_BASE_URL))
+ url = "{}/operations/org-openroadm-service:service-create"
data = {
"input": {
"sdnc-request-header": {
"operator-contact": "pw1234"
}
}
- response = requests.request(
- "POST", url, data=json.dumps(data), headers=test_utils.TYPE_APPLICATION_JSON,
- auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.post_request(url, data)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('PCE calculation in progress',
# contains 2 elements
def test_29_delete_eth_service3(self):
- url = ("{}/operations/org-openroadm-service:service-delete"
- .format(test_utils.RESTCONF_BASE_URL))
+ url = "{}/operations/org-openroadm-service:service-delete"
data = {"input": {
"sdnc-request-header": {
"request-id": "e3028bae-a90f-4ddd-a83f-cf224eba0e58",
}
}
}
- response = requests.request(
- "POST", url, data=json.dumps(data), headers=test_utils.TYPE_APPLICATION_JSON,
- auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.post_request(url, data)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Service \'service3\' does not exist in datastore',
time.sleep(20)
def test_30_delete_eth_service1(self):
- url = ("{}/operations/org-openroadm-service:service-delete"
- .format(test_utils.RESTCONF_BASE_URL))
+ url = "{}/operations/org-openroadm-service:service-delete"
data = {"input": {
"sdnc-request-header": {
"request-id": "e3028bae-a90f-4ddd-a83f-cf224eba0e58",
}
}
}
- response = requests.request(
- "POST", url, data=json.dumps(data), headers=test_utils.TYPE_APPLICATION_JSON,
- auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.post_request(url, data)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Renderer service delete in progress',
time.sleep(20)
def test_31_delete_eth_service2(self):
- url = ("{}/operations/org-openroadm-service:service-delete"
- .format(test_utils.RESTCONF_BASE_URL))
+ url = "{}/operations/org-openroadm-service:service-delete"
data = {"input": {
"sdnc-request-header": {
"request-id": "e3028bae-a90f-4ddd-a83f-cf224eba0e58",
}
}
}
- response = requests.request(
- "POST", url, data=json.dumps(data), headers=test_utils.TYPE_APPLICATION_JSON,
- auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.post_request(url, data)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Renderer service delete in progress',
"network-topology/topology/topology-netconf/"
"node/ROADMA01/yang-ext:"
"mount/org-openroadm-device:org-openroadm-device/"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ )
+ response = test_utils.get_request(url)
res = response.json()
self.assertEqual(response.status_code, requests.codes.ok)
self.assertNotIn('roadm-connections',
time.sleep(2)
def test_33_check_topo_XPDRA(self):
- url1 = (
- "{}/config/ietf-network:"
- "networks/network/openroadm-topology/node/XPDRA01-XPDR1"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url1, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/ietf-network:networks/network/openroadm-topology/node/XPDRA01-XPDR1"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
liste_tp = res['node'][0]['ietf-network-topology:termination-point']
time.sleep(10)
def test_34_check_topo_ROADMA_SRG1(self):
- url1 = (
- "{}/config/ietf-network:"
- "networks/network/openroadm-topology/node/ROADMA01-SRG1"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url1, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/ietf-network:networks/network/openroadm-topology/node/ROADMA01-SRG1"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn({u'index': 1}, res['node'][0][
time.sleep(10)
def test_35_check_topo_ROADMA_DEG1(self):
- url1 = (
- "{}/config/ietf-network:"
- "networks/network/openroadm-topology/node/ROADMA01-DEG1"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url1, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/ietf-network:networks/network/openroadm-topology/node/ROADMA01-DEG1"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn({u'index': 1}, res['node'][0][
# test service-create for Optical Channel (OC) service from srg-pp to srg-pp
def test_36_create_oc_service1(self):
- url = ("{}/operations/org-openroadm-service:service-create"
- .format(test_utils.RESTCONF_BASE_URL))
+ url = "{}/operations/org-openroadm-service:service-create"
data = {
"input": {
"sdnc-request-header": {
"operator-contact": "pw1234"
}
}
- response = requests.request(
- "POST", url, data=json.dumps(data), headers=test_utils.TYPE_APPLICATION_JSON,
- auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.post_request(url, data)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('PCE calculation in progress',
time.sleep(self.WAITING)
def test_37_get_oc_service1(self):
- url = ("{}/operational/org-openroadm-service:"
- "service-list/services/service1"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/operational/org-openroadm-service:service-list/services/service1"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual(
"node/ROADMA01/yang-ext:"
"mount/org-openroadm-device:org-openroadm-device/"
"roadm-connections/SRG1-PP1-TXRX-DEG1-TTP-TXRX-1"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ )
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
# the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
"node/ROADMC01/yang-ext:mount/org-openroadm-device:"
"org-openroadm-device/"
"roadm-connections/SRG1-PP1-TXRX-DEG2-TTP-TXRX-1"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ )
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
# the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
time.sleep(7)
def test_40_create_oc_service2(self):
- url = ("{}/operations/org-openroadm-service:service-create"
- .format(test_utils.RESTCONF_BASE_URL))
+ url = "{}/operations/org-openroadm-service:service-create"
data = {
"input": {
"sdnc-request-header": {
"operator-contact": "pw1234"
}
}
- response = requests.request(
- "POST", url, data=json.dumps(data), headers=test_utils.TYPE_APPLICATION_JSON,
- auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.post_request(url, data)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('PCE calculation in progress',
time.sleep(self.WAITING)
def test_41_get_oc_service2(self):
- url = ("{}/operational/org-openroadm-service:"
- "service-list/services/service2"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/operational/org-openroadm-service:service-list/services/service2"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual(
"node/ROADMA01/yang-ext:mount/org-openroadm-device:"
"org-openroadm-device/"
"roadm-connections/SRG1-PP2-TXRX-DEG1-TTP-TXRX-2"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ )
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
# the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
time.sleep(3)
def test_44_delete_oc_service1(self):
- url = ("{}/operations/org-openroadm-service:service-delete"
- .format(test_utils.RESTCONF_BASE_URL))
+ url = "{}/operations/org-openroadm-service:service-delete"
data = {"input": {
"sdnc-request-header": {
"request-id": "e3028bae-a90f-4ddd-a83f-cf224eba0e58",
}
}
}
- response = requests.request(
- "POST", url, data=json.dumps(data), headers=test_utils.TYPE_APPLICATION_JSON,
- auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.post_request(url, data)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Renderer service delete in progress',
time.sleep(20)
def test_45_delete_oc_service2(self):
- url = ("{}/operations/org-openroadm-service:service-delete"
- .format(test_utils.RESTCONF_BASE_URL))
+ url = "{}/operations/org-openroadm-service:service-delete"
data = {"input": {
"sdnc-request-header": {
"request-id": "e3028bae-a90f-4ddd-a83f-cf224eba0e58",
}
}
}
- response = requests.request(
- "POST", url, data=json.dumps(data), headers=test_utils.TYPE_APPLICATION_JSON,
- auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.post_request(url, data)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Renderer service delete in progress',
def test_46_get_no_oc_services(self):
print("start test")
url = ("{}/operational/org-openroadm-service:service-list"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ )
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.not_found)
res = response.json()
self.assertIn(
"network-topology/topology/topology-netconf"
"/node/ROADMA01/yang-ext:mount/org-openroadm-device:"
"org-openroadm-device/"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ )
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertNotIn(['roadm-connections'][0], res['org-openroadm-device'])
self.test_30_delete_eth_service1()
def test_50_loop_create_oc_service(self):
- url = ("{}/operational/org-openroadm-service:"
- "service-list/services/service1"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request("GET", url, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/operational/org-openroadm-service:service-list/services/service1"
+ response = test_utils.get_request(url)
if response.status_code != 404:
url = ("{}/operations/org-openroadm-service:service-delete"
- .format(test_utils.RESTCONF_BASE_URL))
+ )
data = {"input": {
"sdnc-request-header": {
"request-id": "e3028bae-a90f-4ddd-a83f-cf224eba0e58",
}
}
}
- requests.request("POST", url, data=json.dumps(data),
- headers=test_utils.TYPE_APPLICATION_JSON,
- auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD)
- )
+ test_utils.post_request(url, data)
time.sleep(5)
for i in range(1, 6):
# Mount the different topologies
def test_01_connect_clliNetwork(self):
- url = ("{}/config/ietf-network:networks/network/clli-network"
- .format(test_utils.RESTCONF_BASE_URL))
+ url = "{}/config/ietf-network:networks/network/clli-network"
topo_cllinet_file = "sample_configs/gnpy/clliNetwork.json"
if os.path.isfile(topo_cllinet_file):
with open(topo_cllinet_file, 'r') as clli_net:
- body = clli_net.read()
- response = requests.request(
- "PUT", url, data=body, headers=test_utils.TYPE_APPLICATION_JSON,
- auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ data = clli_net.read()
+ #TODO : review this os specific path and treat error with an else-statement
+ response = test_utils.rawput_request(url, data)
self.assertEqual(response.status_code, requests.codes.ok)
time.sleep(3)
def test_02_connect_openroadmNetwork(self):
- url = ("{}/config/ietf-network:networks/network/openroadm-network"
- .format(test_utils.RESTCONF_BASE_URL))
+ url = "{}/config/ietf-network:networks/network/openroadm-network"
topo_ordnet_file = "sample_configs/gnpy/openroadmNetwork.json"
if os.path.isfile(topo_ordnet_file):
with open(topo_ordnet_file, 'r') as ord_net:
- body = ord_net.read()
- response = requests.request(
- "PUT", url, data=body, headers=test_utils.TYPE_APPLICATION_JSON,
- auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ data = ord_net.read()
+ response = test_utils.rawput_request(url, data)
self.assertEqual(response.status_code, requests.codes.ok)
time.sleep(3)
def test_03_connect_openroadmTopology(self):
- url = ("{}/config/ietf-network:networks/network/openroadm-topology"
- .format(test_utils.RESTCONF_BASE_URL))
+ url = "{}/config/ietf-network:networks/network/openroadm-topology"
topo_ordtopo_file = "sample_configs/gnpy/openroadmTopology.json"
if os.path.isfile(topo_ordtopo_file):
with open(topo_ordtopo_file, 'r') as ord_topo:
- body = ord_topo.read()
- response = requests.request(
- "PUT", url, data=body, headers=test_utils.TYPE_APPLICATION_JSON,
- auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ data = ord_topo.read()
+ response = test_utils.rawput_request(url, data)
self.assertEqual(response.status_code, requests.codes.ok)
time.sleep(3)
# Path computed by PCE is feasible according to Gnpy
def test_04_path_computation_FeasibleWithPCE(self):
- url = ("{}/operations/transportpce-pce:path-computation-request"
- .format(test_utils.RESTCONF_BASE_URL))
- body = {
+ url = "{}/operations/transportpce-pce:path-computation-request"
+ data = {
"input": {
"service-name": "service-1",
"resource-reserve": "true",
}
}
}
- response = requests.request(
- "POST", url, data=json.dumps(body), headers=test_utils.TYPE_APPLICATION_JSON,
- auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.post_request(url, data)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual(res['output']['configuration-response-common'][
# Path computed by PCE is not feasible by GNPy and GNPy cannot find
# another one (low SNR)
def test_05_path_computation_FoundByPCE_NotFeasibleByGnpy(self):
- url = ("{}/operations/transportpce-pce:path-computation-request"
- .format(test_utils.RESTCONF_BASE_URL))
- body = {
+ url = "{}/operations/transportpce-pce:path-computation-request"
+ data = {
"input": {
"service-name": "service-2",
"resource-reserve": "true",
}
}
}
- response = requests.request(
- "POST", url, data=json.dumps(body), headers=test_utils.TYPE_APPLICATION_JSON,
- auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.post_request(url, data)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual(res['output']['configuration-response-common'][
# #PCE cannot find a path while GNPy finds a feasible one
def test_06_path_computation_NotFoundByPCE_FoundByGNPy(self):
- url = ("{}/operations/transportpce-pce:path-computation-request"
- .format(test_utils.RESTCONF_BASE_URL))
- body = {
+ url = "{}/operations/transportpce-pce:path-computation-request"
+ data = {
"input": {
"service-name": "service-3",
"resource-reserve": "true",
}
}
}
- response = requests.request(
- "POST", url, data=json.dumps(body), headers=test_utils.TYPE_APPLICATION_JSON,
- auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.post_request(url, data)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual(res['output']['configuration-response-common'][
# Not found path by PCE and GNPy cannot find another one
def test_07_path_computation_FoundByPCE_NotFeasibleByGnpy(self):
- url = ("{}/operations/transportpce-pce:path-computation-request"
- .format(test_utils.RESTCONF_BASE_URL))
- body = {
+ url = "{}/operations/transportpce-pce:path-computation-request"
+ data = {
"input": {
"service-name": "service-4",
"resource-reserve": "true",
}
}
}
- response = requests.request(
- "POST", url, data=json.dumps(body), headers=test_utils.TYPE_APPLICATION_JSON,
- auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.post_request(url, data)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual(res['output']['configuration-response-common'][
# Disconnect the different topologies
def test_08_disconnect_openroadmTopology(self):
- url = ("{}/config/ietf-network:networks/network/openroadm-topology"
- .format(test_utils.RESTCONF_BASE_URL))
- data = {}
- response = requests.request(
- "DELETE", url, data=json.dumps(data), headers=test_utils.TYPE_APPLICATION_JSON,
- auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/ietf-network:networks/network/openroadm-topology"
+ response = test_utils.delete_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
time.sleep(3)
def test_09_disconnect_openroadmNetwork(self):
- url = ("{}/config/ietf-network:networks/network/openroadm-network"
- .format(test_utils.RESTCONF_BASE_URL))
- data = {}
- response = requests.request(
- "DELETE", url, data=json.dumps(data), headers=test_utils.TYPE_APPLICATION_JSON,
- auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/ietf-network:networks/network/openroadm-network"
+ response = test_utils.delete_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
time.sleep(3)
def test_10_disconnect_clliNetwork(self):
- url = ("{}/config/ietf-network:networks/network/clli-network"
- .format(test_utils.RESTCONF_BASE_URL))
- data = {}
- response = requests.request(
- "DELETE", url, data=json.dumps(data), headers=test_utils.TYPE_APPLICATION_JSON,
- auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/ietf-network:networks/network/clli-network"
+ response = test_utils.delete_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
time.sleep(3)
import unittest
import time
import json
-#from unittest.result import failfast
import requests
from common import test_utils
self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
def test_05_connect_xprdA_to_roadmA(self):
- url = "{}/operations/transportpce-networkutils:init-xpdr-rdm-links".format(test_utils.RESTCONF_BASE_URL)
- data = {
- "networkutils:input": {
- "networkutils:links-input": {
- "networkutils:xpdr-node": "XPDRA01",
- "networkutils:xpdr-num": "1",
- "networkutils:network-num": "1",
- "networkutils:rdm-node": "ROADMA01",
- "networkutils:srg-num": "1",
- "networkutils:termination-point-num": "SRG1-PP1-TXRX"
- }
- }
- }
- response = requests.request(
- "POST", url, data=json.dumps(data),
- headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.connect_xpdr_to_rdm_request("XPDRA01", "1", "1",
+ "ROADMA01", "1", "SRG1-PP1-TXRX")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"])
def test_06_connect_roadmA_to_xpdrA(self):
- url = "{}/operations/transportpce-networkutils:init-rdm-xpdr-links".format(test_utils.RESTCONF_BASE_URL)
- data = {
- "networkutils:input": {
- "networkutils:links-input": {
- "networkutils:xpdr-node": "XPDRA01",
- "networkutils:xpdr-num": "1",
- "networkutils:network-num": "1",
- "networkutils:rdm-node": "ROADMA01",
- "networkutils:srg-num": "1",
- "networkutils:termination-point-num": "SRG1-PP1-TXRX"
- }
- }
- }
- response = requests.request(
- "POST", url, data=json.dumps(data),
- headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.connect_rdm_to_xpdr_request("XPDRA01", "1", "1",
+ "ROADMA01", "1", "SRG1-PP1-TXRX")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Roadm Xponder links created successfully', res["output"]["result"])
def test_07_connect_xprdC_to_roadmC(self):
- url = "{}/operations/transportpce-networkutils:init-xpdr-rdm-links".format(test_utils.RESTCONF_BASE_URL)
- data = {
- "networkutils:input": {
- "networkutils:links-input": {
- "networkutils:xpdr-node": "XPDRC01",
- "networkutils:xpdr-num": "1",
- "networkutils:network-num": "1",
- "networkutils:rdm-node": "ROADMC01",
- "networkutils:srg-num": "1",
- "networkutils:termination-point-num": "SRG1-PP1-TXRX"
- }
- }
- }
- response = requests.request(
- "POST", url, data=json.dumps(data),
- headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.connect_xpdr_to_rdm_request("XPDRC01", "1", "1",
+ "ROADMC01", "1", "SRG1-PP1-TXRX")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"])
def test_08_connect_roadmC_to_xpdrC(self):
- url = "{}/operations/transportpce-networkutils:init-rdm-xpdr-links".format(test_utils.RESTCONF_BASE_URL)
- data = {
- "networkutils:input": {
- "networkutils:links-input": {
- "networkutils:xpdr-node": "XPDRC01",
- "networkutils:xpdr-num": "1",
- "networkutils:network-num": "1",
- "networkutils:rdm-node": "ROADMC01",
- "networkutils:srg-num": "1",
- "networkutils:termination-point-num": "SRG1-PP1-TXRX"
- }
- }
- }
- response = requests.request(
- "POST", url, data=json.dumps(data),
- headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.connect_rdm_to_xpdr_request("XPDRC01", "1", "1",
+ "ROADMC01", "1", "SRG1-PP1-TXRX")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Roadm Xponder links created successfully', res["output"]["result"])
def test_09_create_OTS_ROADMA(self):
- url = "{}/operations/transportpce-device-renderer:create-ots-oms".format(test_utils.RESTCONF_BASE_URL)
+ url = "{}/operations/transportpce-device-renderer:create-ots-oms"
data = {
"input": {
"node-id": "ROADMA01",
"logical-connection-point": "DEG1-TTP-TXRX"
}
}
- response = requests.request(
- "POST", url, data=json.dumps(data),
- headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.post_request(url, data)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Interfaces OTS-DEG1-TTP-TXRX - OMS-DEG1-TTP-TXRX successfully created on node ROADMA01',
res["output"]["result"])
def test_10_create_OTS_ROADMC(self):
- url = "{}/operations/transportpce-device-renderer:create-ots-oms".format(test_utils.RESTCONF_BASE_URL)
+ url = "{}/operations/transportpce-device-renderer:create-ots-oms"
data = {
"input": {
"node-id": "ROADMC01",
"logical-connection-point": "DEG2-TTP-TXRX"
}
}
- response = requests.request(
- "POST", url, data=json.dumps(data),
- headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.post_request(url, data)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Interfaces OTS-DEG2-TTP-TXRX - OMS-DEG2-TTP-TXRX successfully created on node ROADMC01',
res["output"]["result"])
def test_11_get_PM_ROADMA(self):
- url = "{}/operations/transportpce-olm:get-pm".format(test_utils.RESTCONF_BASE_URL)
+ url = "{}/operations/transportpce-olm:get-pm"
data = {
"input": {
"node-id": "ROADMA01",
}
}
}
- response = requests.request(
- "POST", url, data=json.dumps(data),
- headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.post_request(url, data)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn({
}, res["output"]["measurements"])
def test_12_get_PM_ROADMC(self):
- url = "{}/operations/transportpce-olm:get-pm".format(test_utils.RESTCONF_BASE_URL)
+ url = "{}/operations/transportpce-olm:get-pm"
data = {
"input": {
"node-id": "ROADMC01",
}
}
}
- response = requests.request(
- "POST", url, data=json.dumps(data),
- headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.post_request(url, data)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn({
}, res["output"]["measurements"])
def test_13_calculate_span_loss_base_ROADMA_ROADMC(self):
- url = "{}/operations/transportpce-olm:calculate-spanloss-base".format(test_utils.RESTCONF_BASE_URL)
+ url = "{}/operations/transportpce-olm:calculate-spanloss-base"
data = {
"input": {
"src-type": "link",
"link-id": "ROADMA01-DEG1-DEG1-TTP-TXRXtoROADMC01-DEG2-DEG2-TTP-TXRX"
}
}
- response = requests.request(
- "POST", url, data=json.dumps(data),
- headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.post_request(url, data)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Success',
time.sleep(5)
def test_14_calculate_span_loss_base_all(self):
- url = "{}/operations/transportpce-olm:calculate-spanloss-base".format(test_utils.RESTCONF_BASE_URL)
+ url = "{}/operations/transportpce-olm:calculate-spanloss-base"
data = {
"input": {
"src-type": "all"
}
}
- response = requests.request(
- "POST", url, data=json.dumps(data),
- headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.post_request(url, data)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Success',
def test_15_get_OTS_DEG1_TTP_TXRX_ROADMA(self):
url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
"node/ROADMA01/yang-ext:mount/org-openroadm-device:org-openroadm-device/interface/OTS-DEG1-TTP-TXRX/"
- "org-openroadm-optical-transport-interfaces:ots".format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ "org-openroadm-optical-transport-interfaces:ots")
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual(5.7, res['org-openroadm-optical-transport-interfaces:ots']['span-loss-transmit'])
def test_16_get_OTS_DEG2_TTP_TXRX_ROADMC(self):
url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
"node/ROADMC01/yang-ext:mount/org-openroadm-device:org-openroadm-device/interface/OTS-DEG2-TTP-TXRX/"
- "org-openroadm-optical-transport-interfaces:ots".format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ "org-openroadm-optical-transport-interfaces:ots")
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual(15.1, res['org-openroadm-optical-transport-interfaces:ots']['span-loss-transmit'])
self.assertEqual(5.7, res['org-openroadm-optical-transport-interfaces:ots']['span-loss-receive'])
def test_17_servicePath_create_AToZ(self):
- url = "{}/operations/transportpce-device-renderer:service-path".format(test_utils.RESTCONF_BASE_URL)
+ url = "{}/operations/transportpce-device-renderer:service-path"
data = {
"input": {
"service-name": "test",
]
}
}
- response = requests.request(
- "POST", url, data=json.dumps(data),
- headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.post_request(url, data)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Roadm-connection successfully created for nodes', res["output"]["result"])
time.sleep(10)
def test_18_servicePath_create_ZToA(self):
- url = "{}/operations/transportpce-device-renderer:service-path".format(test_utils.RESTCONF_BASE_URL)
+ url = "{}/operations/transportpce-device-renderer:service-path"
data = {
"input": {
"service-name": "test",
]
}
}
- response = requests.request(
- "POST", url, data=json.dumps(data),
- headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.post_request(url, data)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Roadm-connection successfully created for nodes', res["output"]["result"])
time.sleep(10)
def test_19_service_power_setup_XPDRA_XPDRC(self):
- url = "{}/operations/transportpce-olm:service-power-setup".format(test_utils.RESTCONF_BASE_URL)
+ url = "{}/operations/transportpce-olm:service-power-setup"
data = {
"input": {
"service-name": "test",
]
}
}
- response = requests.request(
- "POST", url, data=json.dumps(data),
- headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.post_request(url, data)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Success', res["output"]["result"])
def test_20_get_interface_XPDRA_XPDR1_NETWORK1(self):
url = ("{}/config/network-topology:network-topology/topology/topology-netconf/node/XPDRA01/yang-ext:mount/"
"org-openroadm-device:org-openroadm-device/interface/XPDR1-NETWORK1-1/"
- "org-openroadm-optical-channel-interfaces:och".format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ "org-openroadm-optical-channel-interfaces:och")
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual(0, res['org-openroadm-optical-channel-interfaces:och']['transmit-power'])
def test_21_get_roadmconnection_ROADMA(self):
url = ("{}/config/network-topology:network-topology/topology/topology-netconf/node/ROADMA01/yang-ext:mount/"
"org-openroadm-device:org-openroadm-device/roadm-connections/"
- "SRG1-PP1-TXRX-DEG1-TTP-TXRX-1".format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ "SRG1-PP1-TXRX-DEG1-TTP-TXRX-1")
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual("gainLoss", res['roadm-connections'][0]['opticalControlMode'])
def test_22_get_roadmconnection_ROADMC(self):
url = ("{}/config/network-topology:network-topology/topology/topology-netconf/node/ROADMC01/yang-ext:mount/"
"org-openroadm-device:org-openroadm-device/roadm-connections/"
- "DEG2-TTP-TXRX-SRG1-PP1-TXRX-1".format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ "DEG2-TTP-TXRX-SRG1-PP1-TXRX-1")
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual("power", res['roadm-connections'][0]['opticalControlMode'])
def test_23_service_power_setup_XPDRC_XPDRA(self):
- url = "{}/operations/transportpce-olm:service-power-setup".format(test_utils.RESTCONF_BASE_URL)
+ url = "{}/operations/transportpce-olm:service-power-setup"
data = {
"input": {
"service-name": "test",
]
}
}
- response = requests.request(
- "POST", url, data=json.dumps(data),
- headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.post_request(url, data)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Success', res["output"]["result"])
def test_24_get_interface_XPDRC_XPDR1_NETWORK1(self):
url = ("{}/config/network-topology:network-topology/topology/topology-netconf/node/XPDRC01/yang-ext:mount/"
"org-openroadm-device:org-openroadm-device/interface/XPDR1-NETWORK1-1/"
- "org-openroadm-optical-channel-interfaces:och".format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ "org-openroadm-optical-channel-interfaces:och")
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual(0, res['org-openroadm-optical-channel-interfaces:och']['transmit-power'])
def test_25_get_roadmconnection_ROADMC(self):
url = ("{}/config/network-topology:network-topology/topology/topology-netconf/node/ROADMC01/yang-ext:mount/"
"org-openroadm-device:org-openroadm-device/roadm-connections/"
- "SRG1-PP1-TXRX-DEG2-TTP-TXRX-1".format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ "SRG1-PP1-TXRX-DEG2-TTP-TXRX-1")
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual("gainLoss", res['roadm-connections'][0]['opticalControlMode'])
self.assertEqual(2, res['roadm-connections'][0]['target-output-power'])
def test_26_service_power_turndown_XPDRA_XPDRC(self):
- url = "{}/operations/transportpce-olm:service-power-turndown".format(test_utils.RESTCONF_BASE_URL)
+ url = "{}/operations/transportpce-olm:service-power-turndown"
data = {
"input": {
"service-name": "test",
]
}
}
- response = requests.request(
- "POST", url, data=json.dumps(data),
- headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.post_request(url, data)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Success', res["output"]["result"])
def test_27_get_roadmconnection_ROADMA(self):
url = ("{}/config/network-topology:network-topology/topology/topology-netconf/node/ROADMA01/yang-ext:mount/"
"org-openroadm-device:org-openroadm-device/roadm-connections/"
- "SRG1-PP1-TXRX-DEG1-TTP-TXRX-1".format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ "SRG1-PP1-TXRX-DEG1-TTP-TXRX-1")
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual("off", res['roadm-connections'][0]['opticalControlMode'])
def test_28_get_roadmconnection_ROADMC(self):
url = ("{}/config/network-topology:network-topology/topology/topology-netconf/node/ROADMC01/yang-ext:mount/"
"org-openroadm-device:org-openroadm-device/roadm-connections/"
- "DEG2-TTP-TXRX-SRG1-PP1-TXRX-1".format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ "DEG2-TTP-TXRX-SRG1-PP1-TXRX-1")
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual("off", res['roadm-connections'][0]['opticalControlMode'])
def test_29_servicePath_delete_AToZ(self):
- url = "{}/operations/transportpce-device-renderer:service-path".format(test_utils.RESTCONF_BASE_URL)
+ url = "{}/operations/transportpce-device-renderer:service-path"
data = {
"input": {
"service-name": "test",
]
}
}
- response = requests.request(
- "POST", url, data=json.dumps(data),
- headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.post_request(url, data)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Request processed', res["output"]["result"])
time.sleep(10)
def test_30_servicePath_delete_ZToA(self):
- url = "{}/operations/transportpce-device-renderer:service-path".format(test_utils.RESTCONF_BASE_URL)
+ url = "{}/operations/transportpce-device-renderer:service-path"
data = {
"input": {
"service-name": "test",
]
}
}
- response = requests.request(
- "POST", url, data=json.dumps(data),
- headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.post_request(url, data)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Request processed', res["output"]["result"])
"""to test case where SRG where the xpdr is connected to has no optical range data"""
def test_31_connect_xprdA_to_roadmA(self):
- url = "{}/operations/transportpce-networkutils:init-xpdr-rdm-links".format(test_utils.RESTCONF_BASE_URL)
- data = {
- "networkutils:input": {
- "networkutils:links-input": {
- "networkutils:xpdr-node": "XPDRA01",
- "networkutils:xpdr-num": "1",
- "networkutils:network-num": "2",
- "networkutils:rdm-node": "ROADMA01",
- "networkutils:srg-num": "1",
- "networkutils:termination-point-num": "SRG1-PP2-TXRX"
- }
- }
- }
- response = requests.request(
- "POST", url, data=json.dumps(data),
- headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.connect_xpdr_to_rdm_request("XPDRA01", "1", "2",
+ "ROADMA01", "1", "SRG1-PP2-TXRX")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"])
def test_32_connect_roadmA_to_xpdrA(self):
- url = "{}/operations/transportpce-networkutils:init-rdm-xpdr-links".format(test_utils.RESTCONF_BASE_URL)
- data = {
- "networkutils:input": {
- "networkutils:links-input": {
- "networkutils:xpdr-node": "XPDRA01",
- "networkutils:xpdr-num": "1",
- "networkutils:network-num": "2",
- "networkutils:rdm-node": "ROADMA01",
- "networkutils:srg-num": "1",
- "networkutils:termination-point-num": "SRG1-PP2-TXRX"
- }
- }
- }
- response = requests.request(
- "POST", url, data=json.dumps(data),
- headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.connect_rdm_to_xpdr_request("XPDRA01", "1", "2",
+ "ROADMA01", "1", "SRG1-PP2-TXRX")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Roadm Xponder links created successfully', res["output"]["result"])
def test_33_servicePath_create_AToZ(self):
- url = "{}/operations/transportpce-device-renderer:service-path".format(test_utils.RESTCONF_BASE_URL)
+ url = "{}/operations/transportpce-device-renderer:service-path"
data = {
"input": {
"service-name": "test2",
]
}
}
- response = requests.request(
- "POST", url, data=json.dumps(data),
- headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.post_request(url, data)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Roadm-connection successfully created for nodes', res["output"]["result"])
def test_34_get_interface_XPDRA_XPDR1_NETWORK2(self):
url = ("{}/config/network-topology:network-topology/topology/topology-netconf/node/XPDRA01/yang-ext:mount/"
"org-openroadm-device:org-openroadm-device/interface/XPDR1-NETWORK2-2/"
- "org-openroadm-optical-channel-interfaces:och".format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ "org-openroadm-optical-channel-interfaces:och")
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual(-5, res['org-openroadm-optical-channel-interfaces:och']['transmit-power'])
self.assertEqual(2, res['org-openroadm-optical-channel-interfaces:och']['wavelength-number'])
def test_35_servicePath_delete_AToZ(self):
- url = "{}/operations/transportpce-device-renderer:service-path".format(test_utils.RESTCONF_BASE_URL)
+ url = "{}/operations/transportpce-device-renderer:service-path"
data = {
"input": {
"service-name": "test",
]
}
}
- response = requests.request(
- "POST", url, data=json.dumps(data),
- headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.post_request(url, data)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Request processed', res["output"]["result"])
self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
def test_38_calculate_span_loss_current(self):
- url = "{}/operations/transportpce-olm:calculate-spanloss-current".format(test_utils.RESTCONF_BASE_URL)
- response = requests.request(
- "POST", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/operations/transportpce-olm:calculate-spanloss-current"
+ response = test_utils.post_request(url, None)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Success',
# Load simple bidirectional topology
def test_01_load_simple_topology_bi(self):
- url = ("{}/config/ietf-network:networks/network/openroadm-topology"
- .format(test_utils.RESTCONF_BASE_URL))
- body = self.simple_topo_bi_dir_data
- response = requests.request(
- "PUT", url, data=body, headers=test_utils.TYPE_APPLICATION_XML,
- auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/ietf-network:networks/network/openroadm-topology"
+ data = self.simple_topo_bi_dir_data
+ response = test_utils.put_xmlrequest(url, data)
self.assertEqual(response.status_code, requests.codes.ok)
time.sleep(2)
# Get existing nodeId
def test_02_get_nodeId(self):
- url = ("{}/config/ietf-network:networks/network/openroadm-topology/node/ROADMA01-SRG1"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/ietf-network:networks/network/openroadm-topology/node/ROADMA01-SRG1"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual(
# Get existing linkId
def test_03_get_linkId(self):
- url = ("{}/config/ietf-network:networks/network/openroadm-topology/link/XPDRA01-XPDR1-XPDR1-NETWORK1toROADMA01-SRG1-SRG1-PP1-TXRX"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/ietf-network:networks/network/openroadm-topology/link/XPDRA01-XPDR1-XPDR1-NETWORK1toROADMA01-SRG1-SRG1-PP1-TXRX"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual(
# Path Computation success
def test_04_path_computation_xpdr_bi(self):
- url = ("{}/operations/transportpce-pce:path-computation-request"
- .format(test_utils.RESTCONF_BASE_URL))
- body = {"input": {
+ url = "{}/operations/transportpce-pce:path-computation-request"
+ data = {"input": {
"service-name": "service-1",
"resource-reserve": "true",
"pce-metric": "hop-count",
}
}
}
- response = requests.request(
- "POST", url, data=json.dumps(body), headers=test_utils.TYPE_APPLICATION_JSON,
- auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.post_request(url, data)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Path is calculated',
# Path Computation success
def test_05_path_computation_rdm_bi(self):
- url = ("{}/operations/transportpce-pce:path-computation-request"
- .format(test_utils.RESTCONF_BASE_URL))
- body = {"input": {
+ url = "{}/operations/transportpce-pce:path-computation-request"
+ data = {"input": {
"service-name": "service-1",
"resource-reserve": "true",
"pce-metric": "hop-count",
}
}
}
- response = requests.request(
- "POST", url, data=json.dumps(body), headers=test_utils.TYPE_APPLICATION_JSON,
- auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.post_request(url, data)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Path is calculated',
# Delete topology
def test_06_delete_simple_topology_bi(self):
- url = ("{}/config/ietf-network:networks/network/openroadm-topology"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "DELETE", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/ietf-network:networks/network/openroadm-topology"
+ response = test_utils.delete_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
time.sleep(2)
# Test deleted topology
def test_07_test_topology_simple_bi_deleted(self):
- url = ("{}/config/ietf-network:networks/network/openroadm-topology/node/ROADMA01-SRG1"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/ietf-network:networks/network/openroadm-topology/node/ROADMA01-SRG1"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, 404)
time.sleep(1)
# Load simple bidirectional topology
def test_08_load_simple_topology_uni(self):
- url = ("{}/config/ietf-network:networks/network/openroadm-topology"
- .format(test_utils.RESTCONF_BASE_URL))
- body = self.simple_topo_uni_dir_data
- response = requests.request(
- "PUT", url, data=body, headers=test_utils.TYPE_APPLICATION_XML,
- auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/ietf-network:networks/network/openroadm-topology"
+ data = self.simple_topo_uni_dir_data
+ response = test_utils.put_xmlrequest(url, data)
self.assertEqual(response.status_code, 201)
time.sleep(2)
# Get existing nodeId
def test_09_get_nodeId(self):
- url = ("{}/config/ietf-network:networks/network/openroadm-topology/node/XPONDER-1-2"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/ietf-network:networks/network/openroadm-topology/node/XPONDER-1-2"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual(
# Get existing linkId
def test_10_get_linkId(self):
- url = ("{}/config/ietf-network:networks/network/openroadm-topology/link/XPONDER-1-2XPDR-NW1-TX-toOpenROADM-1-2-SRG1-SRG1-PP1-RX"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/ietf-network:networks/network/openroadm-topology/link/XPONDER-1-2XPDR-NW1-TX-toOpenROADM-1-2-SRG1-SRG1-PP1-RX"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual(
# Path Computation success
def test_11_path_computation_xpdr_uni(self):
- url = ("{}/operations/transportpce-pce:path-computation-request"
- .format(test_utils.RESTCONF_BASE_URL))
- body = {"input": {
+ url = "{}/operations/transportpce-pce:path-computation-request"
+ data = {"input": {
"service-name": "service-1",
"resource-reserve": "true",
"pce-metric": "hop-count",
}
}
}
- response = requests.request(
- "POST", url, data=json.dumps(body), headers=test_utils.TYPE_APPLICATION_JSON,
- auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.post_request(url, data)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Path is calculated',
# Path Computation success
def test_12_path_computation_rdm_uni(self):
- url = ("{}/operations/transportpce-pce:path-computation-request"
- .format(test_utils.RESTCONF_BASE_URL))
- body = {"input": {
+ url = "{}/operations/transportpce-pce:path-computation-request"
+ data = {"input": {
"service-name": "service1",
"resource-reserve": "true",
"service-handler-header": {
"pce-metric": "hop-count"
}
}
- response = requests.request(
- "POST", url, data=json.dumps(body), headers=test_utils.TYPE_APPLICATION_JSON,
- auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.post_request(url, data)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Path is calculated',
# Delete topology
def test_13_delete_simple_topology(self):
- url = ("{}/config/ietf-network:networks/network/openroadm-topology"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "DELETE", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/ietf-network:networks/network/openroadm-topology"
+ response = test_utils.delete_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
time.sleep(2)
# Test deleted topology
def test_14_test_topology_simple_deleted(self):
- url = ("{}/config/ietf-network:networks/network/openroadm-topology/node/XPONDER-1-2"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/ietf-network:networks/network/openroadm-topology/node/XPONDER-1-2"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, 404)
time.sleep(1)
# Load complex topology
def test_15_load_complex_topology(self):
- url = ("{}/config/ietf-network:networks/network/openroadm-topology"
- .format(test_utils.RESTCONF_BASE_URL))
- body = self.complex_topo_uni_dir_data
- response = requests.request(
- "PUT", url, data=body, headers=test_utils.TYPE_APPLICATION_XML,
- auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/ietf-network:networks/network/openroadm-topology"
+ data = self.complex_topo_uni_dir_data
+ response = test_utils.put_xmlrequest(url, data)
self.assertEqual(response.status_code, 201)
time.sleep(2)
# Get existing nodeId
def test_16_get_nodeId(self):
- url = ("{}/config/ietf-network:networks/network/openroadm-topology/node/XPONDER-3-2"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/ietf-network:networks/network/openroadm-topology/node/XPONDER-3-2"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual(
# Test failed path computation
def test_17_fail_path_computation(self):
- url = ("{}/operations/transportpce-pce:path-computation-request"
- .format(test_utils.RESTCONF_BASE_URL))
- body = {"input": {
+ url = "{}/operations/transportpce-pce:path-computation-request"
+ data = {"input": {
"service-handler-header": {
"request-id": "request-1"
}
}
}
- response = requests.request(
- "POST", url, data=json.dumps(body), headers=test_utils.TYPE_APPLICATION_JSON,
- auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.post_request(url, data)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Service Name is not set',
# Test1 success path computation
def test_18_success1_path_computation(self):
- url = ("{}/operations/transportpce-pce:path-computation-request"
- .format(test_utils.RESTCONF_BASE_URL))
- body = {"input": {
+ url = "{}/operations/transportpce-pce:path-computation-request"
+ data = {"input": {
"service-name": "service1",
"resource-reserve": "true",
"service-handler-header": {
"locally-protected-links": "true"
}
}
- response = requests.request(
- "POST", url, data=json.dumps(body), headers=test_utils.TYPE_APPLICATION_JSON,
- auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.post_request(url, data)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Path is calculated',
# Test2 success path computation with path description
def test_19_success2_path_computation(self):
- url = ("{}/operations/transportpce-pce:path-computation-request"
- .format(test_utils.RESTCONF_BASE_URL))
- body = {"input": {
+ url = "{}/operations/transportpce-pce:path-computation-request"
+ data = {"input": {
"service-name": "service 1",
"resource-reserve": "true",
"service-handler-header": {
"pce-metric": "hop-count"
}
}
- response = requests.request(
- "POST", url, data=json.dumps(body), headers=test_utils.TYPE_APPLICATION_JSON,
- auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.post_request(url, data)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Path is calculated',
# Test3 success path computation with hard-constraints exclude
def test_20_success3_path_computation(self):
- url = ("{}/operations/transportpce-pce:path-computation-request"
- .format(test_utils.RESTCONF_BASE_URL))
- body = {"input": {
+ url = "{}/operations/transportpce-pce:path-computation-request"
+ data = {"input": {
"service-name": "service 1",
"resource-reserve": "true",
"service-handler-header": {
"pce-metric": "hop-count"
}
}
- response = requests.request(
- "POST", url, data=json.dumps(body), headers=test_utils.TYPE_APPLICATION_JSON,
- auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.post_request(url, data)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Path is calculated',
# Path computation before deleting oms-attribute of the link :openroadm1-3 to openroadm1-2
def test_21_path_computation_before_oms_attribute_deletion(self):
- url = ("{}/operations/transportpce-pce:path-computation-request"
- .format(test_utils.RESTCONF_BASE_URL))
- body = {"input": {
+ url = "{}/operations/transportpce-pce:path-computation-request"
+ data = {"input": {
"service-name": "service 1",
"resource-reserve": "true",
"service-handler-header": {
"pce-metric": "hop-count"
}
}
- response = requests.request(
- "POST", url, data=json.dumps(body), headers=test_utils.TYPE_APPLICATION_JSON,
- auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.post_request(url, data)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Path is calculated',
def test_22_delete_oms_attribute_in_openroadm13toopenroadm12_link(self):
url = ("{}/config/ietf-network:networks/network/openroadm-topology/ietf-network-topology:link/"
"OpenROADM-1-3-DEG2-to-OpenROADM-1-2-DEG2/org-openroadm-network-topology:OMS-attributes/span"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "DELETE", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ )
+ response = test_utils.delete_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
time.sleep(2)
# Path computation after deleting oms-attribute of the link :openroadm1-3 to openroadm1-2
def test_23_path_computation_after_oms_attribute_deletion(self):
- url = ("{}/operations/transportpce-pce:path-computation-request"
- .format(test_utils.RESTCONF_BASE_URL))
- body = {"input": {
+ url = "{}/operations/transportpce-pce:path-computation-request"
+ data = {"input": {
"service-name": "service 1",
"resource-reserve": "true",
"service-handler-header": {
"pce-metric": "hop-count"
}
}
- response = requests.request(
- "POST", url, data=json.dumps(body), headers=test_utils.TYPE_APPLICATION_JSON,
- auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.post_request(url, data)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Path is calculated',
# Delete complex topology
def test_24_delete_complex_topology(self):
- url = ("{}/config/ietf-network:networks/network/openroadm-topology"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "DELETE", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/ietf-network:networks/network/openroadm-topology"
+ response = test_utils.delete_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
time.sleep(2)
# Test deleted complex topology
def test_25_test_topology_complex_deleted(self):
- url = ("{}/config/ietf-network:networks/network/openroadm-topology/node/XPONDER-3-2"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/ietf-network:networks/network/openroadm-topology/node/XPONDER-3-2"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, 404)
time.sleep(1)
# def test_01_restconfAPI(self):
# url = ("{}/operational/network-topology:network-topology/topology/"
-# "topology-netconf/node/controller-config".format(test_utils.RESTCONF_BASE_URL))
+# "topology-netconf/node/controller-config")
# response = requests.request("GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
# self.assertEqual(response.status_code, requests.codes.ok)
# res = response.json()
# 'connected')
# def test_02_restconfAPI(self):
-# url = ("{}/config/transportpce-portmapping:network/nodes/controller-config".format(test_utils.RESTCONF_BASE_URL))
+# url = ("{}/config/transportpce-portmapping:network/nodes/controller-config")
# response = requests.request(
# "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
# self.assertEqual(response.status_code, requests.codes.not_found)
def test_02_rdm_device_connected(self):
url = ("{}/operational/network-topology:"
"network-topology/topology/topology-netconf/node/ROADMA01"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ )
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual(
def test_03_rdm_portmapping_info(self):
url = ("{}/config/transportpce-portmapping:network/"
"nodes/ROADMA01/node-info"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ )
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual(
def test_04_rdm_portmapping_DEG1_TTP_TXRX(self):
url = ("{}/config/transportpce-portmapping:network/"
"nodes/ROADMA01/mapping/DEG1-TTP-TXRX"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ )
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn(
def test_05_rdm_portmapping_SRG1_PP7_TXRX(self):
url = ("{}/config/transportpce-portmapping:network/"
"nodes/ROADMA01/mapping/SRG1-PP7-TXRX"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ )
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn(
def test_06_rdm_portmapping_SRG3_PP1_TXRX(self):
url = ("{}/config/transportpce-portmapping:network/"
"nodes/ROADMA01/mapping/SRG3-PP1-TXRX"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ )
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn(
def test_08_xpdr_device_connected(self):
url = ("{}/operational/network-topology:"
"network-topology/topology/topology-netconf/node/XPDRA01"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ )
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual(
def test_09_xpdr_portmapping_info(self):
url = ("{}/config/transportpce-portmapping:network/"
"nodes/XPDRA01/node-info"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ )
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual(
def test_10_xpdr_portmapping_NETWORK1(self):
url = ("{}/config/transportpce-portmapping:network/"
"nodes/XPDRA01/mapping/XPDR1-NETWORK1"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ )
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn(
def test_11_xpdr_portmapping_NETWORK2(self):
url = ("{}/config/transportpce-portmapping:network/"
"nodes/XPDRA01/mapping/XPDR1-NETWORK2"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ )
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn(
def test_12_xpdr_portmapping_CLIENT1(self):
url = ("{}/config/transportpce-portmapping:network/"
"nodes/XPDRA01/mapping/XPDR1-CLIENT1"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ )
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn(
def test_13_xpdr_portmapping_CLIENT2(self):
url = ("{}/config/transportpce-portmapping:network/"
"nodes/XPDRA01/mapping/XPDR1-CLIENT2"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ )
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn(
def test_14_xpdr_portmapping_CLIENT3(self):
url = ("{}/config/transportpce-portmapping:network/"
"nodes/XPDRA01/mapping/XPDR1-CLIENT3"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ )
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn(
def test_15_xpdr_portmapping_CLIENT4(self):
url = ("{}/config/transportpce-portmapping:network/"
"nodes/XPDRA01/mapping/XPDR1-CLIENT4"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ )
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn(
def test_17_xpdr_device_disconnected(self):
url = ("{}/operational/network-topology:network-topology/topology/"
- "topology-netconf/node/XPDRA01".format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ "topology-netconf/node/XPDRA01")
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.not_found)
res = response.json()
self.assertIn(
res['errors']['error'])
def test_18_xpdr_device_not_connected(self):
- url = ("{}/config/transportpce-portmapping:network/nodes/XPDRA01".format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = ("{}/config/transportpce-portmapping:network/nodes/XPDRA01")
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.not_found)
res = response.json()
self.assertIn(
def test_20_rdm_device_disconnected(self):
url = ("{}/operational/network-topology:network-topology/topology/topology-netconf/node/ROADMA01"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ )
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.not_found)
res = response.json()
self.assertIn(
res['errors']['error'])
def test_21_rdm_device_not_connected(self):
- url = ("{}/config/transportpce-portmapping:network/nodes/ROADMA01".format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = ("{}/config/transportpce-portmapping:network/nodes/ROADMA01")
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.not_found)
res = response.json()
self.assertIn(
import unittest
import json
import time
-#from unittest.result import failfast
import requests
from common import test_utils
def test_03_rdm_portmapping(self):
url = ("{}/config/transportpce-portmapping:network/"
"nodes/ROADMA01"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ )
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn(
def test_04_xpdr_portmapping(self):
url = ("{}/config/transportpce-portmapping:network/"
"nodes/XPDRA01"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ )
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn(
res['nodes'][0]['mapping'])
def test_05_service_path_create(self):
- url = "{}/operations/transportpce-device-renderer:service-path".format(test_utils.RESTCONF_BASE_URL)
+ url = "{}/operations/transportpce-device-renderer:service-path"
data = {"renderer:input": {
"renderer:service-name": "service_test",
"renderer:wave-number": "7",
{"renderer:node-id": "XPDRA01",
"renderer:src-tp": "XPDR1-CLIENT1",
"renderer:dest-tp": "XPDR1-NETWORK1"}]}}
- response = requests.request(
- "POST", url, data=json.dumps(data),
- headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.post_request(url, data)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Roadm-connection successfully created for nodes: ROADMA01', res["output"]["result"])
url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
"node/ROADMA01/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
"interface/DEG1-TTP-TXRX-7"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ )
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
# the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
"node/ROADMA01/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
"interface/SRG1-PP7-TXRX-7"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ )
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
# the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
"node/ROADMA01/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
"roadm-connections/SRG1-PP7-TXRX-DEG1-TTP-TXRX-7"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ )
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
# the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
"node/XPDRA01/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
"interface/XPDR1-NETWORK1-7"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ )
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
# the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
"node/XPDRA01/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
"interface/XPDR1-NETWORK1-OTU"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ )
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
# the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
"node/XPDRA01/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
"interface/XPDR1-NETWORK1-ODU"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ )
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
# the 2 following statements replace self.assertDictContainsSubset deprecated in python 3.2
url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
"node/XPDRA01/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
"interface/XPDR1-CLIENT1-ETHERNET"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ )
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
# the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
"node/XPDRA01/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
"circuit-packs/1%2F0%2F1-PLUG-NET"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ )
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('not-reserved-inuse', res['circuit-packs'][0]["equipment-state"])
def test_14_service_path_delete(self):
- url = "{}/operations/transportpce-device-renderer:service-path".format(test_utils.RESTCONF_BASE_URL)
+ url = "{}/operations/transportpce-device-renderer:service-path"
data = {"renderer:input": {
"renderer:service-name": "service_test",
"renderer:wave-number": "7",
{"renderer:node-id": "XPDRA01",
"renderer:src-tp": "XPDR1-CLIENT1",
"renderer:dest-tp": "XPDR1-NETWORK1"}]}}
- response = requests.request(
- "POST", url, data=json.dumps(data),
- headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.post_request(url, data)
self.assertEqual(response.status_code, requests.codes.ok)
self.assertEqual(response.json(), {
'output': {'result': 'Request processed', 'success': True}})
url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
"node/ROADMA01/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
"interface/DEG1-TTP-TXRX-7"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ )
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.not_found)
res = response.json()
self.assertIn(
url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
"node/ROADMA01/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
"interface/SRG1-PP7-TXRX-7"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ )
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.not_found)
res = response.json()
self.assertIn(
url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
"node/ROADMA01/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
"roadm-connections/SRG1-PP7-TXRX-DEG1-TTP-TXRX-7"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ )
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.not_found)
res = response.json()
self.assertIn(
url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
"node/XPDRA01/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
"interface/XPDR1-NETWORK1-7"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ )
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.not_found)
res = response.json()
self.assertIn(
url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
"node/XPDRA01/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
"interface/XPDR1-NETWORK1-OTU"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ )
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.not_found)
res = response.json()
self.assertIn(
url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
"node/XPDRA01/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
"interface/XPDR1-NETWORK1-ODU"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ )
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.not_found)
res = response.json()
self.assertIn(
url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
"node/XPDRA01/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
"interface/XPDR1-CLIENT1-ETHERNET"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ )
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.not_found)
res = response.json()
self.assertIn(
url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
"node/XPDRA01/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
"circuit-packs/1%2F0%2F1-PLUG-NET"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ )
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual('not-reserved-available', res["circuit-packs"][0]['equipment-state'])
# Verify the termination points of the ROADMA
def test_02_compareOpenroadmTopologyPortMapping_rdm(self):
- urlTopo = ("{}/config/ietf-network:networks/network/openroadm-topology"
- .format(test_utils.RESTCONF_BASE_URL))
- responseTopo = requests.request("GET", urlTopo, headers=test_utils.TYPE_APPLICATION_JSON,
- auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ urlTopo = "{}/config/ietf-network:networks/network/openroadm-topology"
+ responseTopo = test_utils.get_request(urlTopo)
resTopo = responseTopo.json()
nbNode = len(resTopo['network'][0]['node'])
nbMapCumul = 0
nodeId = resTopo['network'][0]['node'][i]['node-id']
nodeMapId = nodeId.split("-")[0]
urlMapList = "{}/config/transportpce-portmapping:network/nodes/" + nodeMapId
- urlMapListFull = urlMapList.format(test_utils.RESTCONF_BASE_URL)
- responseMapList = requests.request("GET", urlMapListFull, headers=test_utils.TYPE_APPLICATION_JSON,
- auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ responseMapList = test_utils.get_request(urlMapList)
resMapList = responseMapList.json()
-
nbMappings = len(resMapList['nodes'][0]['mapping']) - nbMapCumul
nbTp = len(resTopo['network'][0]['node'][i]['ietf-network-topology:termination-point'])
nbMapCurrent = 0
tpId = resTopo['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]['tp-id']
if((not "CP" in tpId) and (not "CTP" in tpId)):
urlMap = "{}/config/transportpce-portmapping:network/nodes/" + nodeMapId + "/mapping/" + tpId
- urlMapFull = urlMap.format(test_utils.RESTCONF_BASE_URL)
- responseMap = requests.request("GET", urlMapFull, headers=test_utils.TYPE_APPLICATION_JSON,
- auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
- self.assertEqual(responseMap.status_code, requests.codes.ok)
- if responseMap.status_code == requests.codes.ok:
- nbMapCurrent += 1
- nbMapCumul += nbMapCurrent
- nbMappings -= nbMapCurrent
- self.assertEqual(nbMappings, 0)
+ responseMap = test_utils.get_request(urlMap)
# Disconnect the ROADMA
def test_03_disconnect_rdm(self):
def test_02_getClliNetwork(self):
url = ("{}/config/ietf-network:networks/network/clli-network"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ )
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual(res['network'][0]['node'][0]['node-id'], 'NodeA')
def test_03_getOpenRoadmNetwork(self):
url = ("{}/config/ietf-network:networks/network/openroadm-network"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ )
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual(res['network'][0]['node'][0]['node-id'], 'ROADMA01')
self.assertEqual(res['network'][0]['node'][0]['org-openroadm-network:model'], '2')
def test_04_getLinks_OpenroadmTopology(self):
- url = ("{}/config/ietf-network:networks/network/openroadm-topology"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/ietf-network:networks/network/openroadm-topology"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
# Tests related to links
self.assertEqual(len(dropLink), 0)
def test_05_getNodes_OpenRoadmTopology(self):
- url = ("{}/config/ietf-network:networks/network/openroadm-topology"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request("GET", url, headers=test_utils.TYPE_APPLICATION_JSON,
- auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/ietf-network:networks/network/openroadm-topology"
+ response = test_utils.get_request(url)
res = response.json()
# Tests related to nodes
self.assertEqual(response.status_code, requests.codes.ok)
self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
def test_07_getClliNetwork(self):
- url = ("{}/config/ietf-network:networks/network/clli-network"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request("GET", url, headers=test_utils.TYPE_APPLICATION_JSON,
- auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/ietf-network:networks/network/clli-network"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual(res['network'][0]['node'][0]['node-id'], 'NodeA')
self.assertEqual(res['network'][0]['node'][0]['org-openroadm-clli-network:clli'], 'NodeA')
def test_08_getOpenRoadmNetwork(self):
- url = ("{}/config/ietf-network:networks/network/openroadm-network"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/ietf-network:networks/network/openroadm-network"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
nbNode = len(res['network'][0]['node'])
self.assertFalse(True)
def test_09_getNodes_OpenRoadmTopology(self):
- url = ("{}/config/ietf-network:networks/network/openroadm-topology"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/ietf-network:networks/network/openroadm-topology"
+ response = test_utils.get_request(url)
res = response.json()
# Tests related to nodes
self.assertEqual(response.status_code, requests.codes.ok)
# Connect the tail XPDRA to ROADMA and vice versa
def test_10_connect_tail_xpdr_rdm(self):
# Connect the tail: XPDRA to ROADMA
- url = ("{}/operations/transportpce-networkutils:init-xpdr-rdm-links"
- .format(test_utils.RESTCONF_BASE_URL))
- data = {"networkutils:input": {
- "networkutils:links-input": {
- "networkutils:xpdr-node": "XPDRA01",
- "networkutils:xpdr-num": "1",
- "networkutils:network-num": "1",
- "networkutils:rdm-node": "ROADMA01",
- "networkutils:srg-num": "1",
- "networkutils:termination-point-num": "SRG1-PP1-TXRX"
- }
- }
- }
- response = requests.request(
- "POST", url, data=json.dumps(data), headers=test_utils.TYPE_APPLICATION_JSON,
- auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.connect_xpdr_to_rdm_request("XPDRA01", "1", "1",
+ "ROADMA01", "1", "SRG1-PP1-TXRX")
self.assertEqual(response.status_code, requests.codes.ok)
def test_11_connect_tail_rdm_xpdr(self):
# Connect the tail: ROADMA to XPDRA
- url = ("{}/operations/transportpce-networkutils:init-rdm-xpdr-links"
- .format(test_utils.RESTCONF_BASE_URL))
- data = {"networkutils:input": {
- "networkutils:links-input": {
- "networkutils:xpdr-node": "XPDRA01",
- "networkutils:xpdr-num": "1",
- "networkutils:network-num": "1",
- "networkutils:rdm-node": "ROADMA01",
- "networkutils:srg-num": "1",
- "networkutils:termination-point-num": "SRG1-PP1-TXRX"
- }
- }
- }
- response = requests.request(
- "POST", url, data=json.dumps(data), headers=test_utils.TYPE_APPLICATION_JSON,
- auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.connect_rdm_to_xpdr_request("XPDRA01", "1", "1",
+ "ROADMA01", "1", "SRG1-PP1-TXRX")
self.assertEqual(response.status_code, requests.codes.ok)
def test_12_getLinks_OpenRoadmTopology(self):
- url = ("{}/config/ietf-network:networks/network/openroadm-topology"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/ietf-network:networks/network/openroadm-topology"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
# Tests related to links
url = ("{}/config/ietf-network:networks/network/openroadm-topology/ietf-network-topology:"
"link/ROADMA01-DEG1-DEG1-TTP-TXRXtoROADMC01-DEG2-DEG2-TTP-TXRX/org-openroadm-network-topology:"
"OMS-attributes/span"
- .format(test_utils.RESTCONF_BASE_URL))
+ )
data = {"span": {
"auto-spanloss": "true",
"engineered-spanloss": 12.2,
"fiber-type": "smf",
"SRLG-length": 100000,
"pmd": 0.5}]}}
- response = requests.request(
- "PUT", url, data=json.dumps(data), headers=test_utils.TYPE_APPLICATION_JSON,
- auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.put_request(url, data)
self.assertEqual(response.status_code, requests.codes.created)
def test_15_omsAttributes_ROADMC_ROADMA(self):
url = ("{}/config/ietf-network:networks/network/openroadm-topology/ietf-network-topology:"
"link/ROADMC01-DEG2-DEG2-TTP-TXRXtoROADMA01-DEG1-DEG1-TTP-TXRX/org-openroadm-network-topology:"
"OMS-attributes/span"
- .format(test_utils.RESTCONF_BASE_URL))
+ )
data = {"span": {
"auto-spanloss": "true",
"engineered-spanloss": 12.2,
"fiber-type": "smf",
"SRLG-length": 100000,
"pmd": 0.5}]}}
- response = requests.request(
- "PUT", url, data=json.dumps(data), headers=test_utils.TYPE_APPLICATION_JSON,
- auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.put_request(url, data)
self.assertEqual(response.status_code, requests.codes.created)
def test_16_getClliNetwork(self):
- url = ("{}/config/ietf-network:networks/network/clli-network"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/ietf-network:networks/network/clli-network"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
nbNode = len(res['network'][0]['node'])
def test_17_getOpenRoadmNetwork(self):
url = ("{}/config/ietf-network:networks/network/openroadm-network"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ )
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
nbNode = len(res['network'][0]['node'])
self.assertEqual(len(listNode), 0)
def test_18_getROADMLinkOpenRoadmTopology(self):
- url = ("{}/config/ietf-network:networks/network/openroadm-topology"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/ietf-network:networks/network/openroadm-topology"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
# Tests related to links
self.assertEqual(len(XPDR_OUT), 0)
def test_19_getLinkOmsAttributesOpenRoadmTopology(self):
- url = ("{}/config/ietf-network:networks/network/openroadm-topology"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/ietf-network:networks/network/openroadm-topology"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
# Tests related to links
self.assertEqual(len(R2RLink), 0)
def test_20_getNodes_OpenRoadmTopology(self):
- url = ("{}/config/ietf-network:networks/network/openroadm-topology"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/ietf-network:networks/network/openroadm-topology"
+ response = test_utils.get_request(url)
res = response.json()
# Tests related to nodes
self.assertEqual(response.status_code, requests.codes.ok)
url = ("{}/config/ietf-network:networks/network/openroadm-topology/ietf-network-topology:"
"link/ROADMA01-DEG2-DEG2-TTP-TXRXtoROADMB01-DEG1-DEG1-TTP-TXRX/org-openroadm-network-topology:"
"OMS-attributes/span"
- .format(test_utils.RESTCONF_BASE_URL))
+ )
data = {"span": {
"auto-spanloss": "true",
"engineered-spanloss": 12.2,
"fiber-type": "smf",
"SRLG-length": 100000,
"pmd": 0.5}]}}
- response = requests.request(
- "PUT", url, data=json.dumps(data), headers=test_utils.TYPE_APPLICATION_JSON,
- auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.put_request(url, data)
self.assertEqual(response.status_code, requests.codes.created)
def test_23_omsAttributes_ROADMB_ROADMA(self):
url = ("{}/config/ietf-network:networks/network/openroadm-topology/ietf-network-topology:"
"link/ROADMB01-DEG1-DEG1-TTP-TXRXtoROADMA01-DEG2-DEG2-TTP-TXRX/org-openroadm-network-topology:"
"OMS-attributes/span"
- .format(test_utils.RESTCONF_BASE_URL))
+ )
data = {"span": {
"auto-spanloss": "true",
"engineered-spanloss": 12.2,
"fiber-type": "smf",
"SRLG-length": 100000,
"pmd": 0.5}]}}
- response = requests.request(
- "PUT", url, data=json.dumps(data), headers=test_utils.TYPE_APPLICATION_JSON,
- auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.put_request(url, data)
self.assertEqual(response.status_code, requests.codes.created)
def test_24_omsAttributes_ROADMB_ROADMC(self):
url = ("{}/config/ietf-network:networks/network/openroadm-topology/ietf-network-topology:"
"link/ROADMB01-DEG2-DEG2-TTP-TXRXtoROADMC01-DEG1-DEG1-TTP-TXRX/org-openroadm-network-topology:"
"OMS-attributes/span"
- .format(test_utils.RESTCONF_BASE_URL))
+ )
data = {"span": {
"auto-spanloss": "true",
"engineered-spanloss": 12.2,
"fiber-type": "smf",
"SRLG-length": 100000,
"pmd": 0.5}]}}
- response = requests.request(
- "PUT", url, data=json.dumps(data), headers=test_utils.TYPE_APPLICATION_JSON,
- auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.put_request(url, data)
self.assertEqual(response.status_code, requests.codes.created)
def test_25_omsAttributes_ROADMC_ROADMB(self):
url = ("{}/config/ietf-network:networks/network/openroadm-topology/ietf-network-topology:"
"link/ROADMC01-DEG1-DEG1-TTP-TXRXtoROADMB01-DEG2-DEG2-TTP-TXRX/org-openroadm-network-topology:"
"OMS-attributes/span"
- .format(test_utils.RESTCONF_BASE_URL))
+ )
data = {"span": {
"auto-spanloss": "true",
"engineered-spanloss": 12.2,
"fiber-type": "smf",
"SRLG-length": 100000,
"pmd": 0.5}]}}
- response = requests.request(
- "PUT", url, data=json.dumps(data), headers=test_utils.TYPE_APPLICATION_JSON,
- auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.put_request(url, data)
self.assertEqual(response.status_code, requests.codes.created)
def test_26_getClliNetwork(self):
- url = ("{}/config/ietf-network:networks/network/clli-network"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/ietf-network:networks/network/clli-network"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
nbNode = len(res['network'][0]['node'])
self.assertEqual(len(listNode), 0)
def test_27_verifyDegree(self):
- url = ("{}/config/ietf-network:networks/network/openroadm-topology"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/ietf-network:networks/network/openroadm-topology"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
# Tests related to links
self.assertEqual(len(listR2RLink), 0)
def test_28_verifyOppositeLinkTopology(self):
- url = ("{}/config/ietf-network:networks/network/openroadm-topology"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/ietf-network:networks/network/openroadm-topology"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
# Tests related to links
oppLink_id = res['network'][0]['ietf-network-topology:link'][i]['org-openroadm-common-network:opposite-link']
# Find the opposite link
url_oppLink = "{}/config/ietf-network:networks/network/openroadm-topology/ietf-network-topology:link/"+oppLink_id
- url = (url_oppLink.format(test_utils.RESTCONF_BASE_URL))
- response_oppLink = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response_oppLink = test_utils.get_request(url_oppLink)
self.assertEqual(response_oppLink.status_code, requests.codes.ok)
res_oppLink = response_oppLink.json()
self.assertEqual(res_oppLink['ietf-network-topology:link'][0]
def test_29_getLinkOmsAttributesOpenRoadmTopology(self):
url = ("{}/config/ietf-network:networks/network/openroadm-topology"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ )
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
nbLink = len(res['network'][0]['ietf-network-topology:link'])
response = test_utils.unmount_device("ROADMB01")
self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
# Delete in the clli-network
- url = ("{}/config/ietf-network:networks/network/clli-network/node/NodeB"
- .format(test_utils.RESTCONF_BASE_URL))
+ url = "{}/config/ietf-network:networks/network/clli-network/node/NodeB"
data = {}
- response = requests.request(
- "DELETE", url, data=json.dumps(data), headers=test_utils.TYPE_APPLICATION_JSON,
- auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.delete_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
def test_31_disconnect_ROADMC(self):
response = test_utils.unmount_device("ROADMC01")
self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
# Delete in the clli-network
- url = ("{}/config/ietf-network:networks/network/clli-network/node/NodeC"
- .format(test_utils.RESTCONF_BASE_URL))
+ url = "{}/config/ietf-network:networks/network/clli-network/node/NodeC"
data = {}
- response = requests.request(
- "DELETE", url, data=json.dumps(data), headers=test_utils.TYPE_APPLICATION_JSON,
- auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.delete_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
def test_32_getNodes_OpenRoadmTopology(self):
- url = ("{}/config/ietf-network:networks/network/openroadm-topology"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/ietf-network:networks/network/openroadm-topology"
+ response = test_utils.get_request(url)
res = response.json()
# Tests related to nodes
self.assertEqual(response.status_code, requests.codes.ok)
self.assertNotEqual(res['network'][0]['node'][i]['node-id'], 'ROADMC01-DEG2')
def test_33_getOpenRoadmNetwork(self):
- url = ("{}/config/ietf-network:networks/network/openroadm-network"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/ietf-network:networks/network/openroadm-network"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
nbNode = len(res['network'][0]['node'])
self.assertNotEqual(res['network'][0]['node'][i]['node-id'], 'ROADMB01')
def test_34_getClliNetwork(self):
- url = ("{}/config/ietf-network:networks/network/clli-network"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/ietf-network:networks/network/clli-network"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
nbNode = len(res['network'][0]['node'])
def test_36_getClliNetwork(self):
url = ("{}/config/ietf-network:networks/network/clli-network"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ )
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
nbNode = len(res['network'][0]['node'])
def test_37_getOpenRoadmNetwork(self):
url = ("{}/config/ietf-network:networks/network/openroadm-network"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ )
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
nbNode = len(res['network'][0]['node'])
self.assertNotEqual(res['network'][0]['node'][i]['node-id'], 'XPDRA01')
def test_38_getNodes_OpenRoadmTopology(self):
- url = ("{}/config/ietf-network:networks/network/openroadm-topology"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/ietf-network:networks/network/openroadm-topology"
+ response = test_utils.get_request(url)
res = response.json()
# Tests related to nodes
self.assertEqual(response.status_code, requests.codes.ok)
# Link-1
url = ("{}/config/ietf-network:networks/network/openroadm-topology/ietf-network-topology:"
"link/XPDRA01-XPDR1-XPDR1-NETWORK1toROADMA01-SRG1-SRG1-PP1-TXRX"
- .format(test_utils.RESTCONF_BASE_URL))
+ )
data = {}
- response = requests.request(
- "DELETE", url, data=json.dumps(data), headers=test_utils.TYPE_APPLICATION_JSON,
- auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.delete_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
# Link-2
url = ("{}/config/ietf-network:networks/network/openroadm-topology/ietf-network-topology:"
"link/ROADMA01-SRG1-SRG1-PP1-TXRXtoXPDRA01-XPDR1-XPDR1-NETWORK1"
- .format(test_utils.RESTCONF_BASE_URL))
+ )
data = {}
- response = requests.request(
- "DELETE", url, data=json.dumps(data), headers=test_utils.TYPE_APPLICATION_JSON,
- auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.delete_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
def test_40_getLinks_OpenRoadmTopology(self):
- url = ("{}/config/ietf-network:networks/network/openroadm-topology"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/ietf-network:networks/network/openroadm-topology"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
nbLink = len(res['network'][0]['ietf-network-topology:link'])
response = test_utils.unmount_device("ROADMA01")
self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
# Delete in the clli-network
- url = ("{}/config/ietf-network:networks/network/clli-network/node/NodeA"
- .format(test_utils.RESTCONF_BASE_URL))
+ url = "{}/config/ietf-network:networks/network/clli-network/node/NodeA"
data = {}
- response = requests.request(
- "DELETE", url, data=json.dumps(data), headers=test_utils.TYPE_APPLICATION_JSON,
- auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.delete_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
def test_42_getClliNetwork(self):
- url = ("{}/config/ietf-network:networks/network/clli-network"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/ietf-network:networks/network/clli-network"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertNotIn('node', res['network'][0])
def test_43_getOpenRoadmNetwork(self):
- url = ("{}/config/ietf-network:networks/network/openroadm-network"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/ietf-network:networks/network/openroadm-network"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertNotIn('node', res['network'][0])
def test_44_check_roadm2roadm_link_persistence(self):
- url = ("{}/config/ietf-network:networks/network/openroadm-topology"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/ietf-network:networks/network/openroadm-topology"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
nbLink = len(res['network'][0]['ietf-network-topology:link'])
self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
def test_05_connect_xprdA_N1_to_roadmA_PP1(self):
- url = "{}/operations/transportpce-networkutils:init-xpdr-rdm-links".format(test_utils.RESTCONF_BASE_URL)
- 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"
- }
- }
- }
- response = requests.request(
- "POST", url, data=json.dumps(data),
- headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.connect_xpdr_to_rdm_request("XPDR-A1", "1", "1",
+ "ROADM-A1", "1", "SRG1-PP1-TXRX")
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(test_utils.RESTCONF_BASE_URL)
- 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"
- }
- }
- }
- response = requests.request(
- "POST", url, data=json.dumps(data),
- headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.connect_rdm_to_xpdr_request("XPDR-A1", "1", "1",
+ "ROADM-A1", "1", "SRG1-PP1-TXRX")
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(test_utils.RESTCONF_BASE_URL)
- 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"
- }
- }
- }
- response = requests.request(
- "POST", url, data=json.dumps(data),
- headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.connect_xpdr_to_rdm_request("XPDR-C1", "1", "1",
+ "ROADM-C1", "1", "SRG1-PP1-TXRX")
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(test_utils.RESTCONF_BASE_URL)
- 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"
- }
- }
- }
- response = requests.request(
- "POST", url, data=json.dumps(data),
- headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.connect_rdm_to_xpdr_request("XPDR-C1", "1", "1",
+ "ROADM-C1", "1", "SRG1-PP1-TXRX")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Roadm Xponder links created successfully', res["output"]["result"])
# 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(test_utils.RESTCONF_BASE_URL))
+ "OMS-attributes/span")
data = {"span": {
"auto-spanloss": "true",
"spanloss-base": 11.4,
"fiber-type": "smf",
"SRLG-length": 100000,
"pmd": 0.5}]}}
- response = requests.request(
- "PUT", url, data=json.dumps(data), headers=test_utils.TYPE_APPLICATION_JSON,
- auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.put_request(url, data)
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(test_utils.RESTCONF_BASE_URL))
+ "OMS-attributes/span")
data = {"span": {
"auto-spanloss": "true",
"spanloss-base": 11.4,
"fiber-type": "smf",
"SRLG-length": 100000,
"pmd": 0.5}]}}
- response = requests.request(
- "PUT", url, data=json.dumps(data), headers=test_utils.TYPE_APPLICATION_JSON,
- auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.put_request(url, data)
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(test_utils.RESTCONF_BASE_URL))
+ url = "{}/operations/org-openroadm-service:service-create"
data = {"input": {
"sdnc-request-header": {
"request-id": "e3028bae-a90f-4ddd-a83f-cf224eba0e58",
"operator-contact": "pw1234"
}
}
- response = requests.request(
- "POST", url, data=json.dumps(data), headers=test_utils.TYPE_APPLICATION_JSON,
- auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.post_request(url, data)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('PCE calculation in progress',
time.sleep(self.WAITING)
def test_12_get_eth_service1(self):
- url = ("{}/operational/org-openroadm-service:service-list/services/service1"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/operational/org-openroadm-service:service-list/services/service1"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual(
def test_13_check_xc1_ROADMA(self):
url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
"node/ROADM-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
- "roadm-connections/SRG1-PP1-TXRX-DEG2-TTP-TXRX-1"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ "roadm-connections/SRG1-PP1-TXRX-DEG2-TTP-TXRX-1")
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
# the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
def test_14_check_xc1_ROADMC(self):
url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
"node/ROADM-C1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
- "roadm-connections/SRG1-PP1-TXRX-DEG1-TTP-TXRX-1"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ "roadm-connections/SRG1-PP1-TXRX-DEG1-TTP-TXRX-1")
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
# the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
time.sleep(5)
def test_15_check_topo_XPDRA(self):
- url1 = ("{}/config/ietf-network:networks/network/openroadm-topology/node/XPDR-A1-XPDR1"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url1, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/ietf-network:networks/network/openroadm-topology/node/XPDR-A1-XPDR1"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
liste_tp = res['node'][0]['ietf-network-topology:termination-point']
time.sleep(3)
def test_16_check_topo_ROADMA_SRG1(self):
- url1 = ("{}/config/ietf-network:networks/network/openroadm-topology/node/ROADM-A1-SRG1"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url1, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/ietf-network:networks/network/openroadm-topology/node/ROADM-A1-SRG1"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertNotIn({u'index': 1},
time.sleep(3)
def test_17_check_topo_ROADMA_DEG1(self):
- url1 = ("{}/config/ietf-network:networks/network/openroadm-topology/node/ROADM-A1-DEG2"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url1, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/ietf-network:networks/network/openroadm-topology/node/ROADM-A1-DEG2"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertNotIn({u'index': 1},
time.sleep(3)
def test_18_connect_xprdA_N2_to_roadmA_PP2(self):
- url = "{}/operations/transportpce-networkutils:init-xpdr-rdm-links".format(test_utils.RESTCONF_BASE_URL)
- data = {
- "networkutils:input": {
- "networkutils:links-input": {
- "networkutils:xpdr-node": "XPDR-A1",
- "networkutils:xpdr-num": "1",
- "networkutils:network-num": "2",
- "networkutils:rdm-node": "ROADM-A1",
- "networkutils:srg-num": "1",
- "networkutils:termination-point-num": "SRG1-PP2-TXRX"
- }
- }
- }
- response = requests.request(
- "POST", url, data=json.dumps(data),
- headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.connect_xpdr_to_rdm_request("XPDR-A1", "1", "2",
+ "ROADM-A1", "1", "SRG1-PP2-TXRX")
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_19_connect_roadmA_PP2_to_xpdrA_N2(self):
- url = "{}/operations/transportpce-networkutils:init-rdm-xpdr-links".format(test_utils.RESTCONF_BASE_URL)
- data = {
- "networkutils:input": {
- "networkutils:links-input": {
- "networkutils:xpdr-node": "XPDR-A1",
- "networkutils:xpdr-num": "1",
- "networkutils:network-num": "2",
- "networkutils:rdm-node": "ROADM-A1",
- "networkutils:srg-num": "1",
- "networkutils:termination-point-num": "SRG1-PP2-TXRX"
- }
- }
- }
- response = requests.request(
- "POST", url, data=json.dumps(data),
- headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.connect_rdm_to_xpdr_request("XPDR-A1", "1", "2",
+ "ROADM-A1", "1", "SRG1-PP2-TXRX")
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_20_connect_xprdC_N2_to_roadmC_PP2(self):
- url = "{}/operations/transportpce-networkutils:init-xpdr-rdm-links".format(test_utils.RESTCONF_BASE_URL)
- data = {
- "networkutils:input": {
- "networkutils:links-input": {
- "networkutils:xpdr-node": "XPDR-C1",
- "networkutils:xpdr-num": "1",
- "networkutils:network-num": "2",
- "networkutils:rdm-node": "ROADM-C1",
- "networkutils:srg-num": "1",
- "networkutils:termination-point-num": "SRG1-PP2-TXRX"
- }
- }
- }
- response = requests.request(
- "POST", url, data=json.dumps(data),
- headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.connect_xpdr_to_rdm_request("XPDR-C1", "1", "2",
+ "ROADM-C1", "1", "SRG1-PP2-TXRX")
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_21_connect_roadmC_PP2_to_xpdrC_N2(self):
- url = "{}/operations/transportpce-networkutils:init-rdm-xpdr-links".format(test_utils.RESTCONF_BASE_URL)
- data = {
- "networkutils:input": {
- "networkutils:links-input": {
- "networkutils:xpdr-node": "XPDR-C1",
- "networkutils:xpdr-num": "1",
- "networkutils:network-num": "2",
- "networkutils:rdm-node": "ROADM-C1",
- "networkutils:srg-num": "1",
- "networkutils:termination-point-num": "SRG1-PP2-TXRX"
- }
- }
- }
- response = requests.request(
- "POST", url, data=json.dumps(data),
- headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.connect_rdm_to_xpdr_request("XPDR-C1", "1", "2",
+ "ROADM-C1", "1", "SRG1-PP2-TXRX")
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_22_create_eth_service2(self):
- url = ("{}/operations/org-openroadm-service:service-create"
- .format(test_utils.RESTCONF_BASE_URL))
+ url = "{}/operations/org-openroadm-service:service-create"
data = {"input": {
"sdnc-request-header": {
"request-id": "e3028bae-a90f-4ddd-a83f-cf224eba0e58",
"operator-contact": "pw1234"
}
}
- response = requests.request(
- "POST", url, data=json.dumps(data), headers=test_utils.TYPE_APPLICATION_JSON,
- auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.post_request(url, data)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('PCE calculation in progress',
time.sleep(self.WAITING)
def test_23_get_eth_service2(self):
- url = ("{}/operational/org-openroadm-service:service-list/services/service2"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/operational/org-openroadm-service:service-list/services/service2"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual(
def test_24_check_xc2_ROADMA(self):
url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
"node/ROADM-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
- "roadm-connections/DEG2-TTP-TXRX-SRG1-PP2-TXRX-2"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ "roadm-connections/DEG2-TTP-TXRX-SRG1-PP2-TXRX-2")
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
# the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
res['roadm-connections'][0]['destination'])
def test_25_check_topo_XPDRA(self):
- url1 = ("{}/config/ietf-network:networks/network/openroadm-topology/node/XPDR-A1-XPDR1"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url1, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/ietf-network:networks/network/openroadm-topology/node/XPDR-A1-XPDR1"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
liste_tp = res['node'][0]['ietf-network-topology:termination-point']
time.sleep(10)
def test_26_check_topo_ROADMA_SRG1(self):
- url1 = ("{}/config/ietf-network:networks/network/openroadm-topology/node/ROADM-A1-SRG1"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url1, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/ietf-network:networks/network/openroadm-topology/node/ROADM-A1-SRG1"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertNotIn({u'index': 1}, res['node'][0]
time.sleep(10)
def test_27_check_topo_ROADMA_DEG2(self):
- url1 = ("{}/config/ietf-network:networks/network/openroadm-topology/node/ROADM-A1-DEG2"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url1, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/ietf-network:networks/network/openroadm-topology/node/ROADM-A1-DEG2"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertNotIn({u'index': 1}, res['node'][0]
# creation service test on a non-available resource
def test_28_create_eth_service3(self):
- url = ("{}/operations/org-openroadm-service:service-create"
- .format(test_utils.RESTCONF_BASE_URL))
+ url = "{}/operations/org-openroadm-service:service-create"
data = {"input": {
"sdnc-request-header": {
"request-id": "e3028bae-a90f-4ddd-a83f-cf224eba0e58",
"operator-contact": "pw1234"
}
}
- response = requests.request(
- "POST", url, data=json.dumps(data), headers=test_utils.TYPE_APPLICATION_JSON,
- auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.post_request(url, data)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('PCE calculation in progress',
# add a test that check the openroadm-service-list still only contains 2 elements
def test_29_delete_eth_service3(self):
- url = ("{}/operations/org-openroadm-service:service-delete"
- .format(test_utils.RESTCONF_BASE_URL))
+ url = "{}/operations/org-openroadm-service:service-delete"
data = {"input": {
"sdnc-request-header": {
"request-id": "e3028bae-a90f-4ddd-a83f-cf224eba0e58",
}
}
}
- response = requests.request(
- "POST", url, data=json.dumps(data), headers=test_utils.TYPE_APPLICATION_JSON,
- auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.post_request(url, data)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Service \'service3\' does not exist in datastore',
time.sleep(20)
def test_30_delete_eth_service1(self):
- url = ("{}/operations/org-openroadm-service:service-delete"
- .format(test_utils.RESTCONF_BASE_URL))
+ url = "{}/operations/org-openroadm-service:service-delete"
data = {"input": {
"sdnc-request-header": {
"request-id": "e3028bae-a90f-4ddd-a83f-cf224eba0e58",
}
}
}
- response = requests.request(
- "POST", url, data=json.dumps(data), headers=test_utils.TYPE_APPLICATION_JSON,
- auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.post_request(url, data)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Renderer service delete in progress',
time.sleep(20)
def test_31_delete_eth_service2(self):
- url = ("{}/operations/org-openroadm-service:service-delete"
- .format(test_utils.RESTCONF_BASE_URL))
+ url = "{}/operations/org-openroadm-service:service-delete"
data = {"input": {
"sdnc-request-header": {
"request-id": "e3028bae-a90f-4ddd-a83f-cf224eba0e58",
}
}
}
- response = requests.request(
- "POST", url, data=json.dumps(data), headers=test_utils.TYPE_APPLICATION_JSON,
- auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.post_request(url, data)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Renderer service delete in progress',
def test_32_check_no_xc_ROADMA(self):
url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
- "node/ROADM-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ "node/ROADM-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/")
+ response = test_utils.get_request(url)
res = response.json()
self.assertEqual(response.status_code, requests.codes.ok)
self.assertNotIn('roadm-connections', dict.keys(res['org-openroadm-device']))
time.sleep(2)
def test_33_check_topo_XPDRA(self):
- url1 = ("{}/config/ietf-network:networks/network/openroadm-topology/node/XPDR-A1-XPDR1"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url1, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/ietf-network:networks/network/openroadm-topology/node/XPDR-A1-XPDR1"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
liste_tp = res['node'][0]['ietf-network-topology:termination-point']
time.sleep(10)
def test_34_check_topo_ROADMA_SRG1(self):
- url1 = ("{}/config/ietf-network:networks/network/openroadm-topology/node/ROADM-A1-SRG1"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url1, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/ietf-network:networks/network/openroadm-topology/node/ROADM-A1-SRG1"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn({u'index': 1}, res['node'][0]
time.sleep(10)
def test_35_check_topo_ROADMA_DEG2(self):
- url1 = ("{}/config/ietf-network:networks/network/openroadm-topology/node/ROADM-A1-DEG2"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url1, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/ietf-network:networks/network/openroadm-topology/node/ROADM-A1-DEG2"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn({u'index': 1}, res['node'][0]
# test service-create for Optical Channel (OC) service from srg-pp to srg-pp
def test_36_create_oc_service1(self):
- url = ("{}/operations/org-openroadm-service:service-create"
- .format(test_utils.RESTCONF_BASE_URL))
+ url = "{}/operations/org-openroadm-service:service-create"
data = {"input": {
"sdnc-request-header": {
"request-id": "e3028bae-a90f-4ddd-a83f-cf224eba0e58",
"operator-contact": "pw1234"
}
}
- response = requests.request(
- "POST", url, data=json.dumps(data), headers=test_utils.TYPE_APPLICATION_JSON,
- auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.post_request(url, data)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('PCE calculation in progress',
time.sleep(self.WAITING)
def test_37_get_oc_service1(self):
- url = ("{}/operational/org-openroadm-service:service-list/services/service1"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/operational/org-openroadm-service:service-list/services/service1"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual(
def test_38_check_xc1_ROADMA(self):
url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
"node/ROADM-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
- "roadm-connections/SRG1-PP1-TXRX-DEG2-TTP-TXRX-1"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ "roadm-connections/SRG1-PP1-TXRX-DEG2-TTP-TXRX-1")
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
# the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
def test_39_check_xc1_ROADMC(self):
url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
"node/ROADM-C1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
- "roadm-connections/SRG1-PP1-TXRX-DEG1-TTP-TXRX-1"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ "roadm-connections/SRG1-PP1-TXRX-DEG1-TTP-TXRX-1")
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
# the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
time.sleep(7)
def test_40_create_oc_service2(self):
- url = ("{}/operations/org-openroadm-service:service-create"
- .format(test_utils.RESTCONF_BASE_URL))
+ url = "{}/operations/org-openroadm-service:service-create"
data = {"input": {
"sdnc-request-header": {
"request-id": "e3028bae-a90f-4ddd-a83f-cf224eba0e58",
"operator-contact": "pw1234"
}
}
- response = requests.request(
- "POST", url, data=json.dumps(data), headers=test_utils.TYPE_APPLICATION_JSON,
- auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.post_request(url, data)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('PCE calculation in progress',
time.sleep(self.WAITING)
def test_41_get_oc_service2(self):
- url = ("{}/operational/org-openroadm-service:service-list/services/service2"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/operational/org-openroadm-service:service-list/services/service2"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual(
def test_42_check_xc2_ROADMA(self):
url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
"node/ROADM-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
- "roadm-connections/SRG1-PP2-TXRX-DEG2-TTP-TXRX-2"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ "roadm-connections/SRG1-PP2-TXRX-DEG2-TTP-TXRX-2")
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
# the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
time.sleep(3)
def test_44_delete_oc_service1(self):
- url = ("{}/operations/org-openroadm-service:service-delete"
- .format(test_utils.RESTCONF_BASE_URL))
+ url = "{}/operations/org-openroadm-service:service-delete"
data = {"input": {
"sdnc-request-header": {
"request-id": "e3028bae-a90f-4ddd-a83f-cf224eba0e58",
}
}
}
- response = requests.request(
- "POST", url, data=json.dumps(data), headers=test_utils.TYPE_APPLICATION_JSON,
- auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.post_request(url, data)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Renderer service delete in progress',
time.sleep(20)
def test_45_delete_oc_service2(self):
- url = ("{}/operations/org-openroadm-service:service-delete"
- .format(test_utils.RESTCONF_BASE_URL))
+ url = "{}/operations/org-openroadm-service:service-delete"
data = {"input": {
"sdnc-request-header": {
"request-id": "e3028bae-a90f-4ddd-a83f-cf224eba0e58",
}
}
}
- response = requests.request(
- "POST", url, data=json.dumps(data), headers=test_utils.TYPE_APPLICATION_JSON,
- auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.post_request(url, data)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Renderer service delete in progress',
def test_46_get_no_oc_services(self):
print("start test")
- url = ("{}/operational/org-openroadm-service:service-list"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/operational/org-openroadm-service:service-list"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.not_found)
res = response.json()
self.assertIn(
def test_47_get_no_xc_ROADMA(self):
url = ("{}/config/network-topology:network-topology/topology/topology-netconf"
- "/node/ROADM-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ "/node/ROADM-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/")
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertNotIn(['roadm-connections'][0], res['org-openroadm-device'])
self.test_30_delete_eth_service1()
def test_50_loop_create_oc_service(self):
- url = ("{}/operational/org-openroadm-service:service-list/services/service1"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request("GET", url, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/operational/org-openroadm-service:service-list/services/service1"
+ response = test_utils.get_request(url)
if response.status_code != 404:
- url = ("{}/operations/org-openroadm-service:service-delete"
- .format(test_utils.RESTCONF_BASE_URL))
+ url = "{}/operations/org-openroadm-service:service-delete"
data = {"input": {
"sdnc-request-header": {
"request-id": "e3028bae-a90f-4ddd-a83f-cf224eba0e58",
}
}
}
- requests.request("POST", url, data=json.dumps(data), headers=test_utils.TYPE_APPLICATION_JSON,
- auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ test_utils.post_request(url, data)
time.sleep(5)
for i in range(1, 6):
import unittest
import time
import json
-#from unittest.result import failfast
import requests
from common import test_utils
self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
def test_05_connect_xprdA_to_roadmA(self):
- url = "{}/operations/transportpce-networkutils:init-xpdr-rdm-links".format(test_utils.RESTCONF_BASE_URL)
- 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=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.connect_xpdr_to_rdm_request("XPDR-A1", "1", "1",
+ "ROADM-A1", "1", "SRG1-PP1-TXRX")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"])
def test_06_connect_roadmA_to_xpdrA(self):
- url = "{}/operations/transportpce-networkutils:init-rdm-xpdr-links".format(test_utils.RESTCONF_BASE_URL)
- 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=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.connect_rdm_to_xpdr_request("XPDR-A1", "1", "1",
+ "ROADM-A1", "1", "SRG1-PP1-TXRX")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Roadm Xponder links created successfully', res["output"]["result"])
def test_07_connect_xprdC_to_roadmC(self):
- url = "{}/operations/transportpce-networkutils:init-xpdr-rdm-links".format(test_utils.RESTCONF_BASE_URL)
- 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=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.connect_xpdr_to_rdm_request("XPDR-C1", "1", "1",
+ "ROADM-C1", "1", "SRG1-PP1-TXRX")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"])
def test_08_connect_roadmC_to_xpdrC(self):
- url = "{}/operations/transportpce-networkutils:init-rdm-xpdr-links".format(test_utils.RESTCONF_BASE_URL)
- 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=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.connect_rdm_to_xpdr_request("XPDR-C1", "1", "1",
+ "ROADM-C1", "1", "SRG1-PP1-TXRX")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Roadm Xponder links created successfully', res["output"]["result"])
def test_09_create_OTS_ROADMA(self):
- url = "{}/operations/transportpce-device-renderer:create-ots-oms".format(test_utils.RESTCONF_BASE_URL)
+ url = "{}/operations/transportpce-device-renderer:create-ots-oms"
data = {
"input": {
"node-id": "ROADM-A1",
"logical-connection-point": "DEG1-TTP-TXRX"
}
}
- headers = {'content-type': 'application/json'}
- response = requests.request(
- "POST", url, data=json.dumps(data),
- headers=headers, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.post_request(url, data)
time.sleep(10)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
res["output"]["result"])
def test_10_create_OTS_ROADMC(self):
- url = "{}/operations/transportpce-device-renderer:create-ots-oms".format(test_utils.RESTCONF_BASE_URL)
+ url = "{}/operations/transportpce-device-renderer:create-ots-oms"
data = {
"input": {
"node-id": "ROADM-C1",
"logical-connection-point": "DEG2-TTP-TXRX"
}
}
- headers = {'content-type': 'application/json'}
- response = requests.request(
- "POST", url, data=json.dumps(data),
- headers=headers, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.post_request(url, data)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Interfaces OTS-DEG2-TTP-TXRX - OMS-DEG2-TTP-TXRX successfully created on node ROADM-C1',
res["output"]["result"])
def test_11_get_PM_ROADMA(self):
- url = "{}/operations/transportpce-olm:get-pm".format(test_utils.RESTCONF_BASE_URL)
+ url = "{}/operations/transportpce-olm:get-pm"
data = {
"input": {
"node-id": "ROADM-A1",
}
}
}
- headers = {'content-type': 'application/json'}
- response = requests.request(
- "POST", url, data=json.dumps(data),
- headers=headers, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.post_request(url, data)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn({
}, res["output"]["measurements"])
def test_12_get_PM_ROADMC(self):
- url = "{}/operations/transportpce-olm:get-pm".format(test_utils.RESTCONF_BASE_URL)
+ url = "{}/operations/transportpce-olm:get-pm"
data = {
"input": {
"node-id": "ROADM-C1",
}
}
}
- headers = {'content-type': 'application/json'}
- response = requests.request(
- "POST", url, data=json.dumps(data),
- headers=headers, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.post_request(url, data)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn({
}, res["output"]["measurements"])
def test_13_calculate_span_loss_base_ROADMA_ROADMC(self):
- url = "{}/operations/transportpce-olm:calculate-spanloss-base".format(test_utils.RESTCONF_BASE_URL)
+ url = "{}/operations/transportpce-olm:calculate-spanloss-base"
data = {
"input": {
"src-type": "link",
"link-id": "ROADM-A1-DEG2-DEG2-TTP-TXRXtoROADM-C1-DEG1-DEG1-TTP-TXRX"
}
}
- headers = {'content-type': 'application/json'}
- response = requests.request(
- "POST", url, data=json.dumps(data),
- headers=headers, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.post_request(url, data)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Success',
time.sleep(5)
def test_14_calculate_span_loss_base_all(self):
- url = "{}/operations/transportpce-olm:calculate-spanloss-base".format(test_utils.RESTCONF_BASE_URL)
+ url = "{}/operations/transportpce-olm:calculate-spanloss-base"
data = {
"input": {
"src-type": "all"
}
}
- headers = {'content-type': 'application/json'}
- response = requests.request(
- "POST", url, data=json.dumps(data),
- headers=headers, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.post_request(url, data)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Success',
def test_15_get_OTS_DEG2_TTP_TXRX_ROADMA(self):
url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
"node/ROADM-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/interface/OTS-DEG2-TTP-TXRX/"
- "org-openroadm-optical-transport-interfaces:ots".format(test_utils.RESTCONF_BASE_URL))
- headers = {'content-type': 'application/json'}
- response = requests.request(
- "GET", url, headers=headers, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ "org-openroadm-optical-transport-interfaces:ots")
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual(17.6, res['org-openroadm-optical-transport-interfaces:ots']['span-loss-transmit'])
def test_16_get_OTS_DEG1_TTP_TXRX_ROADMC(self):
url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
"node/ROADM-C1/yang-ext:mount/org-openroadm-device:org-openroadm-device/interface/OTS-DEG1-TTP-TXRX/"
- "org-openroadm-optical-transport-interfaces:ots".format(test_utils.RESTCONF_BASE_URL))
- headers = {'content-type': 'application/json'}
- response = requests.request(
- "GET", url, headers=headers, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ "org-openroadm-optical-transport-interfaces:ots")
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual(25.7, res['org-openroadm-optical-transport-interfaces:ots']['span-loss-transmit'])
self.assertEqual(17.6, res['org-openroadm-optical-transport-interfaces:ots']['span-loss-receive'])
def test_17_servicePath_create_AToZ(self):
- url = "{}/operations/transportpce-device-renderer:service-path".format(test_utils.RESTCONF_BASE_URL)
+ url = "{}/operations/transportpce-device-renderer:service-path"
data = {
"input": {
"service-name": "test",
]
}
}
- headers = {'content-type': 'application/json'}
- response = requests.request(
- "POST", url, data=json.dumps(data),
- headers=headers, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.post_request(url, data)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Roadm-connection successfully created for nodes', res["output"]["result"])
time.sleep(10)
def test_18_servicePath_create_ZToA(self):
- url = "{}/operations/transportpce-device-renderer:service-path".format(test_utils.RESTCONF_BASE_URL)
+ url = "{}/operations/transportpce-device-renderer:service-path"
data = {
"input": {
"service-name": "test",
]
}
}
- headers = {'content-type': 'application/json'}
- response = requests.request(
- "POST", url, data=json.dumps(data),
- headers=headers, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.post_request(url, data)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Roadm-connection successfully created for nodes', res["output"]["result"])
time.sleep(10)
def test_19_service_power_setup_XPDRA_XPDRC(self):
- url = "{}/operations/transportpce-olm:service-power-setup".format(test_utils.RESTCONF_BASE_URL)
+ url = "{}/operations/transportpce-olm:service-power-setup"
data = {
"input": {
"service-name": "test",
]
}
}
- headers = {'content-type': 'application/json'}
- response = requests.request(
- "POST", url, data=json.dumps(data),
- headers=headers, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.post_request(url, data)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Success', res["output"]["result"])
def test_20_get_interface_XPDRA_XPDR1_NETWORK1(self):
url = ("{}/config/network-topology:network-topology/topology/topology-netconf/node/XPDR-A1/yang-ext:mount/"
"org-openroadm-device:org-openroadm-device/interface/XPDR1-NETWORK1-1/"
- "org-openroadm-optical-channel-interfaces:och".format(test_utils.RESTCONF_BASE_URL))
- headers = {'content-type': 'application/json'}
- response = requests.request(
- "GET", url, headers=headers, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ "org-openroadm-optical-channel-interfaces:och")
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual(-5, res['org-openroadm-optical-channel-interfaces:och']['transmit-power'])
def test_21_get_roadmconnection_ROADMA(self):
url = ("{}/config/network-topology:network-topology/topology/topology-netconf/node/ROADM-A1/yang-ext:mount/"
"org-openroadm-device:org-openroadm-device/roadm-connections/"
- "SRG1-PP1-TXRX-DEG2-TTP-TXRX-1".format(test_utils.RESTCONF_BASE_URL))
- headers = {'content-type': 'application/json'}
- response = requests.request(
- "GET", url, headers=headers, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ "SRG1-PP1-TXRX-DEG2-TTP-TXRX-1")
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual("gainLoss", res['roadm-connections'][0]['opticalControlMode'])
def test_22_get_roadmconnection_ROADMC(self):
url = ("{}/config/network-topology:network-topology/topology/topology-netconf/node/ROADM-C1/yang-ext:mount/"
"org-openroadm-device:org-openroadm-device/roadm-connections/"
- "DEG1-TTP-TXRX-SRG1-PP1-TXRX-1".format(test_utils.RESTCONF_BASE_URL))
- headers = {'content-type': 'application/json'}
- response = requests.request(
- "GET", url, headers=headers, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ "DEG1-TTP-TXRX-SRG1-PP1-TXRX-1")
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual("power", res['roadm-connections'][0]['opticalControlMode'])
def test_23_service_power_setup_XPDRC_XPDRA(self):
- url = "{}/operations/transportpce-olm:service-power-setup".format(test_utils.RESTCONF_BASE_URL)
+ url = "{}/operations/transportpce-olm:service-power-setup"
data = {
"input": {
"service-name": "test",
]
}
}
- headers = {'content-type': 'application/json'}
- response = requests.request(
- "POST", url, data=json.dumps(data),
- headers=headers, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.post_request(url, data)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Success', res["output"]["result"])
def test_24_get_interface_XPDRC_XPDR1_NETWORK1(self):
url = ("{}/config/network-topology:network-topology/topology/topology-netconf/node/XPDR-C1/yang-ext:mount/"
"org-openroadm-device:org-openroadm-device/interface/XPDR1-NETWORK1-1/"
- "org-openroadm-optical-channel-interfaces:och".format(test_utils.RESTCONF_BASE_URL))
- headers = {'content-type': 'application/json'}
- response = requests.request(
- "GET", url, headers=headers, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ "org-openroadm-optical-channel-interfaces:och")
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual(-5, res['org-openroadm-optical-channel-interfaces:och']['transmit-power'])
def test_25_get_roadmconnection_ROADMC(self):
url = ("{}/config/network-topology:network-topology/topology/topology-netconf/node/ROADM-C1/yang-ext:mount/"
"org-openroadm-device:org-openroadm-device/roadm-connections/"
- "SRG1-PP1-TXRX-DEG1-TTP-TXRX-1".format(test_utils.RESTCONF_BASE_URL))
- headers = {'content-type': 'application/json'}
- response = requests.request(
- "GET", url, headers=headers, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ "SRG1-PP1-TXRX-DEG1-TTP-TXRX-1")
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual("gainLoss", res['roadm-connections'][0]['opticalControlMode'])
self.assertEqual(2.0, res['roadm-connections'][0]['target-output-power'])
def test_26_service_power_turndown_XPDRA_XPDRC(self):
- url = "{}/operations/transportpce-olm:service-power-turndown".format(test_utils.RESTCONF_BASE_URL)
+ url = "{}/operations/transportpce-olm:service-power-turndown"
data = {
"input": {
"service-name": "test",
]
}
}
- headers = {'content-type': 'application/json'}
- response = requests.request(
- "POST", url, data=json.dumps(data),
- headers=headers, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.post_request(url, data)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Success', res["output"]["result"])
def test_27_get_roadmconnection_ROADMA(self):
url = ("{}/config/network-topology:network-topology/topology/topology-netconf/node/ROADM-A1/yang-ext:mount/"
"org-openroadm-device:org-openroadm-device/roadm-connections/"
- "SRG1-PP1-TXRX-DEG2-TTP-TXRX-1".format(test_utils.RESTCONF_BASE_URL))
- headers = {'content-type': 'application/json'}
- response = requests.request(
- "GET", url, headers=headers, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ "SRG1-PP1-TXRX-DEG2-TTP-TXRX-1")
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual("off", res['roadm-connections'][0]['opticalControlMode'])
def test_28_get_roadmconnection_ROADMC(self):
url = ("{}/config/network-topology:network-topology/topology/topology-netconf/node/ROADM-C1/yang-ext:mount/"
"org-openroadm-device:org-openroadm-device/roadm-connections/"
- "DEG1-TTP-TXRX-SRG1-PP1-TXRX-1".format(test_utils.RESTCONF_BASE_URL))
- headers = {'content-type': 'application/json'}
- response = requests.request(
- "GET", url, headers=headers, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ "DEG1-TTP-TXRX-SRG1-PP1-TXRX-1")
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual("off", res['roadm-connections'][0]['opticalControlMode'])
def test_29_servicePath_delete_AToZ(self):
- url = "{}/operations/transportpce-device-renderer:service-path".format(test_utils.RESTCONF_BASE_URL)
+ url = "{}/operations/transportpce-device-renderer:service-path"
data = {
"input": {
"service-name": "test",
]
}
}
- headers = {'content-type': 'application/json'}
- response = requests.request(
- "POST", url, data=json.dumps(data),
- headers=headers, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.post_request(url, data)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Request processed', res["output"]["result"])
time.sleep(10)
def test_30_servicePath_delete_ZToA(self):
- url = "{}/operations/transportpce-device-renderer:service-path".format(test_utils.RESTCONF_BASE_URL)
+ url = "{}/operations/transportpce-device-renderer:service-path"
data = {
"input": {
"service-name": "test",
]
}
}
- headers = {'content-type': 'application/json'}
- response = requests.request(
- "POST", url, data=json.dumps(data),
- headers=headers, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.post_request(url, data)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Request processed', res["output"]["result"])
"""to test case where SRG where the xpdr is connected to has no optical range data"""
def test_31_connect_xprdA_to_roadmA(self):
- url = "{}/operations/transportpce-networkutils:init-xpdr-rdm-links".format(test_utils.RESTCONF_BASE_URL)
- data = {
- "networkutils:input": {
- "networkutils:links-input": {
- "networkutils:xpdr-node": "XPDR-A1",
- "networkutils:xpdr-num": "1",
- "networkutils:network-num": "2",
- "networkutils:rdm-node": "ROADM-A1",
- "networkutils:srg-num": "1",
- "networkutils:termination-point-num": "SRG1-PP2-TXRX"
- }
- }
- }
- headers = {'content-type': 'application/json'}
- response = requests.request(
- "POST", url, data=json.dumps(data),
- headers=headers, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.connect_xpdr_to_rdm_request("XPDR-A1", "1", "2",
+ "ROADM-A1", "1", "SRG1-PP2-TXRX")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"])
def test_32_connect_roadmA_to_xpdrA(self):
- url = "{}/operations/transportpce-networkutils:init-rdm-xpdr-links".format(test_utils.RESTCONF_BASE_URL)
- data = {
- "networkutils:input": {
- "networkutils:links-input": {
- "networkutils:xpdr-node": "XPDR-A1",
- "networkutils:xpdr-num": "1",
- "networkutils:network-num": "2",
- "networkutils:rdm-node": "ROADM-A1",
- "networkutils:srg-num": "1",
- "networkutils:termination-point-num": "SRG1-PP2-TXRX"
- }
- }
- }
- headers = {'content-type': 'application/json'}
- response = requests.request(
- "POST", url, data=json.dumps(data),
- headers=headers, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.connect_rdm_to_xpdr_request("XPDR-A1", "1", "2",
+ "ROADM-A1", "1", "SRG1-PP2-TXRX")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Roadm Xponder links created successfully', res["output"]["result"])
def test_33_servicePath_create_AToZ(self):
- url = "{}/operations/transportpce-device-renderer:service-path".format(test_utils.RESTCONF_BASE_URL)
+ url = "{}/operations/transportpce-device-renderer:service-path"
data = {
"input": {
"service-name": "test2",
]
}
}
- headers = {'content-type': 'application/json'}
- response = requests.request(
- "POST", url, data=json.dumps(data),
- headers=headers, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.post_request(url, data)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Roadm-connection successfully created for nodes', res["output"]["result"])
def test_34_get_interface_XPDRA_XPDR1_NETWORK2(self):
url = ("{}/config/network-topology:network-topology/topology/topology-netconf/node/XPDR-A1/yang-ext:mount/"
"org-openroadm-device:org-openroadm-device/interface/XPDR1-NETWORK2-2/"
- "org-openroadm-optical-channel-interfaces:och".format(test_utils.RESTCONF_BASE_URL))
- headers = {'content-type': 'application/json'}
- response = requests.request(
- "GET", url, headers=headers, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ "org-openroadm-optical-channel-interfaces:och")
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual(-5, res['org-openroadm-optical-channel-interfaces:och']['transmit-power'])
# self.assertEqual(2, res['org-openroadm-optical-channel-interfaces:och']['wavelength-number'])
def test_35_servicePath_delete_AToZ(self):
- url = "{}/operations/transportpce-device-renderer:service-path".format(test_utils.RESTCONF_BASE_URL)
+ url = "{}/operations/transportpce-device-renderer:service-path"
data = {
"input": {
"service-name": "test",
]
}
}
- headers = {'content-type': 'application/json'}
- response = requests.request(
- "POST", url, data=json.dumps(data),
- headers=headers, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.post_request(url, data)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Request processed', res["output"]["result"])
self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
def test_38_calculate_span_loss_current(self):
- url = "{}/operations/transportpce-olm:calculate-spanloss-current".format(test_utils.RESTCONF_BASE_URL)
- headers = {'content-type': 'application/json'}
- response = requests.request(
- "POST", url, headers=headers, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/operations/transportpce-olm:calculate-spanloss-current"
+ response = test_utils.post_request(url, None)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Success',
self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
time.sleep(10)
- url = ("{}/operational/network-topology:"
- "network-topology/topology/topology-netconf/node/SPDR-SA1"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/operational/network-topology:network-topology/topology/topology-netconf/node/SPDR-SA1"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual(
'connected')
def test_02_get_portmapping_CLIENT1(self):
- url = ("{}/config/transportpce-portmapping:network/"
- "nodes/SPDR-SA1/mapping/XPDR1-CLIENT1"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/transportpce-portmapping:network/nodes/SPDR-SA1/mapping/XPDR1-CLIENT1"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn(
res['mapping'])
def test_03_get_portmapping_NETWORK1(self):
- url = ("{}/config/transportpce-portmapping:network/"
- "nodes/SPDR-SA1/mapping/XPDR1-NETWORK1"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/transportpce-portmapping:network/nodes/SPDR-SA1/mapping/XPDR1-NETWORK1"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn(
res['mapping'])
def test_04_service_path_create_OCH_OTU4(self):
- url = "{}/operations/transportpce-device-renderer:service-path".format(test_utils.RESTCONF_BASE_URL)
+ url = "{}/operations/transportpce-device-renderer:service-path"
data = {"renderer:input": {
"service-name": "service_ODU4",
"wave-number": "1",
"nodes": [
{"node-id": "SPDR-SA1",
"dest-tp": "XPDR1-NETWORK1"}]}}
- response = requests.request(
- "POST", url, data=json.dumps(data),
- headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.post_request(url, data)
time.sleep(3)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
'och-interface-id': ['XPDR1-NETWORK1-1']}, res["output"]['node-interface'])
def test_05_get_portmapping_NETWORK1(self):
- url = ("{}/config/transportpce-portmapping:network/"
- "nodes/SPDR-SA1/mapping/XPDR1-NETWORK1"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/transportpce-portmapping:network/nodes/SPDR-SA1/mapping/XPDR1-NETWORK1"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn(
url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
"node/SPDR-SA1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
"interface/XPDR1-NETWORK1-1"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ )
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
"node/SPDR-SA1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
"interface/XPDR1-NETWORK1-OTU"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ )
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
input_dict_1 = {'name': 'XPDR1-NETWORK1-OTU',
res['interface'][0]['org-openroadm-otn-otu-interfaces:otu'])
def test_08_otn_service_path_create_ODU4(self):
- url = "{}/operations/transportpce-device-renderer:otn-service-path".format(test_utils.RESTCONF_BASE_URL)
+ url = "{}/operations/transportpce-device-renderer:otn-service-path"
data = {"renderer:input": {
"service-name": "service_ODU4",
"operation": "create",
"nodes": [
{"node-id": "SPDR-SA1",
"network-tp": "XPDR1-NETWORK1"}]}}
- response = requests.request(
- "POST", url, data=json.dumps(data),
- headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.post_request(url, data)
time.sleep(3)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
'odu-interface-id': ['XPDR1-NETWORK1-ODU4']}, res["output"]['node-interface'])
def test_09_get_portmapping_NETWORK1(self):
- url = ("{}/config/transportpce-portmapping:network/"
- "nodes/SPDR-SA1/mapping/XPDR1-NETWORK1"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/transportpce-portmapping:network/nodes/SPDR-SA1/mapping/XPDR1-NETWORK1"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn(
url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
"node/SPDR-SA1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
"interface/XPDR1-NETWORK1-ODU4"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ )
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
input_dict_1 = {'name': 'XPDR1-NETWORK1-ODU4', 'administrative-state': 'inService',
res['interface'][0]['org-openroadm-otn-odu-interfaces:odu']['opu'])
def test_11_otn_service_path_create_10GE(self):
- url = "{}/operations/transportpce-device-renderer:otn-service-path".format(test_utils.RESTCONF_BASE_URL)
+ url = "{}/operations/transportpce-device-renderer:otn-service-path"
data = {"renderer:input": {
"service-name": "service1",
"operation": "create",
{"node-id": "SPDR-SA1",
"client-tp": "XPDR1-CLIENT1",
"network-tp": "XPDR1-NETWORK1"}]}}
- response = requests.request(
- "POST", url, data=json.dumps(data),
- headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.post_request(url, data)
time.sleep(3)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
"node/SPDR-SA1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
"interface/XPDR1-CLIENT1-ETHERNET10G"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ )
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
input_dict = {'name': 'XPDR1-CLIENT1-ETHERNET10G',
url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
"node/SPDR-SA1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
"interface/XPDR1-CLIENT1-ODU2e-service1"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ )
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
"node/SPDR-SA1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
"interface/XPDR1-NETWORK1-ODU2e-service1"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ )
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
input_dict_1 = {'name': 'XPDR1-NETWORK1-ODU2e-service1', 'administrative-state': 'inService',
url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
"node/SPDR-SA1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
"odu-connection/XPDR1-CLIENT1-ODU2e-service1-x-XPDR1-NETWORK1-ODU2e-service1"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ )
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
input_dict_1 = {
res['odu-connection'][0]['source'])
def test_16_otn_service_path_delete_10GE(self):
- url = "{}/operations/transportpce-device-renderer:otn-service-path".format(test_utils.RESTCONF_BASE_URL)
+ url = "{}/operations/transportpce-device-renderer:otn-service-path"
data = {"renderer:input": {
"service-name": "service1",
"operation": "delete",
{"node-id": "SPDR-SA1",
"client-tp": "XPDR1-CLIENT1",
"network-tp": "XPDR1-NETWORK1"}]}}
- response = requests.request(
- "POST", url, data=json.dumps(data),
- headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.post_request(url, data)
time.sleep(3)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
"node/SPDR-SA1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
"odu-connection/XPDR1-CLIENT1-ODU2e-service1-x-XPDR1-NETWORK1-ODU2e-service1"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ )
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.not_found)
def test_18_check_no_interface_ODU2E_NETWORK(self):
url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
"node/SPDR-SA1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
"interface/XPDR1-NETWORK1-ODU2e-service1"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ )
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.not_found)
def test_19_check_no_interface_ODU2E_CLIENT(self):
url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
"node/SPDR-SA1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
"interface/XPDR1-CLIENT1-ODU2e-service1"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ )
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.not_found)
def test_20_check_no_interface_10GE_CLIENT(self):
url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
"node/SPDR-SA1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
"interface/XPDR1-CLIENT1-ETHERNET10G"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ )
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.not_found)
def test_21_otn_service_path_delete_ODU4(self):
- url = "{}/operations/transportpce-device-renderer:otn-service-path".format(test_utils.RESTCONF_BASE_URL)
+ url = "{}/operations/transportpce-device-renderer:otn-service-path"
data = {"renderer:input": {
"service-name": "service_ODU4",
"operation": "delete",
"nodes": [
{"node-id": "SPDR-SA1",
"network-tp": "XPDR1-NETWORK1"}]}}
- response = requests.request(
- "POST", url, data=json.dumps(data),
- headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.post_request(url, data)
time.sleep(3)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
"node/SPDR-SA1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
"interface/XPDR1-NETWORK1-ODU4"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ )
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.not_found)
def test_23_service_path_delete_OCH_OTU4(self):
- url = "{}/operations/transportpce-device-renderer:service-path".format(test_utils.RESTCONF_BASE_URL)
+ url = "{}/operations/transportpce-device-renderer:service-path"
data = {"renderer:input": {
"service-name": "service_OTU4",
"wave-number": "1",
"nodes": [
{"node-id": "SPDR-SA1",
"dest-tp": "XPDR1-NETWORK1"}]}}
- response = requests.request(
- "POST", url, data=json.dumps(data),
- headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.post_request(url, data)
time.sleep(3)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
"node/SPDR-SA1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
"interface/XPDR1-NETWORK1-OTU"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ )
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.not_found)
def test_25_check_no_interface_OCH(self):
url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
"node/SPDR-SA1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
"interface/XPDR1-NETWORK1-1"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ )
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.not_found)
def test_26_disconnect_SPDR_SA1(self):
self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
time.sleep(10)
- url = ("{}/operational/network-topology:"
- "network-topology/topology/topology-netconf/node/SPDR-SA1"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/operational/network-topology:network-topology/topology/topology-netconf/node/SPDR-SA1"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual(
'connected')
def test_02_getClliNetwork(self):
- url = ("{}/config/ietf-network:networks/network/clli-network"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/ietf-network:networks/network/clli-network"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
logging.info(res)
self.assertEqual(res['network'][0]['node'][0]['org-openroadm-clli-network:clli'], 'NodeSA')
def test_03_getOpenRoadmNetwork(self):
- url = ("{}/config/ietf-network:networks/network/openroadm-network"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/ietf-network:networks/network/openroadm-network"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual(res['network'][0]['node'][0]['node-id'], 'SPDR-SA1')
self.assertEqual(res['network'][0]['node'][0]['org-openroadm-network:ip'], '1.2.3.4')
def test_04_getLinks_OpenroadmTopology(self):
- url = ("{}/config/ietf-network:networks/network/openroadm-topology"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/ietf-network:networks/network/openroadm-topology"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
# Tests related to links
self.assertNotIn('ietf-network-topology:link', res['network'][0])
def test_05_getNodes_OpenRoadmTopology(self):
- url = ("{}/config/ietf-network:networks/network/openroadm-topology"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/ietf-network:networks/network/openroadm-topology"
+ response = test_utils.get_request(url)
res = response.json()
# Tests related to nodes
self.assertEqual(response.status_code, requests.codes.ok)
self.assertEqual(len(listNode), 0)
def test_06_getLinks_OtnTopology(self):
- url = ("{}/config/ietf-network:networks/network/otn-topology"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/ietf-network:networks/network/otn-topology"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertNotIn('ietf-network-topology:link', res['network'][0])
def test_07_getNodes_OtnTopology(self):
- url = ("{}/config/ietf-network:networks/network/otn-topology"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/ietf-network:networks/network/otn-topology"
+ response = test_utils.get_request(url)
res = response.json()
self.assertEqual(response.status_code, requests.codes.ok)
nbNode = len(res['network'][0]['node'])
self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
def test_09_getClliNetwork(self):
- url = ("{}/config/ietf-network:networks/network/clli-network"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/ietf-network:networks/network/clli-network"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
nbNode = len(res['network'][0]['node'])
self.assertEqual(res['network'][0]['node'][0]['org-openroadm-clli-network:clli'], 'NodeSA')
def test_10_getOpenRoadmNetwork(self):
- url = ("{}/config/ietf-network:networks/network/openroadm-network"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/ietf-network:networks/network/openroadm-network"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertNotIn('node', res['network'][0])
def test_11_getNodes_OpenRoadmTopology(self):
- url = ("{}/config/ietf-network:networks/network/openroadm-topology"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/ietf-network:networks/network/openroadm-topology"
+ response = test_utils.get_request(url)
res = response.json()
self.assertNotIn('node', res['network'][0])
def test_12_getNodes_OtnTopology(self):
- url = ("{}/config/ietf-network:networks/network/otn-topology"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/ietf-network:networks/network/otn-topology"
+ response = test_utils.get_request(url)
res = response.json()
self.assertNotIn('node', res['network'][0])
self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
def test_02_rdm_device_connected(self):
- url = ("{}/operational/network-topology:"
- "network-topology/topology/topology-netconf/node/ROADM-A1"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/operational/network-topology:network-topology/topology/topology-netconf/node/ROADM-A1"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual(
time.sleep(10)
def test_03_rdm_portmapping_info(self):
- url = ("{}/config/transportpce-portmapping:network/"
- "nodes/ROADM-A1/node-info"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/transportpce-portmapping:network/nodes/ROADM-A1/node-info"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual(
time.sleep(3)
def test_04_rdm_portmapping_DEG1_TTP_TXRX(self):
- url = ("{}/config/transportpce-portmapping:network/"
- "nodes/ROADM-A1/mapping/DEG1-TTP-TXRX"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/transportpce-portmapping:network/nodes/ROADM-A1/mapping/DEG1-TTP-TXRX"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn(
res['mapping'])
def test_05_rdm_portmapping_DEG2_TTP_TXRX_with_ots_oms(self):
- url = ("{}/config/transportpce-portmapping:network/"
- "nodes/ROADM-A1/mapping/DEG2-TTP-TXRX"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/transportpce-portmapping:network/nodes/ROADM-A1/mapping/DEG2-TTP-TXRX"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn(
res['mapping'])
def test_06_rdm_portmapping_SRG1_PP3_TXRX(self):
- url = ("{}/config/transportpce-portmapping:network/"
- "nodes/ROADM-A1/mapping/SRG1-PP3-TXRX"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/transportpce-portmapping:network/nodes/ROADM-A1/mapping/SRG1-PP3-TXRX"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn(
res['mapping'])
def test_07_rdm_portmapping_SRG3_PP1_TXRX(self):
- url = ("{}/config/transportpce-portmapping:network/"
- "nodes/ROADM-A1/mapping/SRG3-PP1-TXRX"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/transportpce-portmapping:network/nodes/ROADM-A1/mapping/SRG3-PP1-TXRX"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn(
self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
def test_09_xpdr_device_connected(self):
- url = ("{}/operational/network-topology:"
- "network-topology/topology/topology-netconf/node/XPDR-A1"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/operational/network-topology:network-topology/topology/topology-netconf/node/XPDR-A1"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual(
time.sleep(10)
def test_10_xpdr_portmapping_info(self):
- url = ("{}/config/transportpce-portmapping:network/"
- "nodes/XPDR-A1/node-info"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/transportpce-portmapping:network/nodes/XPDR-A1/node-info"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual(
time.sleep(3)
def test_11_xpdr_portmapping_NETWORK1(self):
- url = ("{}/config/transportpce-portmapping:network/"
- "nodes/XPDR-A1/mapping/XPDR1-NETWORK1"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/transportpce-portmapping:network/nodes/XPDR-A1/mapping/XPDR1-NETWORK1"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn(
res['mapping'])
def test_12_xpdr_portmapping_NETWORK2(self):
- url = ("{}/config/transportpce-portmapping:network/"
- "nodes/XPDR-A1/mapping/XPDR1-NETWORK2"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/transportpce-portmapping:network/nodes/XPDR-A1/mapping/XPDR1-NETWORK2"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn(
res['mapping'])
def test_13_xpdr_portmapping_CLIENT1(self):
- url = ("{}/config/transportpce-portmapping:network/"
- "nodes/XPDR-A1/mapping/XPDR1-CLIENT1"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/transportpce-portmapping:network/nodes/XPDR-A1/mapping/XPDR1-CLIENT1"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn(
res['mapping'])
def test_14_xpdr_portmapping_CLIENT2(self):
- url = ("{}/config/transportpce-portmapping:network/"
- "nodes/XPDR-A1/mapping/XPDR1-CLIENT2"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/transportpce-portmapping:network/nodes/XPDR-A1/mapping/XPDR1-CLIENT2"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn(
self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
def test_16_xpdr_device_disconnected(self):
- url = ("{}/operational/network-topology:network-topology/topology/"
- "topology-netconf/node/XPDR-A1".format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/operational/network-topology:network-topology/topology/topology-netconf/node/XPDR-A1"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.not_found)
res = response.json()
self.assertIn(
res['errors']['error'])
def test_17_xpdr_device_not_connected(self):
- url = ("{}/config/transportpce-portmapping:network/nodes/XPDR-A1".format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/transportpce-portmapping:network/nodes/XPDR-A1"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.not_found)
res = response.json()
self.assertIn(
self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
def test_19_rdm_device_disconnected(self):
- url = ("{}/operational/network-topology:network-topology/topology/topology-netconf/node/ROADM-A1"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/operational/network-topology:network-topology/topology/topology-netconf/node/ROADM-A1"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.not_found)
res = response.json()
self.assertIn(
res['errors']['error'])
def test_20_rdm_device_not_connected(self):
- url = ("{}/config/transportpce-portmapping:network/nodes/ROADM-A1".format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/transportpce-portmapping:network/nodes/ROADM-A1"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.not_found)
res = response.json()
self.assertIn(
self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
def test_03_rdm_portmapping(self):
- url = ("{}/config/transportpce-portmapping:network/"
- "nodes/ROADM-A1"
- .format(test_utils.RESTCONF_BASE_URL))
- headers = {'content-type': 'application/json'}
- response = requests.request(
- "GET", url, headers=headers, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/transportpce-portmapping:network/nodes/ROADM-A1"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn(
res['nodes'][0]['mapping'])
def test_04_xpdr_portmapping(self):
- url = ("{}/config/transportpce-portmapping:network/"
- "nodes/XPDR-A1"
- .format(test_utils.RESTCONF_BASE_URL))
- headers = {'content-type': 'application/json'}
- response = requests.request(
- "GET", url, headers=headers, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/transportpce-portmapping:network/nodes/XPDR-A1"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn(
res['nodes'][0]['mapping'])
def test_05_service_path_create(self):
- url = "{}/operations/transportpce-device-renderer:service-path".format(test_utils.RESTCONF_BASE_URL)
+ url = "{}/operations/transportpce-device-renderer:service-path"
data = {"renderer:input": {
"renderer:service-name": "service_test",
"renderer:wave-number": "7",
{"renderer:node-id": "XPDR-A1",
"renderer:src-tp": "XPDR1-CLIENT1",
"renderer:dest-tp": "XPDR1-NETWORK1"}]}}
- headers = {'content-type': 'application/json'}
- response = requests.request(
- "POST", url, data=json.dumps(data),
- headers=headers, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.post_request(url, data)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('Roadm-connection successfully created for nodes: ROADM-A1', res["output"]["result"])
url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
"node/ROADM-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
"interface/DEG1-TTP-TXRX-nmc-7"
- .format(test_utils.RESTCONF_BASE_URL))
- headers = {'content-type': 'application/json'}
- response = requests.request(
- "GET", url, headers=headers, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ )
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
# the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
"node/ROADM-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
"interface/DEG1-TTP-TXRX-mc-7"
- .format(test_utils.RESTCONF_BASE_URL))
- headers = {'content-type': 'application/json'}
- response = requests.request(
- "GET", url, headers=headers, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ )
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
# the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
"node/ROADM-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
"interface/SRG1-PP3-TXRX-nmc-7"
- .format(test_utils.RESTCONF_BASE_URL))
- headers = {'content-type': 'application/json'}
- response = requests.request(
- "GET", url, headers=headers, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ )
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
# the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
"node/ROADM-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
"interface/SRG1-PP3-TXRX-mc-7"
- .format(test_utils.RESTCONF_BASE_URL))
- headers = {'content-type': 'application/json'}
- response = requests.request(
- "GET", url, headers=headers, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ )
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.not_found)
res = response.json()
self.assertIn(
url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
"node/ROADM-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
"roadm-connections/SRG1-PP3-TXRX-DEG1-TTP-TXRX-7"
- .format(test_utils.RESTCONF_BASE_URL))
- headers = {'content-type': 'application/json'}
- response = requests.request(
- "GET", url, headers=headers, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ )
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
# the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
"node/XPDR-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
"interface/XPDR1-NETWORK1-7"
- .format(test_utils.RESTCONF_BASE_URL))
- headers = {'content-type': 'application/json'}
- response = requests.request(
- "GET", url, headers=headers, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ )
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
# the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
"node/XPDR-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
"interface/XPDR1-NETWORK1-OTU"
- .format(test_utils.RESTCONF_BASE_URL))
- headers = {'content-type': 'application/json'}
- response = requests.request(
- "GET", url, headers=headers, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ )
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
# the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
"node/XPDR-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
"interface/XPDR1-NETWORK1-ODU"
- .format(test_utils.RESTCONF_BASE_URL))
- headers = {'content-type': 'application/json'}
- response = requests.request(
- "GET", url, headers=headers, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ )
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
# the 2 following statements replace self.assertDictContainsSubset deprecated in python 3.2
url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
"node/XPDR-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
"interface/XPDR1-CLIENT1-ETHERNET"
- .format(test_utils.RESTCONF_BASE_URL))
- headers = {'content-type': 'application/json'}
- response = requests.request(
- "GET", url, headers=headers, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ )
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
# the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
"node/XPDR-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
"circuit-packs/1%2F0%2F1-PLUG-NET"
- .format(test_utils.RESTCONF_BASE_URL))
- headers = {'content-type': 'application/json'}
- response = requests.request(
- "GET", url, headers=headers, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ )
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('not-reserved-inuse', res['circuit-packs'][0]["equipment-state"])
url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
"node/XPDR-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
"circuit-packs/1%2F0%2F1-PLUG-CLIENT"
- .format(test_utils.RESTCONF_BASE_URL))
- headers = {'content-type': 'application/json'}
- response = requests.request(
- "GET", url, headers=headers, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ )
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn('not-reserved-inuse', res['circuit-packs'][0]["equipment-state"])
def test_17_service_path_delete(self):
- url = "{}/operations/transportpce-device-renderer:service-path".format(test_utils.RESTCONF_BASE_URL)
+ url = "{}/operations/transportpce-device-renderer:service-path"
data = {"renderer:input": {
"renderer:service-name": "service_test",
"renderer:wave-number": "7",
{"renderer:node-id": "XPDR-A1",
"renderer:src-tp": "XPDR1-CLIENT1",
"renderer:dest-tp": "XPDR1-NETWORK1"}]}}
- headers = {'content-type': 'application/json'}
- response = requests.request(
- "POST", url, data=json.dumps(data),
- headers=headers, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.post_request(url, data)
self.assertEqual(response.status_code, requests.codes.ok)
self.assertEqual(response.json(), {
'output': {'result': 'Request processed', 'success': True}})
url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
"node/ROADM-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
"interface/DEG1-TTP-TXRX-mc-7"
- .format(test_utils.RESTCONF_BASE_URL))
- headers = {'content-type': 'application/json'}
- response = requests.request(
- "GET", url, headers=headers, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ )
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.not_found)
res = response.json()
self.assertIn(
url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
"node/ROADM-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
"interface/DEG1-TTP-TXRX-nmc-7"
- .format(test_utils.RESTCONF_BASE_URL))
- headers = {'content-type': 'application/json'}
- response = requests.request(
- "GET", url, headers=headers, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ )
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.not_found)
res = response.json()
self.assertIn(
url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
"node/ROADM-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
"interface/SRG1-PP3-TXRX-mc-7"
- .format(test_utils.RESTCONF_BASE_URL))
- headers = {'content-type': 'application/json'}
- response = requests.request(
- "GET", url, headers=headers, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ )
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.not_found)
res = response.json()
self.assertIn({
url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
"node/ROADM-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
"interface/SRG1-PP3-TXRX-nmc-7"
- .format(test_utils.RESTCONF_BASE_URL))
- headers = {'content-type': 'application/json'}
- response = requests.request(
- "GET", url, headers=headers, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ )
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.not_found)
res = response.json()
self.assertIn({
url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
"node/ROADM-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
"roadm-connections/SRG1-PP3-TXRX-DEG1-TTP-TXRX-7"
- .format(test_utils.RESTCONF_BASE_URL))
- headers = {'content-type': 'application/json'}
- response = requests.request(
- "GET", url, headers=headers, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ )
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.not_found)
res = response.json()
self.assertIn({
url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
"node/XPDR-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
"interface/XPDR1-NETWORK1-7"
- .format(test_utils.RESTCONF_BASE_URL))
- headers = {'content-type': 'application/json'}
- response = requests.request(
- "GET", url, headers=headers, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ )
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.not_found)
res = response.json()
self.assertIn({
url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
"node/XPDR-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
"interface/XPDR1-NETWORK1-OTU"
- .format(test_utils.RESTCONF_BASE_URL))
- headers = {'content-type': 'application/json'}
- response = requests.request(
- "GET", url, headers=headers, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ )
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.not_found)
res = response.json()
self.assertIn({
url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
"node/XPDR-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
"interface/XPDR1-NETWORK1-ODU"
- .format(test_utils.RESTCONF_BASE_URL))
- headers = {'content-type': 'application/json'}
- response = requests.request(
- "GET", url, headers=headers, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ )
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.not_found)
res = response.json()
self.assertIn({
url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
"node/XPDR-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
"interface/XPDR1-CLIENT1-ETHERNET"
- .format(test_utils.RESTCONF_BASE_URL))
- headers = {'content-type': 'application/json'}
- response = requests.request(
- "GET", url, headers=headers, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ )
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.not_found)
res = response.json()
self.assertIn({
url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
"node/XPDR-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
"circuit-packs/1%2F0%2F1-PLUG-NET"
- .format(test_utils.RESTCONF_BASE_URL))
- headers = {'content-type': 'application/json'}
- response = requests.request(
- "GET", url, headers=headers, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ )
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual('not-reserved-available', res["circuit-packs"][0]['equipment-state'])
url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
"node/XPDR-A1/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
"circuit-packs/1%2F0%2F1-PLUG-CLIENT"
- .format(test_utils.RESTCONF_BASE_URL))
- headers = {'content-type': 'application/json'}
- response = requests.request(
- "GET", url, headers=headers, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ )
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual('not-reserved-available', res["circuit-packs"][0]['equipment-state'])
time.sleep(20)
def test_05_connect_xprda_n1_to_roadma_pp1(self):
- url = "{}/operations/transportpce-networkutils:init-xpdr-rdm-links".format(test_utils.RESTCONF_BASE_URL)
- data = test_utils.generate_link_data("XPDR-A1", "1", "1", "ROADM-A1", "1", "SRG1-PP1-TXRX")
- response = test_utils.post_request(url, data, test_utils.ODL_LOGIN, test_utils.ODL_PWD)
+ response = test_utils.connect_xpdr_to_rdm_request("XPDR-A1", "1", "1",
+ "ROADM-A1", "1", "SRG1-PP1-TXRX")
self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
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(test_utils.RESTCONF_BASE_URL)
- data = test_utils.generate_link_data("XPDR-A1", "1", "1", "ROADM-A1", "1", "SRG1-PP1-TXRX")
- response = test_utils.post_request(url, data, test_utils.ODL_LOGIN, test_utils.ODL_PWD)
+ response = test_utils.connect_rdm_to_xpdr_request("XPDR-A1", "1", "1",
+ "ROADM-A1", "1", "SRG1-PP1-TXRX")
self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
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(test_utils.RESTCONF_BASE_URL)
- data = test_utils.generate_link_data("XPDR-C1", "1", "1", "ROADM-C1", "1", "SRG1-PP1-TXRX")
- response = test_utils.post_request(url, data, test_utils.ODL_LOGIN, test_utils.ODL_PWD)
+ response = test_utils.connect_xpdr_to_rdm_request("XPDR-C1", "1", "1",
+ "ROADM-C1", "1", "SRG1-PP1-TXRX")
self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
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(test_utils.RESTCONF_BASE_URL)
- data = test_utils.generate_link_data("XPDR-C1", "1", "1", "ROADM-C1", "1", "SRG1-PP1-TXRX")
- response = test_utils.post_request(url, data, test_utils.ODL_LOGIN, test_utils.ODL_PWD)
+ response = test_utils.connect_rdm_to_xpdr_request("XPDR-C1", "1", "1",
+ "ROADM-C1", "1", "SRG1-PP1-TXRX")
self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
res = response.json()
self.assertIn('Roadm Xponder links created successfully', res["output"]["result"],
time.sleep(2)
def test_09_connect_xprda_n2_to_roadma_pp2(self):
- url = "{}/operations/transportpce-networkutils:init-xpdr-rdm-links".format(test_utils.RESTCONF_BASE_URL)
- data = test_utils.generate_link_data("XPDR-A1", "1", "2", "ROADM-A1", "1", "SRG1-PP2-TXRX")
- response = test_utils.post_request(url, data, test_utils.ODL_LOGIN, test_utils.ODL_PWD)
+ response = test_utils.connect_xpdr_to_rdm_request("XPDR-A1", "1", "2",
+ "ROADM-A1", "1", "SRG1-PP2-TXRX")
self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
res = response.json()
self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"],
time.sleep(2)
def test_10_connect_roadma_pp2_to_xpdra_n2(self):
- url = "{}/operations/transportpce-networkutils:init-rdm-xpdr-links".format(test_utils.RESTCONF_BASE_URL)
- data = test_utils.generate_link_data("XPDR-A1", "1", "2", "ROADM-A1", "1", "SRG1-PP2-TXRX")
- response = test_utils.post_request(url, data, test_utils.ODL_LOGIN, test_utils.ODL_PWD)
+ response = test_utils.connect_rdm_to_xpdr_request("XPDR-A1", "1", "2",
+ "ROADM-A1", "1", "SRG1-PP2-TXRX")
self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
res = response.json()
self.assertIn('Roadm Xponder links created successfully', res["output"]["result"],
time.sleep(2)
def test_11_connect_xprdc_n2_to_roadmc_pp2(self):
- url = "{}/operations/transportpce-networkutils:init-xpdr-rdm-links".format(test_utils.RESTCONF_BASE_URL)
- data = test_utils.generate_link_data("XPDR-C1", "1", "2", "ROADM-C1", "1", "SRG1-PP2-TXRX")
- response = test_utils.post_request(url, data, test_utils.ODL_LOGIN, test_utils.ODL_PWD)
+ response = test_utils.connect_xpdr_to_rdm_request("XPDR-C1", "1", "2",
+ "ROADM-C1", "1", "SRG1-PP2-TXRX")
self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
res = response.json()
self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"],
time.sleep(2)
def test_12_connect_roadmc_pp2_to_xpdrc_n2(self):
- url = "{}/operations/transportpce-networkutils:init-rdm-xpdr-links".format(test_utils.RESTCONF_BASE_URL)
- data = test_utils.generate_link_data("XPDR-C1", "1", "2", "ROADM-C1", "1", "SRG1-PP2-TXRX")
- response = test_utils.post_request(url, data, test_utils.ODL_LOGIN, test_utils.ODL_PWD)
+ response = test_utils.connect_rdm_to_xpdr_request("XPDR-C1", "1", "2",
+ "ROADM-C1", "1", "SRG1-PP2-TXRX")
self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
res = response.json()
self.assertIn('Roadm Xponder links created successfully', res["output"]["result"],
time.sleep(2)
def test_13_get_tapi_openroadm_topology(self):
- url = "{}/operations/tapi-topology:get-topology-details".format(test_utils.RESTCONF_BASE_URL)
+ url = "{}/operations/tapi-topology:get-topology-details"
data = {
"tapi-topology:input": {
"tapi-topology:topology-id-or-name": "openroadm-topology"
}
}
- response = test_utils.post_request(url, data, test_utils.ODL_LOGIN, test_utils.ODL_PWD)
+ response = test_utils.post_request(url, data)
self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
res = response.json()
self.assertEqual(len(res["output"]["topology"]["node"]), 1, 'There should be 1 node')
'There should be 4 owned-node-edge-points')
def test_14_get_tapi_otn_topology(self):
- url = "{}/operations/tapi-topology:get-topology-details".format(test_utils.RESTCONF_BASE_URL)
+ url = "{}/operations/tapi-topology:get-topology-details"
data = {
"tapi-topology:input": {
"tapi-topology:topology-id-or-name": "otn-topology"
}
}
- response = test_utils.post_request(url, data, test_utils.ODL_LOGIN, test_utils.ODL_PWD)
+ response = test_utils.post_request(url, data)
self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
res = response.json()
self.assertEqual(len(res["output"]["topology"]["node"]), 4, 'There should be 4 nodes')
# Verify the termination points of the ROADMA
def test_02_compareOpenroadmTopologyPortMapping_rdm(self):
- urlTopo = ("{}/config/ietf-network:networks/network/openroadm-topology"
- .format(test_utils.RESTCONF_BASE_URL))
- responseTopo = requests.request(
- "GET", urlTopo, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ urlTopo = "{}/config/ietf-network:networks/network/openroadm-topology"
+ responseTopo = test_utils.get_request(urlTopo)
resTopo = responseTopo.json()
nbNode = len(resTopo['network'][0]['node'])
nbMapCumul = 0
nodeMapId = nodeId.split("-")[0] + "-" + nodeId.split("-")[1]
print("nodeMapId={}".format(nodeMapId))
urlMapList = "{}/config/transportpce-portmapping:network/nodes/" + nodeMapId
- urlMapListFull = urlMapList.format(test_utils.RESTCONF_BASE_URL)
- responseMapList = requests.request(
- "GET", urlMapListFull, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ responseMapList = test_utils.get_request(urlMapList)
resMapList = responseMapList.json()
nbMappings = len(resMapList['nodes'][0]['mapping']) - nbMapCumul
tpId = resTopo['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]['tp-id']
if((not "CP" in tpId) and (not "CTP" in tpId)):
urlMap = "{}/config/transportpce-portmapping:network/nodes/" + nodeMapId + "/mapping/" + tpId
- urlMapFull = urlMap.format(test_utils.RESTCONF_BASE_URL)
- responseMap = requests.request(
- "GET", urlMapFull, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ responseMap = test_utils.get_request(urlMap)
self.assertEqual(responseMap.status_code, requests.codes.ok)
if(responseMap.status_code == requests.codes.ok):
nbMapCurrent += 1
# Disconnect the ROADMA
def test_03_disconnect_rdm(self):
- url = ("{}/config/network-topology:"
- "network-topology/topology/topology-netconf/node/ROADM-A1"
- .format(test_utils.RESTCONF_BASE_URL))
- data = {}
- response = requests.request(
- "DELETE", url, data=json.dumps(data), headers=test_utils.TYPE_APPLICATION_JSON,
- auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
- self.assertEqual(response.status_code, requests.codes.ok)
+ response = test_utils.unmount_device("ROADM-A1")
+ self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
# #Connect the XPDRA
def test_04_connect_xpdr(self):
# Disconnect the XPDRA
def test_06_disconnect_device(self):
- url = ("{}/config/network-topology:"
- "network-topology/topology/topology-netconf/node/XPDR-A1"
- .format(test_utils.RESTCONF_BASE_URL))
- data = {}
- response = requests.request(
- "DELETE", url, data=json.dumps(data), headers=test_utils.TYPE_APPLICATION_JSON,
- auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
- self.assertEqual(response.status_code, requests.codes.ok)
+ response = test_utils.unmount_device("XPDR-A1")
+ self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
if __name__ == "__main__":
self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
def test_02_getClliNetwork(self):
- url = ("{}/config/ietf-network:networks/network/clli-network"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/ietf-network:networks/network/clli-network"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
logging.info(res)
self.assertEqual(res['network'][0]['node'][0]['org-openroadm-clli-network:clli'], 'NodeA')
def test_03_getOpenRoadmNetwork(self):
- url = ("{}/config/ietf-network:networks/network/openroadm-network"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/ietf-network:networks/network/openroadm-network"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual(res['network'][0]['node'][0]['node-id'], 'ROADM-A1')
self.assertEqual(res['network'][0]['node'][0]['org-openroadm-network:model'], 'model2')
def test_04_getLinks_OpenroadmTopology(self):
- url = ("{}/config/ietf-network:networks/network/openroadm-topology"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/ietf-network:networks/network/openroadm-topology"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
# Tests related to links
self.assertEqual(len(dropLink), 0)
def test_05_getNodes_OpenRoadmTopology(self):
- url = ("{}/config/ietf-network:networks/network/openroadm-topology"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/ietf-network:networks/network/openroadm-topology"
+ response = test_utils.get_request(url)
res = response.json()
# Tests related to nodes
self.assertEqual(response.status_code, requests.codes.ok)
self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
def test_07_getClliNetwork(self):
- url = ("{}/config/ietf-network:networks/network/clli-network"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/ietf-network:networks/network/clli-network"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual(res['network'][0]['node'][0]['node-id'], 'NodeA')
self.assertEqual(res['network'][0]['node'][0]['org-openroadm-clli-network:clli'], 'NodeA')
def test_08_getOpenRoadmNetwork(self):
- url = ("{}/config/ietf-network:networks/network/openroadm-network"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/ietf-network:networks/network/openroadm-network"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
nbNode = len(res['network'][0]['node'])
self.assertFalse(True)
def test_09_getNodes_OpenRoadmTopology(self):
- url = ("{}/config/ietf-network:networks/network/openroadm-topology"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/ietf-network:networks/network/openroadm-topology"
+ response = test_utils.get_request(url)
res = response.json()
# Tests related to nodes
self.assertEqual(response.status_code, requests.codes.ok)
# Connect the tail XPDRA to ROADMA and vice versa
def test_10_connect_tail_xpdr_rdm(self):
# Connect the tail: XPDRA to ROADMA
- url = ("{}/operations/transportpce-networkutils:init-xpdr-rdm-links"
- .format(test_utils.RESTCONF_BASE_URL))
+ url = "{}/operations/transportpce-networkutils:init-xpdr-rdm-links"
data = {"networkutils:input": {
"networkutils:links-input": {
"networkutils:xpdr-node": "XPDR-A1",
}
}
}
- response = requests.request(
- "POST", url, data=json.dumps(data), headers=test_utils.TYPE_APPLICATION_JSON,
- auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.post_request(url, data)
self.assertEqual(response.status_code, requests.codes.ok)
def test_11_connect_tail_rdm_xpdr(self):
# Connect the tail: ROADMA to XPDRA
- url = ("{}/operations/transportpce-networkutils:init-rdm-xpdr-links"
- .format(test_utils.RESTCONF_BASE_URL))
+ url = "{}/operations/transportpce-networkutils:init-rdm-xpdr-links"
data = {"networkutils:input": {
"networkutils:links-input": {
"networkutils:xpdr-node": "XPDR-A1",
}
}
}
- response = requests.request(
- "POST", url, data=json.dumps(data), headers=test_utils.TYPE_APPLICATION_JSON,
- auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.post_request(url, data)
self.assertEqual(response.status_code, requests.codes.ok)
def test_12_getLinks_OpenRoadmTopology(self):
- url = ("{}/config/ietf-network:networks/network/openroadm-topology"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/ietf-network:networks/network/openroadm-topology"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
# Tests related to links
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(test_utils.RESTCONF_BASE_URL))
+ )
data = {"span": {
"auto-spanloss": "true",
"engineered-spanloss": 12.2,
"fiber-type": "smf",
"SRLG-length": 100000,
"pmd": 0.5}]}}
- response = requests.request(
- "PUT", url, data=json.dumps(data), headers=test_utils.TYPE_APPLICATION_JSON,
- auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.put_request(url, data)
self.assertEqual(response.status_code, requests.codes.created)
def test_15_omsAttributes_ROADMC_ROADMA(self):
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(test_utils.RESTCONF_BASE_URL))
+ )
data = {"span": {
"auto-spanloss": "true",
"engineered-spanloss": 12.2,
"SRLG-length": 100000,
"pmd": 0.5}]}}
- response = requests.request(
- "PUT", url, data=json.dumps(data), headers=test_utils.TYPE_APPLICATION_JSON,
- auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.put_request(url, data)
self.assertEqual(response.status_code, requests.codes.created)
def test_16_getClliNetwork(self):
- url = ("{}/config/ietf-network:networks/network/clli-network"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/ietf-network:networks/network/clli-network"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
nbNode = len(res['network'][0]['node'])
self.assertEqual(len(listNode), 0)
def test_17_getOpenRoadmNetwork(self):
- url = ("{}/config/ietf-network:networks/network/openroadm-network"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/ietf-network:networks/network/openroadm-network"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
nbNode = len(res['network'][0]['node'])
self.assertEqual(len(listNode), 0)
def test_18_getROADMLinkOpenRoadmTopology(self):
- url = ("{}/config/ietf-network:networks/network/openroadm-topology"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/ietf-network:networks/network/openroadm-topology"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
# Tests related to links
self.assertEqual(len(XPDR_OUT), 0)
def test_19_getLinkOmsAttributesOpenRoadmTopology(self):
- url = ("{}/config/ietf-network:networks/network/openroadm-topology"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/ietf-network:networks/network/openroadm-topology"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
# Tests related to links
self.assertEqual(len(R2RLink), 0)
def test_20_getNodes_OpenRoadmTopology(self):
- url = ("{}/config/ietf-network:networks/network/openroadm-topology"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/ietf-network:networks/network/openroadm-topology"
+ response = test_utils.get_request(url)
res = response.json()
# Tests related to nodes
self.assertEqual(response.status_code, requests.codes.ok)
url = ("{}/config/ietf-network:networks/network/openroadm-topology/ietf-network-topology:"
"link/ROADM-A1-DEG1-DEG1-TTP-TXRXtoROADM-B1-DEG1-DEG1-TTP-TXRX/org-openroadm-network-topology:"
"OMS-attributes/span"
- .format(test_utils.RESTCONF_BASE_URL))
+ )
data = {"span": {
"auto-spanloss": "true",
"engineered-spanloss": 12.2,
"fiber-type": "smf",
"SRLG-length": 100000,
"pmd": 0.5}]}}
- response = requests.request(
- "PUT", url, data=json.dumps(data), headers=test_utils.TYPE_APPLICATION_JSON,
- auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.put_request(url, data)
self.assertEqual(response.status_code, requests.codes.created)
def test_23_omsAttributes_ROADMB_ROADMA(self):
url = ("{}/config/ietf-network:networks/network/openroadm-topology/ietf-network-topology:"
"link/ROADM-B1-DEG1-DEG1-TTP-TXRXtoROADM-A1-DEG1-DEG1-TTP-TXRX/org-openroadm-network-topology:"
"OMS-attributes/span"
- .format(test_utils.RESTCONF_BASE_URL))
+ )
data = {"span": {
"auto-spanloss": "true",
"engineered-spanloss": 12.2,
"fiber-type": "smf",
"SRLG-length": 100000,
"pmd": 0.5}]}}
- response = requests.request(
- "PUT", url, data=json.dumps(data), headers=test_utils.TYPE_APPLICATION_JSON,
- auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.put_request(url, data)
self.assertEqual(response.status_code, requests.codes.created)
def test_24_omsAttributes_ROADMB_ROADMC(self):
url = ("{}/config/ietf-network:networks/network/openroadm-topology/ietf-network-topology:"
"link/ROADM-B1-DEG2-DEG2-TTP-TXRXtoROADM-C1-DEG2-DEG2-TTP-TXRX/org-openroadm-network-topology:"
"OMS-attributes/span"
- .format(test_utils.RESTCONF_BASE_URL))
+ )
data = {"span": {
"auto-spanloss": "true",
"engineered-spanloss": 12.2,
"fiber-type": "smf",
"SRLG-length": 100000,
"pmd": 0.5}]}}
- response = requests.request(
- "PUT", url, data=json.dumps(data), headers=test_utils.TYPE_APPLICATION_JSON,
- auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.put_request(url, data)
self.assertEqual(response.status_code, requests.codes.created)
def test_25_omsAttributes_ROADMC_ROADMB(self):
url = ("{}/config/ietf-network:networks/network/openroadm-topology/ietf-network-topology:"
"link/ROADM-C1-DEG2-DEG2-TTP-TXRXtoROADM-B1-DEG2-DEG2-TTP-TXRX/org-openroadm-network-topology:"
"OMS-attributes/span"
- .format(test_utils.RESTCONF_BASE_URL))
+ )
data = {"span": {
"auto-spanloss": "true",
"engineered-spanloss": 12.2,
"fiber-type": "smf",
"SRLG-length": 100000,
"pmd": 0.5}]}}
- response = requests.request(
- "PUT", url, data=json.dumps(data), headers=test_utils.TYPE_APPLICATION_JSON,
- auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.put_request(url, data)
self.assertEqual(response.status_code, requests.codes.created)
def test_26_getClliNetwork(self):
- url = ("{}/config/ietf-network:networks/network/clli-network"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/ietf-network:networks/network/clli-network"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
nbNode = len(res['network'][0]['node'])
self.assertEqual(len(listNode), 0)
def test_27_verifyDegree(self):
- url = ("{}/config/ietf-network:networks/network/openroadm-topology"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/ietf-network:networks/network/openroadm-topology"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
# Tests related to links
self.assertEqual(len(listR2RLink), 0)
def test_28_verifyOppositeLinkTopology(self):
- url = ("{}/config/ietf-network:networks/network/openroadm-topology"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/ietf-network:networks/network/openroadm-topology"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
# Tests related to links
oppLink_id = res['network'][0]['ietf-network-topology:link'][i]['org-openroadm-common-network:opposite-link']
# Find the opposite link
url_oppLink = "{}/config/ietf-network:networks/network/openroadm-topology/ietf-network-topology:link/"+oppLink_id
- url = (url_oppLink.format(test_utils.RESTCONF_BASE_URL))
- response_oppLink = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response_oppLink = test_utils.get_request(url_oppLink)
self.assertEqual(response_oppLink.status_code, requests.codes.ok)
res_oppLink = response_oppLink.json()
self.assertEqual(res_oppLink['ietf-network-topology:link'][0]
self.assertEqual(oppLink_type, 'XPONDER-INPUT')
def test_29_getLinkOmsAttributesOpenRoadmTopology(self):
- url = ("{}/config/ietf-network:networks/network/openroadm-topology"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/ietf-network:networks/network/openroadm-topology"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
nbLink = len(res['network'][0]['ietf-network-topology:link'])
# Delete in the topology-netconf
response = test_utils.unmount_device("ROADM-B1")
# Delete in the clli-network
- url = ("{}/config/ietf-network:networks/network/clli-network/node/NodeB"
- .format(test_utils.RESTCONF_BASE_URL))
+ url = "{}/config/ietf-network:networks/network/clli-network/node/NodeB"
data = {}
- response = requests.request(
- "DELETE", url, data=json.dumps(data), headers=test_utils.TYPE_APPLICATION_JSON,
- auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.delete_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
def test_31_disconnect_ROADMC(self):
response = test_utils.unmount_device("ROADM-C1")
self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
# Delete in the clli-network
- url = ("{}/config/ietf-network:networks/network/clli-network/node/NodeC"
- .format(test_utils.RESTCONF_BASE_URL))
+ url = "{}/config/ietf-network:networks/network/clli-network/node/NodeC"
data = {}
- response = requests.request(
- "DELETE", url, data=json.dumps(data), headers=test_utils.TYPE_APPLICATION_JSON,
- auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.delete_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
# def test_24_check_roadm2roadm_links_deletion(self):
-# url = ("{}/config/ietf-network:networks/network/openroadm-topology"
-# .format(test_utils.RESTCONF_BASE_URL))
-# response = requests.request(
-# "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+# url = "{}/config/ietf-network:networks/network/openroadm-topology"
+# response = test_utils.request(url)
# self.assertEqual(response.status_code, requests.codes.ok)
# res = response.json()
# #Write the response in the log
# self.assertNotEqual(res['network'][0]['ietf-network-topology:link'][i]['link-id'],'ROADMC-DEG2-DEG2-CTP-TXRXtoROADMC-DEG1-DEG1-CTP-TXRX')
def test_32_getNodes_OpenRoadmTopology(self):
- url = ("{}/config/ietf-network:networks/network/openroadm-topology"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/ietf-network:networks/network/openroadm-topology"
+ response = test_utils.get_request(url)
res = response.json()
# Tests related to nodes
self.assertEqual(response.status_code, requests.codes.ok)
self.assertNotEqual(res['network'][0]['node'][i]['node-id'], 'ROADM-C1-DEG2')
def test_33_getOpenRoadmNetwork(self):
- url = ("{}/config/ietf-network:networks/network/openroadm-network"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/ietf-network:networks/network/openroadm-network"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
nbNode = len(res['network'][0]['node'])
self.assertNotEqual(res['network'][0]['node'][i]['node-id'], 'ROADM-B1')
def test_34_getClliNetwork(self):
- url = ("{}/config/ietf-network:networks/network/clli-network"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/ietf-network:networks/network/clli-network"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
nbNode = len(res['network'][0]['node'])
self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
def test_36_getClliNetwork(self):
- url = ("{}/config/ietf-network:networks/network/clli-network"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/ietf-network:networks/network/clli-network"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
nbNode = len(res['network'][0]['node'])
self.assertEqual(res['network'][0]['node'][0]['org-openroadm-clli-network:clli'], 'NodeA')
def test_37_getOpenRoadmNetwork(self):
- url = ("{}/config/ietf-network:networks/network/openroadm-network"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/ietf-network:networks/network/openroadm-network"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
nbNode = len(res['network'][0]['node'])
self.assertNotEqual(res['network'][0]['node'][i]['node-id'], 'XPDR-A1')
def test_38_getNodes_OpenRoadmTopology(self):
- url = ("{}/config/ietf-network:networks/network/openroadm-topology"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/ietf-network:networks/network/openroadm-topology"
+ response = test_utils.get_request(url)
res = response.json()
# Tests related to nodes
self.assertEqual(response.status_code, requests.codes.ok)
# Link-1
url = ("{}/config/ietf-network:networks/network/openroadm-topology/ietf-network-topology:"
"link/XPDR-A1-XPDR1-XPDR1-NETWORK1toROADM-A1-SRG1-SRG1-PP1-TXRX"
- .format(test_utils.RESTCONF_BASE_URL))
+ )
data = {}
- response = requests.request(
- "DELETE", url, data=json.dumps(data), headers=test_utils.TYPE_APPLICATION_JSON,
- auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.delete_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
# Link-2
url = ("{}/config/ietf-network:networks/network/openroadm-topology/ietf-network-topology:"
"link/ROADM-A1-SRG1-SRG1-PP1-TXRXtoXPDR-A1-XPDR1-XPDR1-NETWORK1"
- .format(test_utils.RESTCONF_BASE_URL))
+ )
data = {}
- response = requests.request(
- "DELETE", url, data=json.dumps(data), headers=test_utils.TYPE_APPLICATION_JSON,
- auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.delete_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
def test_40_getLinks_OpenRoadmTopology(self):
- url = ("{}/config/ietf-network:networks/network/openroadm-topology"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/ietf-network:networks/network/openroadm-topology"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
nbLink = len(res['network'][0]['ietf-network-topology:link'])
response = test_utils.unmount_device("ROADM-A1")
self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
# Delete in the clli-network
- url = ("{}/config/ietf-network:networks/network/clli-network/node/NodeA"
- .format(test_utils.RESTCONF_BASE_URL))
+ url = "{}/config/ietf-network:networks/network/clli-network/node/NodeA"
data = {}
- response = requests.request(
- "DELETE", url, data=json.dumps(data), headers=test_utils.TYPE_APPLICATION_JSON,
- auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ response = test_utils.delete_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
def test_42_getClliNetwork(self):
- url = ("{}/config/ietf-network:networks/network/clli-network"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/ietf-network:networks/network/clli-network"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertNotIn('node', res['network'][0])
def test_43_getOpenRoadmNetwork(self):
- url = ("{}/config/ietf-network:networks/network/openroadm-network"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/ietf-network:networks/network/openroadm-network"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertNotIn('node', res['network'][0])
def test_44_check_roadm2roadm_link_persistence(self):
- url = ("{}/config/ietf-network:networks/network/openroadm-topology"
- .format(test_utils.RESTCONF_BASE_URL))
- response = requests.request(
- "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
+ url = "{}/config/ietf-network:networks/network/openroadm-topology"
+ response = test_utils.get_request(url)
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
nbLink = len(res['network'][0]['ietf-network-topology:link'])
universal_newlines=True)
-def post_request(url, data, username, password):
+def get_request(url):
return requests.request(
- "POST", url, data=json.dumps(data),
- headers=TYPE_APPLICATION_JSON, auth=(username, password))
+ "GET", url.format(RESTCONF_BASE_URL),
+ headers=TYPE_APPLICATION_JSON,
+ auth=(ODL_LOGIN, ODL_PWD))
+
+
+def post_request(url, data):
+ if data:
+ return requests.request(
+ "POST", url.format(RESTCONF_BASE_URL),
+ data=json.dumps(data),
+ headers=TYPE_APPLICATION_JSON,
+ auth=(ODL_LOGIN, ODL_PWD))
+ else:
+ return requests.request(
+ "POST", url.format(RESTCONF_BASE_URL),
+ headers=TYPE_APPLICATION_JSON,
+ auth=(ODL_LOGIN, ODL_PWD))
+
+
+def post_xmlrequest(url, data):
+ if data:
+ return requests.request(
+ "POST", url.format(RESTCONF_BASE_URL),
+ data=data,
+ headers=TYPE_APPLICATION_XML,
+ auth=(ODL_LOGIN, ODL_PWD))
+
+
+def put_request(url, data):
+ return requests.request(
+ "PUT", url.format(RESTCONF_BASE_URL),
+ data=json.dumps(data),
+ headers=TYPE_APPLICATION_JSON,
+ auth=(ODL_LOGIN, ODL_PWD))
-def put_request(url, data, username, password):
+def put_xmlrequest(url, data):
return requests.request(
- "PUT", url, data=json.dumps(data), headers=TYPE_APPLICATION_JSON,
- auth=(username, password))
+ "PUT", url.format(RESTCONF_BASE_URL),
+ data=data,
+ headers=TYPE_APPLICATION_XML,
+ auth=(ODL_LOGIN, ODL_PWD))
-def delete_request(url, username, password):
+def rawput_request(url, data):
return requests.request(
- "DELETE", url, headers=TYPE_APPLICATION_JSON,
- auth=(username, password))
+ "PUT", url.format(RESTCONF_BASE_URL),
+ data=data,
+ headers=TYPE_APPLICATION_JSON,
+ auth=(ODL_LOGIN, ODL_PWD))
+
+
+def delete_request(url):
+ return requests.request(
+ "DELETE", url.format(RESTCONF_BASE_URL),
+ headers=TYPE_APPLICATION_JSON,
+ auth=(ODL_LOGIN, ODL_PWD))
def mount_device(node_id, sim):
- url = ("{}/config/network-topology:network-topology/topology/topology-netconf/node/"
- + node_id).format(RESTCONF_BASE_URL)
- headers = {"node": [{
+ url = "{}/config/network-topology:network-topology/topology/topology-netconf/node/"+node_id
+ body = {"node": [{
"node-id": node_id,
"netconf-node-topology:username": NODES_LOGIN,
"netconf-node-topology:password": NODES_PWD,
"netconf-node-topology:port": SIMS[sim]['port'],
"netconf-node-topology:tcp-only": "false",
"netconf-node-topology:pass-through": {}}]}
- response = put_request(url, headers, ODL_LOGIN, ODL_PWD)
+ response = put_request(url, body)
if wait_until_log_contains(TPCE_LOG, re.escape("Triggering notification stream NETCONF for node "+node_id), 60):
print("Node "+node_id+" correctly added to tpce topology", end='... ', flush=True)
else:
def unmount_device(node_id):
- url = ("{}/config/network-topology:network-topology/topology/topology-netconf/node/"
- + node_id).format(RESTCONF_BASE_URL)
- response = delete_request(url, ODL_LOGIN, ODL_PWD)
+ url = "{}/config/network-topology:network-topology/topology/topology-netconf/node/"+node_id
+ response = delete_request(url)
if wait_until_log_contains(TPCE_LOG, re.escape("onDeviceDisConnected: "+node_id), 60):
print("Node "+node_id+" correctly deleted from tpce topology", end='... ', flush=True)
else:
return response
-def generate_link_data(xpdr_node: str, xpdr_num: str, network_num: str, rdm_node: str, srg_num: str,
- termination_num: str):
+def connect_xpdr_to_rdm_request(xpdr_node: str, xpdr_num: str, network_num: str,
+ rdm_node: str, srg_num: str, termination_num: str):
+ url = "{}/operations/transportpce-networkutils:init-xpdr-rdm-links"
+ data = {
+ "networkutils:input": {
+ "networkutils:links-input": {
+ "networkutils:xpdr-node": xpdr_node,
+ "networkutils:xpdr-num": xpdr_num,
+ "networkutils:network-num": network_num,
+ "networkutils:rdm-node": rdm_node,
+ "networkutils:srg-num": srg_num,
+ "networkutils:termination-point-num": termination_num
+ }
+ }
+ }
+ return post_request(url, data)
+
+
+def connect_rdm_to_xpdr_request(xpdr_node: str, xpdr_num: str, network_num: str,
+ rdm_node: str, srg_num: str, termination_num: str):
+ url = "{}/operations/transportpce-networkutils:init-rdm-xpdr-links"
data = {
"networkutils:input": {
"networkutils:links-input": {
}
}
}
- return data
+ return post_request(url, data)
def shutdown_process(process):
while True:
line = filelogs.readline()
if compiled_regexp.search(line):
- print("String found!", end=' ')
+ print("Pattern found!", end=' ')
stringfound = True
break
if not line:
time.sleep(0.1)
except TimeoutError:
- print("String not found after "+str(time_to_wait), end=" seconds! ", flush=True)
+ print("Pattern not found after "+str(time_to_wait), end=" seconds! ", flush=True)
except PermissionError:
print("Permission Error when trying to access the log file", end=" ... ", flush=True)
finally:
Code Review / transportpce.git / commitdiff