X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=tests%2Ftransportpce_tests%2F1.2.1%2Ftest_pce.py;h=c33af419a172d12ac174d5480fe3f78742fcbf58;hb=78881abcc31dc039c374bcc15266030e8dabcf87;hp=1465bf2dd64cc6f5b218169e1588012bc0d37316;hpb=e547f6844b7abb6ea16aa5d9cbceaab1e10e7cbe;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 1465bf2dd..c33af419a 100644 --- a/tests/transportpce_tests/1.2.1/test_pce.py +++ b/tests/transportpce_tests/1.2.1/test_pce.py @@ -8,56 +8,50 @@ # http://www.apache.org/licenses/LICENSE-2.0 ############################################################################## +import unittest import json import os -import psutil -import requests -import signal -import shutil -import subprocess import time -import unittest -import test_utils +import requests +from common import test_utils class TransportPCEtesting(unittest.TestCase): - odl_process = None simple_topo_bi_dir_data = None simple_topo_uni_dir_data = None complex_topo_uni_dir_data = None - restconf_baseurl = "http://localhost:8181/restconf" @classmethod def _get_file(cls): - topo_bi_dir_file = "sample_configs/honeynode-topo.xml" + topo_bi_dir_file = os.path.join(os.path.dirname(os.path.realpath(__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(); - topo_uni_dir_file = "sample_configs/NW-simple-topology.xml" + cls.simple_topo_bi_dir_data = topo_bi_dir.read() + topo_uni_dir_file = os.path.join(os.path.dirname(os.path.realpath(__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(); - topo_uni_dir_complex_file = "sample_configs/NW-for-test-5-4.xml" + cls.simple_topo_uni_dir_data = topo_uni_dir.read() + topo_uni_dir_complex_file = os.path.join(os.path.dirname(os.path.realpath(__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() + + processes = None @classmethod - def setUpClass(cls): # a class method called before tests in an individual class run. + def setUpClass(cls): cls._get_file() - print ("starting opendaylight...") - cls.odl_process = test_utils.start_tpce() - time.sleep(90) - print ("opendaylight started") + cls.processes = test_utils.start_tpce() @classmethod def tearDownClass(cls): - for child in psutil.Process(cls.odl_process.pid).children(): - child.send_signal(signal.SIGINT) - child.wait() - cls.odl_process.send_signal(signal.SIGINT) - cls.odl_process.wait() + for process in cls.processes: + test_utils.shutdown_process(process) + print("all processes killed") def setUp(self): # instruction executed before each test method time.sleep(1) @@ -65,24 +59,20 @@ 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(test_utils.RESTCONF_BASE_URL)) body = self.simple_topo_bi_dir_data - headers = {'content-type': 'application/xml', - "Accept": "application/xml"} response = requests.request( - "PUT", url, data=body, headers=headers, - auth=('admin', 'admin')) + "PUT", url, data=body, headers=test_utils.TYPE_APPLICATION_XML, + auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD)) 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(self.restconf_baseurl)) - headers = {'content-type': 'application/json', - "Accept": "application/json"} + .format(test_utils.RESTCONF_BASE_URL)) response = requests.request( - "GET", url, headers=headers, auth=('admin', 'admin')) + "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() self.assertEqual( @@ -92,11 +82,9 @@ class TransportPCEtesting(unittest.TestCase): # 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(self.restconf_baseurl)) - headers = {'content-type': 'application/json', - "Accept": "application/json"} + .format(test_utils.RESTCONF_BASE_URL)) response = requests.request( - "GET", url, headers=headers, auth=('admin', 'admin')) + "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() self.assertEqual( @@ -107,7 +95,7 @@ class TransportPCEtesting(unittest.TestCase): # Path Computation success def test_04_path_computation_xpdr_bi(self): url = ("{}/operations/transportpce-pce:path-computation-request" - .format(self.restconf_baseurl)) + .format(test_utils.RESTCONF_BASE_URL)) body = {"input": { "service-name": "service-1", "resource-reserve": "true", @@ -117,31 +105,31 @@ class TransportPCEtesting(unittest.TestCase): }, "service-a-end": { "node-id": "XPDRA01", - "service-rate": "0", + "service-rate": "100", + "service-format": "Ethernet", "clli": "nodeA" }, "service-z-end": { "node-id": "XPDRC01", - "service-rate": "0", + "service-rate": "100", + "service-format": "Ethernet", "clli": "nodeC" } - } - } - headers = {'content-type': 'application/json', - "Accept": "application/json"} + } + } response = requests.request( - "POST", url, data=json.dumps(body), headers=headers, - auth=('admin', 'admin')) + "POST", url, data=json.dumps(body), 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() 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(test_utils.RESTCONF_BASE_URL)) body = {"input": { "service-name": "service-1", "resource-reserve": "true", @@ -151,70 +139,62 @@ class TransportPCEtesting(unittest.TestCase): }, "service-a-end": { "node-id": "ROADMA01", - "service-rate": "0", - "clli": "nodeA" + "service-rate": "100", + "service-format": "Ethernet", + "clli": "NodeA" }, "service-z-end": { "node-id": "ROADMC01", - "service-rate": "0", - "clli": "nodeC" + "service-rate": "100", + "service-format": "Ethernet", + "clli": "NodeC" + } + } } - } - } - headers = {'content-type': 'application/json', - "Accept": "application/json"} response = requests.request( - "POST", url, data=json.dumps(body), headers=headers, - auth=('admin', 'admin')) + "POST", url, data=json.dumps(body), 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() 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)) - headers = {'content-type': 'application/xml', - "Accept": "application/json"} + .format(test_utils.RESTCONF_BASE_URL)) response = requests.request( - "DELETE", url, headers=headers, auth=('admin', 'admin')) + "DELETE", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD)) 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(self.restconf_baseurl)) - headers = {'content-type': 'application/json', - "Accept": "application/json"} + .format(test_utils.RESTCONF_BASE_URL)) response = requests.request( - "GET", url, headers=headers, auth=('admin', 'admin')) + "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD)) 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(self.restconf_baseurl)) + .format(test_utils.RESTCONF_BASE_URL)) body = self.simple_topo_uni_dir_data - headers = {'content-type': 'application/xml', - "Accept": "application/xml"} response = requests.request( - "PUT", url, data=body, headers=headers, - auth=('admin', 'admin')) + "PUT", url, data=body, headers=test_utils.TYPE_APPLICATION_XML, + auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD)) 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(self.restconf_baseurl)) - headers = {'content-type': 'application/json', - "Accept": "application/json"} + .format(test_utils.RESTCONF_BASE_URL)) response = requests.request( - "GET", url, headers=headers, auth=('admin', 'admin')) + "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() self.assertEqual( @@ -225,11 +205,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)) - headers = {'content-type': 'application/json', - "Accept": "application/json"} + .format(test_utils.RESTCONF_BASE_URL)) response = requests.request( - "GET", url, headers=headers, auth=('admin', 'admin')) + "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() self.assertEqual( @@ -240,7 +218,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(test_utils.RESTCONF_BASE_URL)) body = {"input": { "service-name": "service-1", "resource-reserve": "true", @@ -250,65 +228,65 @@ 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"} response = requests.request( - "POST", url, data=json.dumps(body), headers=headers, - auth=('admin', 'admin')) + "POST", url, data=json.dumps(body), 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() 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(test_utils.RESTCONF_BASE_URL)) 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"} + } + } response = requests.request( - "POST", url, data=json.dumps(body), headers=headers, - auth=('admin', 'admin')) + "POST", url, data=json.dumps(body), 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() 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,46 +298,38 @@ 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)) - headers = {'content-type': 'application/xml', - "Accept": "application/json"} + .format(test_utils.RESTCONF_BASE_URL)) response = requests.request( - "DELETE", url, headers=headers, auth=('admin', 'admin')) + "DELETE", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD)) 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(self.restconf_baseurl)) - headers = {'content-type': 'application/json', - "Accept": "application/json"} + .format(test_utils.RESTCONF_BASE_URL)) response = requests.request( - "GET", url, headers=headers, auth=('admin', 'admin')) + "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD)) 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(test_utils.RESTCONF_BASE_URL)) body = self.complex_topo_uni_dir_data - headers = {'content-type': 'application/xml', - "Accept": "application/json"} response = requests.request( - "PUT", url, data=body, headers=headers, - auth=('admin', 'admin')) + "PUT", url, data=body, headers=test_utils.TYPE_APPLICATION_XML, + auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD)) 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(self.restconf_baseurl)) - headers = {'content-type': 'application/json', - "Accept": "application/json"} + .format(test_utils.RESTCONF_BASE_URL)) response = requests.request( - "GET", url, headers=headers, auth=('admin', 'admin')) + "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() self.assertEqual( @@ -370,129 +340,125 @@ 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(test_utils.RESTCONF_BASE_URL)) body = {"input": { "service-handler-header": { "request-id": "request-1" } - } - } - headers = {'content-type': 'application/json', - "Accept": "application/json"} + } + } response = requests.request( - "POST", url, data=json.dumps(body), headers=headers, - auth=('admin', 'admin')) + "POST", url, data=json.dumps(body), 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() 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(test_utils.RESTCONF_BASE_URL)) 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"} + } + } response = requests.request( - "POST", url, data=json.dumps(body), headers=headers, - auth=('admin', 'admin')) + "POST", url, data=json.dumps(body), 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() 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(test_utils.RESTCONF_BASE_URL)) body = {"input": { "service-name": "service 1", "resource-reserve": "true", @@ -500,35 +466,37 @@ 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" - } - } - headers = {'content-type': 'application/json', - "Accept": "application/json"} + "service-rate": "100", + "service-format": "Ethernet", + "node-id": "XPONDER-3-2", + "clli": "ORANGE3" + }, + "pce-metric": "hop-count" + } + } response = requests.request( - "POST", url, data=json.dumps(body), headers=headers, - auth=('admin', 'admin')) + "POST", url, data=json.dumps(body), 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() 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(test_utils.RESTCONF_BASE_URL)) body = {"input": { "service-name": "service 1", "resource-reserve": "true", @@ -536,12 +504,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,42 +521,134 @@ 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=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() + 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(test_utils.RESTCONF_BASE_URL)) + 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" + } + } + response = requests.request( + "POST", url, data=json.dumps(body), 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() + 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(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)) + 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": { + "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" + } + } response = requests.request( - "POST", url, data=json.dumps(body), headers=headers, - auth=('admin', 'admin')) + "POST", url, data=json.dumps(body), 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() 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)) - headers = {'content-type': 'application/xml', - "Accept": "application/json"} + .format(test_utils.RESTCONF_BASE_URL)) response = requests.request( - "DELETE", url, headers=headers, auth=('admin', 'admin')) + "DELETE", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD)) 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)) - headers = {'content-type': 'application/json', - "Accept": "application/json"} + .format(test_utils.RESTCONF_BASE_URL)) response = requests.request( - "GET", url, headers=headers, auth=('admin', 'admin')) + "GET", url, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD)) self.assertEqual(response.status_code, 404) time.sleep(1)