"fiber-type": "smf",
"SRLG-length": 100000,
"pmd": 0.5}]}}
- response = test_utils.add_oms_attr_request("ROADMA01-DEG1-DEG1-TTP-TXRXtoROADMC01-DEG2-DEG2-TTP-TXRX", data)
+ response = test_utils_rfc8040.add_oms_attr_request(
+ "ROADMA01-DEG1-DEG1-TTP-TXRXtoROADMC01-DEG2-DEG2-TTP-TXRX", data)
self.assertEqual(response.status_code, requests.codes.created)
def test_15_omsAttributes_ROADMC_ROADMA(self):
"fiber-type": "smf",
"SRLG-length": 100000,
"pmd": 0.5}]}}
- response = test_utils.add_oms_attr_request("ROADMC01-DEG2-DEG2-TTP-TXRXtoROADMA01-DEG1-DEG1-TTP-TXRX", data)
+ response = test_utils_rfc8040.add_oms_attr_request(
+ "ROADMC01-DEG2-DEG2-TTP-TXRXtoROADMA01-DEG1-DEG1-TTP-TXRX", data)
self.assertEqual(response.status_code, requests.codes.created)
def test_16_getClliNetwork(self):
"fiber-type": "smf",
"SRLG-length": 100000,
"pmd": 0.5}]}}
- response = test_utils.add_oms_attr_request("ROADMA01-DEG2-DEG2-TTP-TXRXtoROADMB01-DEG1-DEG1-TTP-TXRX", data)
+ response = test_utils_rfc8040.add_oms_attr_request(
+ "ROADMA01-DEG2-DEG2-TTP-TXRXtoROADMB01-DEG1-DEG1-TTP-TXRX", data)
self.assertEqual(response.status_code, requests.codes.created)
def test_23_omsAttributes_ROADMB_ROADMA(self):
"fiber-type": "smf",
"SRLG-length": 100000,
"pmd": 0.5}]}}
- response = test_utils.add_oms_attr_request("ROADMB01-DEG1-DEG1-TTP-TXRXtoROADMA01-DEG2-DEG2-TTP-TXRX", data)
+ response = test_utils_rfc8040.add_oms_attr_request(
+ "ROADMB01-DEG1-DEG1-TTP-TXRXtoROADMA01-DEG2-DEG2-TTP-TXRX", data)
self.assertEqual(response.status_code, requests.codes.created)
def test_24_omsAttributes_ROADMB_ROADMC(self):
"fiber-type": "smf",
"SRLG-length": 100000,
"pmd": 0.5}]}}
- response = test_utils.add_oms_attr_request("ROADMB01-DEG2-DEG2-TTP-TXRXtoROADMC01-DEG1-DEG1-TTP-TXRX", data)
+ response = test_utils_rfc8040.add_oms_attr_request(
+ "ROADMB01-DEG2-DEG2-TTP-TXRXtoROADMC01-DEG1-DEG1-TTP-TXRX", data)
self.assertEqual(response.status_code, requests.codes.created)
def test_25_omsAttributes_ROADMC_ROADMB(self):
"fiber-type": "smf",
"SRLG-length": 100000,
"pmd": 0.5}]}}
- response = test_utils.add_oms_attr_request("ROADMC01-DEG1-DEG1-TTP-TXRXtoROADMB01-DEG2-DEG2-TTP-TXRX", data)
+ response = test_utils_rfc8040.add_oms_attr_request(
+ "ROADMC01-DEG1-DEG1-TTP-TXRXtoROADMB01-DEG2-DEG2-TTP-TXRX", data)
self.assertEqual(response.status_code, requests.codes.created)
def test_26_getClliNetwork(self):
link_dest = val['destination']['dest-node']
oppLink_id = val['org-openroadm-common-network:opposite-link']
# Find the opposite link
- response_oppLink = test_utils.get_ordm_topo_request("ietf-network-topology:link/"+oppLink_id)
- self.assertEqual(response_oppLink.status_code, requests.codes.ok)
- res_oppLink = response_oppLink.json()
- self.assertEqual(res_oppLink['ietf-network-topology:link'][0]
- ['org-openroadm-common-network:opposite-link'], link_id)
- self.assertEqual(res_oppLink['ietf-network-topology:link'][0]['source']['source-node'], link_dest)
- self.assertEqual(res_oppLink['ietf-network-topology:link'][0]['destination']['dest-node'], link_src)
- oppLink_type = res_oppLink['ietf-network-topology:link'][0]['org-openroadm-common-network:link-type']
+ res_oppLink = test_utils_rfc8040.get_ietf_network_link_request('openroadm-topology', oppLink_id, 'config')
+ self.assertEqual(res_oppLink['status_code'], requests.codes.ok)
+ self.assertEqual(res_oppLink['link']['org-openroadm-common-network:opposite-link'], link_id)
+ self.assertEqual(res_oppLink['link']['source']['source-node'], link_dest)
+ self.assertEqual(res_oppLink['link']['destination']['dest-node'], link_src)
+ oppLink_type = res_oppLink['link']['org-openroadm-common-network:link-type']
CHECK_DICT = {'ADD-LINK': 'DROP-LINK', 'DROP-LINK': 'ADD-LINK',
'EXPRESS-LINK': 'EXPRESS-LINK', 'ROADM-TO-ROADM': 'ROADM-TO-ROADM',
'XPONDER-INPUT': 'XPONDER-OUTPUT', 'XPONDER-OUTUT': 'XPONDER-INPUT'}
response = test_utils_rfc8040.unmount_device("ROADMB01")
self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
# Delete in the clli-network
- response = test_utils.del_node_request("NodeB")
- self.assertEqual(response.status_code, requests.codes.ok)
+ response = test_utils_rfc8040.del_ietf_network_node_request('clli-network', 'NodeB', 'config')
+ self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
def test_31_disconnect_ROADMC(self):
response = test_utils_rfc8040.unmount_device("ROADMC01")
self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
# Delete in the clli-network
- response = test_utils.del_node_request("NodeC")
- self.assertEqual(response.status_code, requests.codes.ok)
+ response = test_utils_rfc8040.del_ietf_network_node_request('clli-network', 'NodeC', 'config')
+ self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
def test_32_getNodes_OpenRoadmTopology(self):
# pylint: disable=redundant-unittest-assert
def test_39_disconnect_ROADM_XPDRA_link(self):
# Link-1
- response = test_utils.del_link_request("XPDRA01-XPDR1-XPDR1-NETWORK1toROADMA01-SRG1-SRG1-PP1-TXRX")
- self.assertEqual(response.status_code, requests.codes.ok)
+ response = test_utils_rfc8040.del_ietf_network_link_request(
+ 'openroadm-topology',
+ 'XPDRA01-XPDR1-XPDR1-NETWORK1toROADMA01-SRG1-SRG1-PP1-TXRX',
+ 'config')
# Link-2
- response = test_utils.del_link_request("ROADMA01-SRG1-SRG1-PP1-TXRXtoXPDRA01-XPDR1-XPDR1-NETWORK1")
- self.assertEqual(response.status_code, requests.codes.ok)
+ response2 = test_utils_rfc8040.del_ietf_network_link_request(
+ 'openroadm-topology',
+ 'ROADMA01-SRG1-SRG1-PP1-TXRXtoXPDRA01-XPDR1-XPDR1-NETWORK1',
+ 'config')
+ self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
+ self.assertIn(response2.status_code, (requests.codes.ok, requests.codes.no_content))
def test_40_getLinks_OpenRoadmTopology(self):
response = test_utils_rfc8040.get_ietf_network_request('openroadm-topology', 'config')
response = test_utils_rfc8040.unmount_device("ROADMA01")
self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
# Delete in the clli-network
- response = test_utils.del_node_request("NodeA")
- self.assertEqual(response.status_code, requests.codes.ok)
+ response = test_utils_rfc8040.del_ietf_network_node_request('clli-network', 'NodeA', 'config')
+ self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
def test_42_getClliNetwork(self):
response = test_utils_rfc8040.get_ietf_network_request('clli-network', 'config')
"fiber-type": "smf",
"SRLG-length": 100000,
"pmd": 0.5}]}}
- response = test_utils.add_oms_attr_request("ROADM-A1-DEG2-DEG2-TTP-TXRXtoROADM-C1-DEG1-DEG1-TTP-TXRX", data)
+ response = test_utils_rfc8040.add_oms_attr_request(
+ "ROADM-A1-DEG2-DEG2-TTP-TXRXtoROADM-C1-DEG1-DEG1-TTP-TXRX", data)
self.assertEqual(response.status_code, requests.codes.created)
def test_15_omsAttributes_ROADMC_ROADMA(self):
"SRLG-length": 100000,
"pmd": 0.5}]}}
- response = test_utils.add_oms_attr_request("ROADM-C1-DEG1-DEG1-TTP-TXRXtoROADM-A1-DEG2-DEG2-TTP-TXRX", data)
+ response = test_utils_rfc8040.add_oms_attr_request(
+ "ROADM-C1-DEG1-DEG1-TTP-TXRXtoROADM-A1-DEG2-DEG2-TTP-TXRX", data)
self.assertEqual(response.status_code, requests.codes.created)
def test_16_getClliNetwork(self):
"fiber-type": "smf",
"SRLG-length": 100000,
"pmd": 0.5}]}}
- response = test_utils.add_oms_attr_request("ROADM-A1-DEG1-DEG1-TTP-TXRXtoROADM-B1-DEG1-DEG1-TTP-TXRX", data)
+ response = test_utils_rfc8040.add_oms_attr_request(
+ "ROADM-A1-DEG1-DEG1-TTP-TXRXtoROADM-B1-DEG1-DEG1-TTP-TXRX", data)
self.assertEqual(response.status_code, requests.codes.created)
def test_23_omsAttributes_ROADMB_ROADMA(self):
"fiber-type": "smf",
"SRLG-length": 100000,
"pmd": 0.5}]}}
- response = test_utils.add_oms_attr_request("ROADM-B1-DEG1-DEG1-TTP-TXRXtoROADM-A1-DEG1-DEG1-TTP-TXRX", data)
+ response = test_utils_rfc8040.add_oms_attr_request(
+ "ROADM-B1-DEG1-DEG1-TTP-TXRXtoROADM-A1-DEG1-DEG1-TTP-TXRX", data)
self.assertEqual(response.status_code, requests.codes.created)
def test_24_omsAttributes_ROADMB_ROADMC(self):
"fiber-type": "smf",
"SRLG-length": 100000,
"pmd": 0.5}]}}
- response = test_utils.add_oms_attr_request("ROADM-B1-DEG2-DEG2-TTP-TXRXtoROADM-C1-DEG2-DEG2-TTP-TXRX", data)
+ response = test_utils_rfc8040.add_oms_attr_request(
+ "ROADM-B1-DEG2-DEG2-TTP-TXRXtoROADM-C1-DEG2-DEG2-TTP-TXRX", data)
self.assertEqual(response.status_code, requests.codes.created)
def test_25_omsAttributes_ROADMC_ROADMB(self):
"fiber-type": "smf",
"SRLG-length": 100000,
"pmd": 0.5}]}}
- response = test_utils.add_oms_attr_request("ROADM-C1-DEG2-DEG2-TTP-TXRXtoROADM-B1-DEG2-DEG2-TTP-TXRX", data)
+ response = test_utils_rfc8040.add_oms_attr_request(
+ "ROADM-C1-DEG2-DEG2-TTP-TXRXtoROADM-B1-DEG2-DEG2-TTP-TXRX", data)
self.assertEqual(response.status_code, requests.codes.created)
def test_26_getClliNetwork(self):
link_dest = val['destination']['dest-node']
oppLink_id = val['org-openroadm-common-network:opposite-link']
# Find the opposite link
- response_oppLink = test_utils.get_ordm_topo_request("ietf-network-topology:link/"+oppLink_id)
- self.assertEqual(response_oppLink.status_code, requests.codes.ok)
- res_oppLink = response_oppLink.json()
- self.assertEqual(res_oppLink['ietf-network-topology:link'][0]
- ['org-openroadm-common-network:opposite-link'], link_id)
- self.assertEqual(res_oppLink['ietf-network-topology:link'][0]['source']['source-node'], link_dest)
- self.assertEqual(res_oppLink['ietf-network-topology:link'][0]['destination']['dest-node'], link_src)
- oppLink_type = res_oppLink['ietf-network-topology:link'][0]['org-openroadm-common-network:link-type']
+ res_oppLink = test_utils_rfc8040.get_ietf_network_link_request('openroadm-topology', oppLink_id, 'config')
+ self.assertEqual(res_oppLink['status_code'], requests.codes.ok)
+ self.assertEqual(res_oppLink['link']['org-openroadm-common-network:opposite-link'], link_id)
+ self.assertEqual(res_oppLink['link']['source']['source-node'], link_dest)
+ self.assertEqual(res_oppLink['link']['destination']['dest-node'], link_src)
+ oppLink_type = res_oppLink['link']['org-openroadm-common-network:link-type']
CHECK_DICT = {'ADD-LINK': 'DROP-LINK', 'DROP-LINK': 'ADD-LINK',
'EXPRESS-LINK': 'EXPRESS-LINK', 'ROADM-TO-ROADM': 'ROADM-TO-ROADM',
'XPONDER-INPUT': 'XPONDER-OUTPUT', 'XPONDER-OUTUT': 'XPONDER-INPUT'}
response = test_utils_rfc8040.unmount_device("ROADM-B1")
self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
# Delete in the clli-network
- response = test_utils.del_node_request("NodeB")
- self.assertEqual(response.status_code, requests.codes.ok)
+ response = test_utils_rfc8040.del_ietf_network_node_request('clli-network', 'NodeB', 'config')
+ self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
def test_31_disconnect_ROADMC(self):
# Delete in the topology-netconf
response = test_utils_rfc8040.unmount_device("ROADM-C1")
self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
# Delete in the clli-network
- response = test_utils.del_node_request("NodeC")
- self.assertEqual(response.status_code, requests.codes.ok)
+ response = test_utils_rfc8040.del_ietf_network_node_request('clli-network', 'NodeC', 'config')
+ self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
def test_32_getNodes_OpenRoadmTopology(self):
# pylint: disable=redundant-unittest-assert
def test_39_disconnect_ROADM_XPDRA_link(self):
# Link-1
- response = test_utils.del_link_request("XPDR-A1-XPDR1-XPDR1-NETWORK1toROADM-A1-SRG1-SRG1-PP1-TXRX")
- self.assertEqual(response.status_code, requests.codes.ok)
+ response = test_utils_rfc8040.del_ietf_network_link_request(
+ 'openroadm-topology',
+ 'XPDR-A1-XPDR1-XPDR1-NETWORK1toROADM-A1-SRG1-SRG1-PP1-TXRX',
+ 'config')
# Link-2
- response = test_utils.del_link_request("ROADM-A1-SRG1-SRG1-PP1-TXRXtoXPDR-A1-XPDR1-XPDR1-NETWORK1")
- self.assertEqual(response.status_code, requests.codes.ok)
+ response2 = test_utils_rfc8040.del_ietf_network_link_request(
+ 'openroadm-topology',
+ 'ROADM-A1-SRG1-SRG1-PP1-TXRXtoXPDR-A1-XPDR1-XPDR1-NETWORK1',
+ 'config')
+ self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
+ self.assertIn(response2.status_code, (requests.codes.ok, requests.codes.no_content))
def test_40_getLinks_OpenRoadmTopology(self):
response = test_utils_rfc8040.get_ietf_network_request('openroadm-topology', 'config')
response = test_utils_rfc8040.unmount_device("ROADM-A1")
self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
# Delete in the clli-network
- response = test_utils.del_node_request("NodeA")
- self.assertEqual(response.status_code, requests.codes.ok)
+ response = test_utils_rfc8040.del_ietf_network_node_request('clli-network', 'NodeA', 'config')
+ self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
def test_42_getClliNetwork(self):
response = test_utils_rfc8040.get_ietf_network_request('clli-network', 'config')
'draft-bierman02': '/restconf'}
if 'USE_ODL_RESTCONF_VERSION' in os.environ:
RESTCONF_VERSION = os.environ['USE_ODL_RESTCONF_VERSION']
- if RESTCONF_VERSION not in RESTCONF_PATH_PREFIX.keys():
+ if RESTCONF_VERSION not in RESTCONF_PATH_PREFIX:
print('unsupported RESTCONF version ' + RESTCONF_VERSION)
sys.exit(3)
else:
networks = res[return_key[RESTCONF_VERSION]]
return {'status_code': response.status_code,
'network': networks}
+
+
+def get_ietf_network_link_request(network: str, link: str, content: str):
+ url = {'rfc8040': '{}/data/ietf-network:networks/network={}/ietf-network-topology:link={}?content={}',
+ 'draft-bierman02': '{}/{}/ietf-network:networks/network/{}/ietf-network-topology:link/{}'}
+ if RESTCONF_VERSION == 'rfc8040':
+ format_args = ('{}', network, link, content)
+ elif content == 'config':
+ format_args = ('{}', content, network, link)
+ else:
+ format_args = ('{}', 'operational', network, link)
+ response = get_request(url[RESTCONF_VERSION].format(*format_args))
+ res = response.json()
+ return_key = {'rfc8040': 'ietf-network-topology:link',
+ 'draft-bierman02': 'ietf-network-topology:link'}
+ link = res[return_key[RESTCONF_VERSION]][0]
+ return {'status_code': response.status_code,
+ 'link': link}
+
+
+def del_ietf_network_link_request(network: str, link: str, content: str):
+ url = {'rfc8040': '{}/data/ietf-network:networks/network={}/ietf-network-topology:link={}?content={}',
+ 'draft-bierman02': '{}/{}/ietf-network:networks/network/{}/ietf-network-topology:link/{}'}
+ if RESTCONF_VERSION == 'rfc8040':
+ format_args = ('{}', network, link, content)
+ elif content == 'config':
+ format_args = ('{}', content, network, link)
+ else:
+ format_args = ('{}', 'operational', network, link)
+ response = delete_request(url[RESTCONF_VERSION].format(*format_args))
+ return response
+
+
+def add_oms_attr_request(link: str, oms_attr: str):
+ url = {'rfc8040': '{}/data/ietf-network:networks/network={}/ietf-network-topology:link={}',
+ 'draft-bierman02': '{}/config/ietf-network:networks/network/{}/ietf-network-topology:link/{}'}
+ url2 = url[RESTCONF_VERSION] + '/org-openroadm-network-topology:OMS-attributes/span'
+ network = 'openroadm-topology'
+ response = put_request(url2.format('{}', network, link), oms_attr)
+ return response
+
+
+def del_oms_attr_request(link: str,):
+ url = {'rfc8040': '{}/data/ietf-network:networks/network={}/ietf-network-topology:link={}',
+ 'draft-bierman02': '{}/config/ietf-network:networks/network/{}/ietf-network-topology:link/{}'}
+ url2 = url[RESTCONF_VERSION] + '/org-openroadm-network-topology:OMS-attributes/span'
+ network = 'openroadm-topology'
+ response = delete_request(url2.format('{}', network, link))
+ return response
+
+
+def del_ietf_network_node_request(network: str, node: str, content: str):
+ url = {'rfc8040': '{}/data/ietf-network:networks/network={}/node={}?content={}',
+ 'draft-bierman02': '{}/{}/ietf-network:networks/network/{}/node/{}'}
+ if RESTCONF_VERSION == 'rfc8040':
+ format_args = ('{}', network, node, content)
+ elif content == 'config':
+ format_args = ('{}', content, network, node)
+ else:
+ format_args = ('{}', 'operational', network, node)
+ response = delete_request(url[RESTCONF_VERSION].format(*format_args))
+ return response
passenv = LAUNCHER USE_LIGHTY USE_ODL_RESTCONF_VERSION
setenv =
# USE_LIGHTY=True
- USE_ODL_RESTCONF_VERSION=draft-bierman02
+# USE_ODL_RESTCONF_VERSION=draft-bierman02
USE_ODL_ALT_KARAF_ENV=./karaf121.env
USE_ODL_ALT_KARAF_INSTALL_DIR=karaf121
commands =
passenv = LAUNCHER USE_LIGHTY USE_ODL_RESTCONF_VERSION
setenv =
# USE_LIGHTY=True
- USE_ODL_RESTCONF_VERSION=draft-bierman02
+# USE_ODL_RESTCONF_VERSION=draft-bierman02
USE_ODL_ALT_KARAF_ENV=./karaf221.env
USE_ODL_ALT_KARAF_INSTALL_DIR=karaf221
commands =
passenv = LAUNCHER USE_LIGHTY USE_ODL_RESTCONF_VERSION
setenv =
# USE_LIGHTY=True
- USE_ODL_RESTCONF_VERSION=draft-bierman02
+# USE_ODL_RESTCONF_VERSION=draft-bierman02
USE_ODL_ALT_KARAF_ENV=./karaf71.env
USE_ODL_ALT_KARAF_INSTALL_DIR=karaf71
commands =