X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=tests%2Ftransportpce_tests%2F1.2.1%2Ftest_pce.py;h=f97dda23adb6159908643770b7da8f828808a17a;hb=77fcc6a019ffb351ee6b5d13757cefecdd837a30;hp=3b3cf13e6c02230cd2c2703add4487421d83b97c;hpb=1b290f8e8a33836fcf5960ebbd19ae5afe4b17ce;p=transportpce.git diff --git a/tests/transportpce_tests/1.2.1/test_pce.py b/tests/transportpce_tests/1.2.1/test_pce.py index 3b3cf13e6..f97dda23a 100644 --- a/tests/transportpce_tests/1.2.1/test_pce.py +++ b/tests/transportpce_tests/1.2.1/test_pce.py @@ -33,23 +33,23 @@ class TransportPCEtesting(unittest.TestCase): topo_bi_dir_file = "sample_configs/honeynode-topo.xml" if os.path.isfile(topo_bi_dir_file): with open(topo_bi_dir_file, 'r') as topo_bi_dir: - cls.simple_topo_bi_dir_data = topo_bi_dir.read(); + cls.simple_topo_bi_dir_data = topo_bi_dir.read() topo_uni_dir_file = "sample_configs/NW-simple-topology.xml" if os.path.isfile(topo_uni_dir_file): with open(topo_uni_dir_file, 'r') as topo_uni_dir: - cls.simple_topo_uni_dir_data = topo_uni_dir.read(); + cls.simple_topo_uni_dir_data = topo_uni_dir.read() topo_uni_dir_complex_file = "sample_configs/NW-for-test-5-4.xml" if os.path.isfile(topo_uni_dir_complex_file): with open(topo_uni_dir_complex_file, 'r') as topo_uni_dir_complex: - cls.complex_topo_uni_dir_data = topo_uni_dir_complex.read(); + cls.complex_topo_uni_dir_data = topo_uni_dir_complex.read() @classmethod def setUpClass(cls): # a class method called before tests in an individual class run. cls._get_file() - print ("starting opendaylight...") + print("starting opendaylight...") cls.odl_process = test_utils.start_tpce() time.sleep(90) - print ("opendaylight started") + print("opendaylight started") @classmethod def tearDownClass(cls): @@ -65,10 +65,10 @@ class TransportPCEtesting(unittest.TestCase): # Load simple bidirectional topology def test_01_load_simple_topology_bi(self): url = ("{}/config/ietf-network:networks/network/openroadm-topology" - .format(self.restconf_baseurl)) + .format(self.restconf_baseurl)) body = self.simple_topo_bi_dir_data headers = {'content-type': 'application/xml', - "Accept": "application/xml"} + "Accept": "application/xml"} response = requests.request( "PUT", url, data=body, headers=headers, auth=('admin', 'admin')) @@ -77,37 +77,37 @@ class TransportPCEtesting(unittest.TestCase): # Get existing nodeId def test_02_get_nodeId(self): - url = ("{}/config/ietf-network:networks/network/openroadm-topology/node/ROADMA-SRG1" - .format(self.restconf_baseurl)) + url = ("{}/config/ietf-network:networks/network/openroadm-topology/node/ROADMA01-SRG1" + .format(self.restconf_baseurl)) headers = {'content-type': 'application/json', - "Accept": "application/json"} + "Accept": "application/json"} response = requests.request( "GET", url, headers=headers, auth=('admin', 'admin')) self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertEqual( - res['node'][0]['node-id'], 'ROADMA-SRG1') + res['node'][0]['node-id'], 'ROADMA01-SRG1') time.sleep(1) # Get existing linkId def test_03_get_linkId(self): - url = ("{}/config/ietf-network:networks/network/openroadm-topology/link/XPDRA-XPDR1-XPDR1-NETWORK1toROADMA-SRG1-SRG1-PP1-TXRX" - .format(self.restconf_baseurl)) + url = ("{}/config/ietf-network:networks/network/openroadm-topology/link/XPDRA01-XPDR1-XPDR1-NETWORK1toROADMA01-SRG1-SRG1-PP1-TXRX" + .format(self.restconf_baseurl)) headers = {'content-type': 'application/json', - "Accept": "application/json"} + "Accept": "application/json"} response = requests.request( "GET", url, headers=headers, auth=('admin', 'admin')) self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertEqual( res['ietf-network-topology:link'][0]['link-id'], - 'XPDRA-XPDR1-XPDR1-NETWORK1toROADMA-SRG1-SRG1-PP1-TXRX') + 'XPDRA01-XPDR1-XPDR1-NETWORK1toROADMA01-SRG1-SRG1-PP1-TXRX') time.sleep(1) # Path Computation success def test_04_path_computation_xpdr_bi(self): url = ("{}/operations/transportpce-pce:path-computation-request" - .format(self.restconf_baseurl)) + .format(self.restconf_baseurl)) body = {"input": { "service-name": "service-1", "resource-reserve": "true", @@ -116,32 +116,34 @@ class TransportPCEtesting(unittest.TestCase): "request-id": "request-1" }, "service-a-end": { - "node-id": "XPDRA", - "service-rate": "0", + "node-id": "XPDRA01", + "service-rate": "100", + "service-format": "Ethernet", "clli": "nodeA" }, "service-z-end": { - "node-id": "XPDRC", - "service-rate": "0", + "node-id": "XPDRC01", + "service-rate": "100", + "service-format": "Ethernet", "clli": "nodeC" } - } - } + } + } headers = {'content-type': 'application/json', - "Accept": "application/json"} + "Accept": "application/json"} response = requests.request( "POST", url, data=json.dumps(body), headers=headers, auth=('admin', 'admin')) self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertIn('Path is calculated', - res['output']['configuration-response-common']['response-message']) + res['output']['configuration-response-common']['response-message']) time.sleep(5) # Path Computation success def test_05_path_computation_rdm_bi(self): url = ("{}/operations/transportpce-pce:path-computation-request" - .format(self.restconf_baseurl)) + .format(self.restconf_baseurl)) body = {"input": { "service-name": "service-1", "resource-reserve": "true", @@ -150,34 +152,36 @@ class TransportPCEtesting(unittest.TestCase): "request-id": "request-1" }, "service-a-end": { - "node-id": "ROADMA", - "service-rate": "0", - "clli": "nodeA" + "node-id": "ROADMA01", + "service-rate": "100", + "service-format": "Ethernet", + "clli": "NodeA" }, "service-z-end": { - "node-id": "ROADMC", - "service-rate": "0", - "clli": "nodeC" + "node-id": "ROADMC01", + "service-rate": "100", + "service-format": "Ethernet", + "clli": "NodeC" + } + } } - } - } headers = {'content-type': 'application/json', - "Accept": "application/json"} + "Accept": "application/json"} response = requests.request( "POST", url, data=json.dumps(body), headers=headers, auth=('admin', 'admin')) self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertIn('Path is calculated', - res['output']['configuration-response-common']['response-message']) + res['output']['configuration-response-common']['response-message']) time.sleep(5) # Delete topology def test_06_delete_simple_topology_bi(self): url = ("{}/config/ietf-network:networks/network/openroadm-topology" - .format(self.restconf_baseurl)) + .format(self.restconf_baseurl)) headers = {'content-type': 'application/xml', - "Accept": "application/json"} + "Accept": "application/json"} response = requests.request( "DELETE", url, headers=headers, auth=('admin', 'admin')) self.assertEqual(response.status_code, requests.codes.ok) @@ -185,10 +189,10 @@ class TransportPCEtesting(unittest.TestCase): # Test deleted topology def test_07_test_topology_simple_bi_deleted(self): - url = ("{}/config/ietf-network:networks/network/openroadm-topology/node/ROADMA-SRG1" - .format(self.restconf_baseurl)) + url = ("{}/config/ietf-network:networks/network/openroadm-topology/node/ROADMA01-SRG1" + .format(self.restconf_baseurl)) headers = {'content-type': 'application/json', - "Accept": "application/json"} + "Accept": "application/json"} response = requests.request( "GET", url, headers=headers, auth=('admin', 'admin')) self.assertEqual(response.status_code, 404) @@ -197,10 +201,10 @@ class TransportPCEtesting(unittest.TestCase): # Load simple bidirectional topology def test_08_load_simple_topology_uni(self): url = ("{}/config/ietf-network:networks/network/openroadm-topology" - .format(self.restconf_baseurl)) + .format(self.restconf_baseurl)) body = self.simple_topo_uni_dir_data headers = {'content-type': 'application/xml', - "Accept": "application/xml"} + "Accept": "application/xml"} response = requests.request( "PUT", url, data=body, headers=headers, auth=('admin', 'admin')) @@ -210,9 +214,9 @@ class TransportPCEtesting(unittest.TestCase): # Get existing nodeId def test_09_get_nodeId(self): url = ("{}/config/ietf-network:networks/network/openroadm-topology/node/XPONDER-1-2" - .format(self.restconf_baseurl)) + .format(self.restconf_baseurl)) headers = {'content-type': 'application/json', - "Accept": "application/json"} + "Accept": "application/json"} response = requests.request( "GET", url, headers=headers, auth=('admin', 'admin')) self.assertEqual(response.status_code, requests.codes.ok) @@ -225,9 +229,9 @@ class TransportPCEtesting(unittest.TestCase): # 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(self.restconf_baseurl)) + .format(self.restconf_baseurl)) headers = {'content-type': 'application/json', - "Accept": "application/json"} + "Accept": "application/json"} response = requests.request( "GET", url, headers=headers, auth=('admin', 'admin')) self.assertEqual(response.status_code, requests.codes.ok) @@ -240,7 +244,7 @@ class TransportPCEtesting(unittest.TestCase): # Path Computation success def test_11_path_computation_xpdr_uni(self): url = ("{}/operations/transportpce-pce:path-computation-request" - .format(self.restconf_baseurl)) + .format(self.restconf_baseurl)) body = {"input": { "service-name": "service-1", "resource-reserve": "true", @@ -250,65 +254,69 @@ class TransportPCEtesting(unittest.TestCase): }, "service-a-end": { "node-id": "XPONDER-1-2", - "service-rate": "0", + "service-rate": "100", + "service-format": "Ethernet", "clli": "ORANGE1" }, "service-z-end": { "node-id": "XPONDER-3-2", - "service-rate": "0", - "clli": "ORANGE1" + "service-rate": "100", + "service-format": "Ethernet", + "clli": "ORANGE3" + } + } } - } - } headers = {'content-type': 'application/json', - "Accept": "application/json"} + "Accept": "application/json"} response = requests.request( "POST", url, data=json.dumps(body), headers=headers, auth=('admin', 'admin')) self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertIn('Path is calculated', - res['output']['configuration-response-common']['response-message']) + res['output']['configuration-response-common']['response-message']) time.sleep(5) # Path Computation success def test_12_path_computation_rdm_uni(self): url = ("{}/operations/transportpce-pce:path-computation-request" - .format(self.restconf_baseurl)) + .format(self.restconf_baseurl)) body = {"input": { "service-name": "service1", "resource-reserve": "true", "service-handler-header": { - "request-id": "request1" + "request-id": "request1" }, "service-a-end": { - "service-rate": "0", - "clli": "cll21", - "node-id": "OpenROADM-2-1" + "service-rate": "100", + "service-format": "Ethernet", + "clli": "cll21", + "node-id": "OpenROADM-2-1" }, "service-z-end": { - "service-rate": "0", - "clli": "ncli22", - "node-id": "OpenROADM-2-2" - }, + "service-rate": "100", + "service-format": "Ethernet", + "clli": "ncli22", + "node-id": "OpenROADM-2-2" + }, "pce-metric": "hop-count" - } - } + } + } headers = {'content-type': 'application/json', - "Accept": "application/json"} + "Accept": "application/json"} response = requests.request( "POST", url, data=json.dumps(body), headers=headers, auth=('admin', 'admin')) self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertIn('Path is calculated', - res['output']['configuration-response-common']['response-message']) - #ZtoA path test + res['output']['configuration-response-common']['response-message']) + # ZtoA path test atozList = len(res['output']['response-parameters']['path-description']['aToZ-direction']['aToZ']) ztoaList = len(res['output']['response-parameters']['path-description']['zToA-direction']['zToA']) - self.assertEqual(atozList,15) - self.assertEqual(ztoaList,15) - for i in range(0,15): + self.assertEqual(atozList, 15) + self.assertEqual(ztoaList, 15) + for i in range(0, 15): atoz = res['output']['response-parameters']['path-description']['aToZ-direction']['aToZ'][i] ztoa = res['output']['response-parameters']['path-description']['zToA-direction']['zToA'][i] if (atoz['id'] == '14'): @@ -320,9 +328,9 @@ class TransportPCEtesting(unittest.TestCase): # Delete topology def test_13_delete_simple_topology(self): url = ("{}/config/ietf-network:networks/network/openroadm-topology" - .format(self.restconf_baseurl)) + .format(self.restconf_baseurl)) headers = {'content-type': 'application/xml', - "Accept": "application/json"} + "Accept": "application/json"} response = requests.request( "DELETE", url, headers=headers, auth=('admin', 'admin')) self.assertEqual(response.status_code, requests.codes.ok) @@ -331,21 +339,21 @@ class TransportPCEtesting(unittest.TestCase): # Test deleted topology def test_14_test_topology_simple_deleted(self): url = ("{}/config/ietf-network:networks/network/openroadm-topology/node/XPONDER-1-2" - .format(self.restconf_baseurl)) + .format(self.restconf_baseurl)) headers = {'content-type': 'application/json', - "Accept": "application/json"} + "Accept": "application/json"} response = requests.request( "GET", url, headers=headers, auth=('admin', 'admin')) self.assertEqual(response.status_code, 404) time.sleep(1) - # Load simple topology + # Load complex topology def test_15_load_complex_topology(self): url = ("{}/config/ietf-network:networks/network/openroadm-topology" - .format(self.restconf_baseurl)) + .format(self.restconf_baseurl)) body = self.complex_topo_uni_dir_data headers = {'content-type': 'application/xml', - "Accept": "application/json"} + "Accept": "application/json"} response = requests.request( "PUT", url, data=body, headers=headers, auth=('admin', 'admin')) @@ -355,9 +363,9 @@ class TransportPCEtesting(unittest.TestCase): # Get existing nodeId def test_16_get_nodeId(self): url = ("{}/config/ietf-network:networks/network/openroadm-topology/node/XPONDER-3-2" - .format(self.restconf_baseurl)) + .format(self.restconf_baseurl)) headers = {'content-type': 'application/json', - "Accept": "application/json"} + "Accept": "application/json"} response = requests.request( "GET", url, headers=headers, auth=('admin', 'admin')) self.assertEqual(response.status_code, requests.codes.ok) @@ -370,129 +378,129 @@ class TransportPCEtesting(unittest.TestCase): # Test failed path computation def test_17_fail_path_computation(self): url = ("{}/operations/transportpce-pce:path-computation-request" - .format(self.restconf_baseurl)) + .format(self.restconf_baseurl)) body = {"input": { "service-handler-header": { "request-id": "request-1" } - } - } + } + } headers = {'content-type': 'application/json', - "Accept": "application/json"} + "Accept": "application/json"} response = requests.request( "POST", url, data=json.dumps(body), headers=headers, auth=('admin', 'admin')) self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertIn('Service Name is not set', - res['output']['configuration-response-common']['response-message']) + res['output']['configuration-response-common']['response-message']) time.sleep(2) # Test1 success path computation def test_18_success1_path_computation(self): url = ("{}/operations/transportpce-pce:path-computation-request" - .format(self.restconf_baseurl)) + .format(self.restconf_baseurl)) body = {"input": { "service-name": "service1", "resource-reserve": "true", "service-handler-header": { - "request-id": "request1" + "request-id": "request1" }, "service-a-end": { - "service-format": "Ethernet", - "service-rate": "0", - "clli": "clli11", - "node-id": "XPONDER-2-2", - "tx-direction": { - "port": { - "port-device-name": "Some port-device-name", - "port-type": "Some port-type", - "port-name": "Some port-name", - "port-rack": "Some port-rack", - "port-shelf": "Some port-shelf", - "port-slot": "Some port-slot", - "port-sub-slot": "Some port-sub-slot" - } - }, - "rx-direction": { - "port": { - "port-device-name": "Some port-device-name", - "port-type": "Some port-type", - "port-name": "Some port-name", - "port-rack": "Some port-rack", - "port-shelf": "Some port-shelf", - "port-slot": "Some port-slot", - "port-sub-slot": "Some port-sub-slot" + "service-format": "Ethernet", + "service-rate": "100", + "clli": "ORANGE2", + "node-id": "XPONDER-2-2", + "tx-direction": { + "port": { + "port-device-name": "Some port-device-name", + "port-type": "Some port-type", + "port-name": "Some port-name", + "port-rack": "Some port-rack", + "port-shelf": "Some port-shelf", + "port-slot": "Some port-slot", + "port-sub-slot": "Some port-sub-slot" + } + }, + "rx-direction": { + "port": { + "port-device-name": "Some port-device-name", + "port-type": "Some port-type", + "port-name": "Some port-name", + "port-rack": "Some port-rack", + "port-shelf": "Some port-shelf", + "port-slot": "Some port-slot", + "port-sub-slot": "Some port-sub-slot" + } } - } }, "service-z-end": { - "service-format": "Ethernet", - "service-rate": "0", - "clli": "clli11", - "node-id": "XPONDER-1-2", - "tx-direction": { - "port": { - "port-device-name": "Some port-device-name", - "port-type": "Some port-type", - "port-name": "Some port-name", - "port-rack": "Some port-rack", - "port-shelf": "Some port-shelf", - "port-slot": "Some port-slot", - "port-sub-slot": "Some port-sub-slot" - } - }, - "rx-direction": { - "port": { - "port-device-name": "Some port-device-name", - "port-type": "Some port-type", - "port-name": "Some port-name", - "port-rack": "Some port-rack", - "port-shelf": "Some port-shelf", - "port-slot": "Some port-slot", - "port-sub-slot": "Some port-sub-slot" + "service-format": "Ethernet", + "service-rate": "100", + "clli": "ORANGE1", + "node-id": "XPONDER-1-2", + "tx-direction": { + "port": { + "port-device-name": "Some port-device-name", + "port-type": "Some port-type", + "port-name": "Some port-name", + "port-rack": "Some port-rack", + "port-shelf": "Some port-shelf", + "port-slot": "Some port-slot", + "port-sub-slot": "Some port-sub-slot" + } + }, + "rx-direction": { + "port": { + "port-device-name": "Some port-device-name", + "port-type": "Some port-type", + "port-name": "Some port-name", + "port-rack": "Some port-rack", + "port-shelf": "Some port-shelf", + "port-slot": "Some port-slot", + "port-sub-slot": "Some port-sub-slot" + } } - } }, "hard-constraints": { - "customer-code": [ - "Some customer-code" - ], - "co-routing": { - "existing-service": [ - "Some existing-service" - ] - } + "customer-code": [ + "Some customer-code" + ], + "co-routing": { + "existing-service": [ + "Some existing-service" + ] + } }, "soft-constraints": { - "customer-code": [ - "Some customer-code" - ], - "co-routing": { - "existing-service": [ - "Some existing-service" - ] - } + "customer-code": [ + "Some customer-code" + ], + "co-routing": { + "existing-service": [ + "Some existing-service" + ] + } }, "pce-metric": "hop-count", "locally-protected-links": "true" - } - } + } + } headers = {'content-type': 'application/json', - "Accept": "application/json"} + "Accept": "application/json"} response = requests.request( "POST", url, data=json.dumps(body), headers=headers, auth=('admin', 'admin')) self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertIn('Path is calculated', - res['output']['configuration-response-common']['response-message']) + res['output']['configuration-response-common']['response-message']) time.sleep(5) # Test2 success path computation with path description def test_19_success2_path_computation(self): url = ("{}/operations/transportpce-pce:path-computation-request" - .format(self.restconf_baseurl)) + .format(self.restconf_baseurl)) body = {"input": { "service-name": "service 1", "resource-reserve": "true", @@ -500,35 +508,39 @@ class TransportPCEtesting(unittest.TestCase): "request-id": "request 1" }, "service-a-end": { - "service-rate": "0", - "node-id": "XPONDER-1-2" + "service-rate": "100", + "service-format": "Ethernet", + "node-id": "XPONDER-1-2", + "clli": "ORANGE1" }, "service-z-end": { - "service-rate": "0", - "node-id": "XPONDER-3-2" - }, - "pce-metric": "hop-count" - } - } + "service-rate": "100", + "service-format": "Ethernet", + "node-id": "XPONDER-3-2", + "clli": "ORANGE3" + }, + "pce-metric": "hop-count" + } + } headers = {'content-type': 'application/json', - "Accept": "application/json"} + "Accept": "application/json"} response = requests.request( "POST", url, data=json.dumps(body), headers=headers, auth=('admin', 'admin')) self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertIn('Path is calculated', - res['output']['configuration-response-common']['response-message']) - self.assertEqual(5 , res['output']['response-parameters']['path-description'] - ['aToZ-direction']['aToZ-wavelength-number']) - self.assertEqual(5 , res['output']['response-parameters']['path-description'] - ['zToA-direction']['zToA-wavelength-number']) + res['output']['configuration-response-common']['response-message']) + self.assertEqual(5, res['output']['response-parameters']['path-description'] + ['aToZ-direction']['aToZ-wavelength-number']) + self.assertEqual(5, res['output']['response-parameters']['path-description'] + ['zToA-direction']['zToA-wavelength-number']) time.sleep(5) # Test3 success path computation with hard-constraints exclude def test_20_success3_path_computation(self): url = ("{}/operations/transportpce-pce:path-computation-request" - .format(self.restconf_baseurl)) + .format(self.restconf_baseurl)) body = {"input": { "service-name": "service 1", "resource-reserve": "true", @@ -536,12 +548,16 @@ class TransportPCEtesting(unittest.TestCase): "request-id": "request 1" }, "service-a-end": { - "service-rate": "0", - "node-id": "XPONDER-1-2" + "service-rate": "100", + "service-format": "Ethernet", + "node-id": "XPONDER-1-2", + "clli": "ORANGE1" }, "service-z-end": { - "service-rate": "0", - "node-id": "XPONDER-3-2" + "service-rate": "100", + "service-format": "Ethernet", + "node-id": "XPONDER-3-2", + "clli": "ORANGE3" }, "hard-constraints": { "exclude_": { @@ -549,40 +565,144 @@ class TransportPCEtesting(unittest.TestCase): } }, "pce-metric": "hop-count" - } - } + } + } + headers = {'content-type': 'application/json', + "Accept": "application/json"} + response = requests.request( + "POST", url, data=json.dumps(body), headers=headers, + auth=('admin', 'admin')) + self.assertEqual(response.status_code, requests.codes.ok) + res = response.json() + self.assertIn('Path is calculated', + res['output']['configuration-response-common']['response-message']) + self.assertEqual(9, res['output']['response-parameters']['path-description'] + ['aToZ-direction']['aToZ-wavelength-number']) + self.assertEqual(9, res['output']['response-parameters']['path-description'] + ['zToA-direction']['zToA-wavelength-number']) + time.sleep(5) + + # 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(self.restconf_baseurl)) + body = {"input": { + "service-name": "service 1", + "resource-reserve": "true", + "service-handler-header": { + "request-id": "request 1" + }, + "service-a-end": { + "service-rate": "100", + "service-format": "Ethernet", + "node-id": "XPONDER-2-2", + "clli": "ORANGE2" + }, + "service-z-end": { + "service-rate": "100", + "service-format": "Ethernet", + "node-id": "XPONDER-1-2", + "clli": "ORANGE1" + }, + "pce-metric": "hop-count" + } + } + headers = {'content-type': 'application/json', + "Accept": "application/json"} + response = requests.request( + "POST", url, data=json.dumps(body), headers=headers, + auth=('admin', 'admin')) + self.assertEqual(response.status_code, requests.codes.ok) + res = response.json() + self.assertIn('Path is calculated', + res['output']['configuration-response-common']['response-message']) + nbElmPath = len(res['output']['response-parameters']['path-description'] + ['aToZ-direction']['aToZ']) + self.assertEqual(31, nbElmPath) + link = {"link-id": "OpenROADM-1-3-DEG2-to-OpenROADM-1-2-DEG2"} + find = False + for i in range(0, nbElmPath): + resource_i = res['output']['response-parameters']['path-description']['aToZ-direction']['aToZ'][i]['resource'] + if(resource_i == link): + find = True + self.assertEqual(find, True) + time.sleep(5) + + # Delete oms-attribute in the link :openroadm1-3 to openroadm1-2 + 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(self.restconf_baseurl)) + headers = {'content-type': 'application/xml', + "Accept": "application/json"} + response = requests.request( + "DELETE", url, headers=headers, auth=('admin', 'admin')) + 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(self.restconf_baseurl)) + body = {"input": { + "service-name": "service 1", + "resource-reserve": "true", + "service-handler-header": { + "request-id": "request 1" + }, + "service-a-end": { + "service-rate": "100", + "service-format": "Ethernet", + "node-id": "XPONDER-2-2", + "clli": "ORANGE2" + }, + "service-z-end": { + "service-rate": "100", + "service-format": "Ethernet", + "node-id": "XPONDER-1-2", + "clli": "ORANGE1" + }, + "pce-metric": "hop-count" + } + } headers = {'content-type': 'application/json', - "Accept": "application/json"} + "Accept": "application/json"} response = requests.request( "POST", url, data=json.dumps(body), headers=headers, auth=('admin', 'admin')) self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertIn('Path is calculated', - res['output']['configuration-response-common']['response-message']) - self.assertEqual(9 , res['output']['response-parameters']['path-description'] - ['aToZ-direction']['aToZ-wavelength-number']) - self.assertEqual(9 , res['output']['response-parameters']['path-description'] - ['zToA-direction']['zToA-wavelength-number']) + res['output']['configuration-response-common']['response-message']) + nbElmPath = len(res['output']['response-parameters']['path-description'] + ['aToZ-direction']['aToZ']) + self.assertEqual(47, nbElmPath) + link = {"link-id": "OpenROADM-1-3-DEG2-to-OpenROADM-1-2-DEG2"} + find = False + for i in range(0, nbElmPath): + resource_i = res['output']['response-parameters']['path-description']['aToZ-direction']['aToZ'][i]['resource'] + if (resource_i == link): + find = True + self.assertNotEqual(find, True) time.sleep(5) # Delete complex topology - def test_21_delete_complex_topology(self): + def test_24_delete_complex_topology(self): url = ("{}/config/ietf-network:networks/network/openroadm-topology" - .format(self.restconf_baseurl)) + .format(self.restconf_baseurl)) headers = {'content-type': 'application/xml', - "Accept": "application/json"} + "Accept": "application/json"} response = requests.request( "DELETE", url, headers=headers, auth=('admin', 'admin')) self.assertEqual(response.status_code, requests.codes.ok) time.sleep(2) # Test deleted complex topology - def test_22_test_topology_complex_deleted(self): + def test_25_test_topology_complex_deleted(self): url = ("{}/config/ietf-network:networks/network/openroadm-topology/node/XPONDER-3-2" - .format(self.restconf_baseurl)) + .format(self.restconf_baseurl)) headers = {'content-type': 'application/json', - "Accept": "application/json"} + "Accept": "application/json"} response = requests.request( "GET", url, headers=headers, auth=('admin', 'admin')) self.assertEqual(response.status_code, 404)