# 'connected')
# def test_02_restconfAPI(self):
-# 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))
+# response = test_utils.portmapping_request("controller-config")
# self.assertEqual(response.status_code, requests.codes.not_found)
# res = response.json()
# self.assertIn(
time.sleep(10)
def test_03_rdm_portmapping_info(self):
- url = ("{}/config/transportpce-portmapping:network/"
- "nodes/ROADMA01/node-info"
- )
- response = test_utils.get_request(url)
+ response = test_utils.portmapping_request("ROADMA01/node-info")
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/ROADMA01/mapping/DEG1-TTP-TXRX"
- )
- response = test_utils.get_request(url)
+ response = test_utils.portmapping_request("ROADMA01/mapping/DEG1-TTP-TXRX")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn(
res['mapping'])
def test_05_rdm_portmapping_SRG1_PP7_TXRX(self):
- url = ("{}/config/transportpce-portmapping:network/"
- "nodes/ROADMA01/mapping/SRG1-PP7-TXRX"
- )
- response = test_utils.get_request(url)
+ response = test_utils.portmapping_request("ROADMA01/mapping/SRG1-PP7-TXRX")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn(
res['mapping'])
def test_06_rdm_portmapping_SRG3_PP1_TXRX(self):
- url = ("{}/config/transportpce-portmapping:network/"
- "nodes/ROADMA01/mapping/SRG3-PP1-TXRX"
- )
- response = test_utils.get_request(url)
+ response = test_utils.portmapping_request("ROADMA01/mapping/SRG3-PP1-TXRX")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn(
time.sleep(10)
def test_09_xpdr_portmapping_info(self):
- url = ("{}/config/transportpce-portmapping:network/"
- "nodes/XPDRA01/node-info"
- )
- response = test_utils.get_request(url)
+ response = test_utils.portmapping_request("XPDRA01/node-info")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertEqual(
time.sleep(3)
def test_10_xpdr_portmapping_NETWORK1(self):
- url = ("{}/config/transportpce-portmapping:network/"
- "nodes/XPDRA01/mapping/XPDR1-NETWORK1"
- )
- response = test_utils.get_request(url)
+ response = test_utils.portmapping_request("XPDRA01/mapping/XPDR1-NETWORK1")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn(
res['mapping'])
def test_11_xpdr_portmapping_NETWORK2(self):
- url = ("{}/config/transportpce-portmapping:network/"
- "nodes/XPDRA01/mapping/XPDR1-NETWORK2"
- )
- response = test_utils.get_request(url)
+ response = test_utils.portmapping_request("XPDRA01/mapping/XPDR1-NETWORK2")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn(
res['mapping'])
def test_12_xpdr_portmapping_CLIENT1(self):
- url = ("{}/config/transportpce-portmapping:network/"
- "nodes/XPDRA01/mapping/XPDR1-CLIENT1"
- )
- response = test_utils.get_request(url)
+ response = test_utils.portmapping_request("XPDRA01/mapping/XPDR1-CLIENT1")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn(
res['mapping'])
def test_13_xpdr_portmapping_CLIENT2(self):
- url = ("{}/config/transportpce-portmapping:network/"
- "nodes/XPDRA01/mapping/XPDR1-CLIENT2"
- )
- response = test_utils.get_request(url)
+ response = test_utils.portmapping_request("XPDRA01/mapping/XPDR1-CLIENT2")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn(
res['mapping'])
def test_14_xpdr_portmapping_CLIENT3(self):
- url = ("{}/config/transportpce-portmapping:network/"
- "nodes/XPDRA01/mapping/XPDR1-CLIENT3"
- )
- response = test_utils.get_request(url)
+ response = test_utils.portmapping_request("XPDRA01/mapping/XPDR1-CLIENT3")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn(
'port-qual': 'xpdr-client', 'lcp-hash-val': 'AO9UFkY/TLYy'},
res['mapping'])
- def test_15_xpdr_portmapping_CLIENT4(self):
- url = ("{}/config/transportpce-portmapping:network/"
- "nodes/XPDRA01/mapping/XPDR1-CLIENT4"
- )
- response = test_utils.get_request(url)
+ response = test_utils.portmapping_request("XPDRA01/mapping/XPDR1-CLIENT4")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn(
res['errors']['error'])
def test_18_xpdr_device_not_connected(self):
- url = ("{}/config/transportpce-portmapping:network/nodes/XPDRA01")
- response = test_utils.get_request(url)
+ response = test_utils.portmapping_request("XPDRA01")
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")
- response = test_utils.get_request(url)
+ response = test_utils.portmapping_request("ROADMA01")
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/ROADMA01"
- )
- response = test_utils.get_request(url)
+ response = test_utils.portmapping_request("ROADMA01")
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/XPDRA01"
- )
- response = test_utils.get_request(url)
+ response = test_utils.portmapping_request("XPDRA01")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn(
for i in range(0, nbNode):
nodeId = resTopo['network'][0]['node'][i]['node-id']
nodeMapId = nodeId.split("-")[0]
- urlMapList = "{}/config/transportpce-portmapping:network/nodes/" + nodeMapId
- responseMapList = test_utils.get_request(urlMapList)
+ responseMapList = test_utils.portmapping_request(nodeMapId)
resMapList = responseMapList.json()
nbMappings = len(resMapList['nodes'][0]['mapping']) - nbMapCumul
nbTp = len(resTopo['network'][0]['node'][i]['ietf-network-topology:termination-point'])
for j in range(0, nbTp):
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
- responseMap = test_utils.get_request(urlMap)
+ responseMap = test_utils.portmapping_request(nodeMapId+"/mapping/"+tpId)
# Disconnect the ROADMA
def test_03_disconnect_rdm(self):
'connected')
def test_02_get_portmapping_CLIENT1(self):
- url = "{}/config/transportpce-portmapping:network/nodes/SPDR-SA1/mapping/XPDR1-CLIENT1"
- response = test_utils.get_request(url)
+ response = test_utils.portmapping_request("SPDR-SA1/mapping/XPDR1-CLIENT1")
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"
- response = test_utils.get_request(url)
+ response = test_utils.portmapping_request("SPDR-SA1/mapping/XPDR1-NETWORK1")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn(
'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"
- response = test_utils.get_request(url)
+ response = test_utils.portmapping_request("SPDR-SA1/mapping/XPDR1-NETWORK1")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn(
'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"
- response = test_utils.get_request(url)
+ response = test_utils.portmapping_request("SPDR-SA1/mapping/XPDR1-NETWORK1")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn(
time.sleep(10)
def test_03_rdm_portmapping_info(self):
- url = "{}/config/transportpce-portmapping:network/nodes/ROADM-A1/node-info"
- response = test_utils.get_request(url)
+ response = test_utils.portmapping_request("ROADM-A1/node-info")
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"
- response = test_utils.get_request(url)
+ response = test_utils.portmapping_request("ROADM-A1/mapping/DEG1-TTP-TXRX")
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"
- response = test_utils.get_request(url)
+ response = test_utils.portmapping_request("ROADM-A1/mapping/DEG2-TTP-TXRX")
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"
- response = test_utils.get_request(url)
+ response = test_utils.portmapping_request("ROADM-A1/mapping/SRG1-PP3-TXRX")
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"
- response = test_utils.get_request(url)
+ response = test_utils.portmapping_request("ROADM-A1/mapping/SRG3-PP1-TXRX")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn(
time.sleep(10)
def test_10_xpdr_portmapping_info(self):
- url = "{}/config/transportpce-portmapping:network/nodes/XPDR-A1/node-info"
- response = test_utils.get_request(url)
+ response = test_utils.portmapping_request("XPDR-A1/node-info")
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"
- response = test_utils.get_request(url)
+ response = test_utils.portmapping_request("XPDR-A1/mapping/XPDR1-NETWORK1")
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"
- response = test_utils.get_request(url)
+ response = test_utils.portmapping_request("XPDR-A1/mapping/XPDR1-NETWORK2")
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"
- response = test_utils.get_request(url)
+ response = test_utils.portmapping_request("XPDR-A1/mapping/XPDR1-CLIENT1")
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"
- response = test_utils.get_request(url)
+ response = test_utils.portmapping_request("XPDR-A1/mapping/XPDR1-CLIENT2")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn(
res['errors']['error'])
def test_17_xpdr_device_not_connected(self):
- url = "{}/config/transportpce-portmapping:network/nodes/XPDR-A1"
- response = test_utils.get_request(url)
+ response = test_utils.portmapping_request("XPDR-A1")
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"
- response = test_utils.get_request(url)
+ response = test_utils.portmapping_request("ROADM-A1")
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"
- response = test_utils.get_request(url)
+ response = test_utils.portmapping_request("ROADM-A1")
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"
- response = test_utils.get_request(url)
+ response = test_utils.portmapping_request("XPDR-A1")
self.assertEqual(response.status_code, requests.codes.ok)
res = response.json()
self.assertIn(
print("nodeId={}".format(nodeId))
nodeMapId = nodeId.split("-")[0] + "-" + nodeId.split("-")[1]
print("nodeMapId={}".format(nodeMapId))
- urlMapList = "{}/config/transportpce-portmapping:network/nodes/" + nodeMapId
- responseMapList = test_utils.get_request(urlMapList)
+ responseMapList = test_utils.portmapping_request(nodeMapId)
resMapList = responseMapList.json()
nbMappings = len(resMapList['nodes'][0]['mapping']) - nbMapCumul
for j in range(0, nbTp):
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
- responseMap = test_utils.get_request(urlMap)
+ responseMap = test_utils.portmapping_request(nodeMapId+"/mapping/"+tpId)
self.assertEqual(responseMap.status_code, requests.codes.ok)
if(responseMap.status_code == requests.codes.ok):
nbMapCurrent += 1
URL_CONFIG_OTN_TOPO = "{}/config/ietf-network:networks/network/otn-topology/"
URL_CONFIG_CLLI_NET = "{}/config/ietf-network:networks/network/clli-network/"
URL_CONFIG_ORDM_NET = "{}/config/ietf-network:networks/network/openroadm-network/"
+URL_PORTMAPPING = "{}/config/transportpce-portmapping:network/nodes/"
TYPE_APPLICATION_JSON = {'Content-Type': 'application/json', 'Accept': 'application/json'}
TYPE_APPLICATION_XML = {'Content-Type': 'application/xml', 'Accept': 'application/xml'}
return delete_request(url)
+def portmapping_request(suffix: str):
+ url = URL_PORTMAPPING + suffix
+ return get_request(url)
+
+
def shutdown_process(process):
if process is not None:
for child in psutil.Process(process.pid).children():