X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=tests%2Ftransportpce_tests%2F1.2.1%2Ftest_pce.py;h=e7e255f9134ecd311beb997f59dae9ab4c646ef8;hb=6fe500a1bf557eb44d22584ac6bf7c924f12c47f;hp=5375dabc7e75c216f305ef20e29e6b7808627842;hpb=f9bb9e323c48640b14908cbc4101c2ad41b1de60;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 5375dabc7..e7e255f91 100644 --- a/tests/transportpce_tests/1.2.1/test_pce.py +++ b/tests/transportpce_tests/1.2.1/test_pce.py @@ -8,80 +8,79 @@ # http://www.apache.org/licenses/LICENSE-2.0 ############################################################################## -import json +# pylint: disable=no-member +# pylint: disable=too-many-public-methods + +import unittest import os -import psutil -import requests -import signal -import shutil -import subprocess +import sys 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" + processes = None @classmethod - def _get_file(cls): - 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: + def setUpClass(cls): + # pylint: disable=bare-except + try: + sample_files_parsed = False + TOPO_BI_DIR_FILE = os.path.join(os.path.dirname(os.path.realpath(__file__)), + "..", "..", "sample_configs", "honeynode-topo.xml") + 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" - if os.path.isfile(topo_uni_dir_file): - with open(topo_uni_dir_file, 'r') as topo_uni_dir: + + TOPO_UNI_DIR_FILE = os.path.join(os.path.dirname(os.path.realpath(__file__)), + "..", "..", "sample_configs", "NW-simple-topology.xml") + + 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" - 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() - @classmethod - def setUpClass(cls): # a class method called before tests in an individual class run. - cls._get_file() - cls.odl_process = test_utils.start_tpce() - time.sleep(90) - print("opendaylight started") + TOPO_UNI_DIR_COMPLEX_FILE = os.path.join(os.path.dirname(os.path.realpath(__file__)), + "..", "..", "sample_configs", "NW-for-test-5-4.xml") + with open(TOPO_UNI_DIR_COMPLEX_FILE, 'r') as topo_uni_dir_complex: + cls.complex_topo_uni_dir_data = topo_uni_dir_complex.read() + sample_files_parsed = True + except PermissionError as err: + print("Permission Error when trying to read sample files\n", err) + sys.exit(2) + except FileNotFoundError as err: + print("File Not found Error when trying to read sample files\n", err) + sys.exit(2) + except: + print("Unexpected error when trying to read sample files\n", sys.exc_info()[0]) + sys.exit(2) + finally: + if sample_files_parsed: + print("sample files content loaded") + + 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() + # pylint: disable=not-an-iterable + 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) # Load simple bidirectional topology def test_01_load_simple_topology_bi(self): - url = ("{}/config/ietf-network:networks/network/openroadm-topology" - .format(self.restconf_baseurl)) - 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')) + response = test_utils.put_xmlrequest(test_utils.URL_CONFIG_ORDM_TOPO, self.simple_topo_bi_dir_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(self.restconf_baseurl)) - headers = {'content-type': 'application/json', - "Accept": "application/json"} - response = requests.request( - "GET", url, headers=headers, auth=('admin', 'admin')) + response = test_utils.get_ordm_topo_request("node/ROADMA01-SRG1") self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertEqual( @@ -90,12 +89,7 @@ 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"} - response = requests.request( - "GET", url, headers=headers, auth=('admin', 'admin')) + response = test_utils.get_ordm_topo_request("link/XPDRA01-XPDR1-XPDR1-NETWORK1toROADMA01-SRG1-SRG1-PP1-TXRX") self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertEqual( @@ -105,34 +99,11 @@ 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)) - body = {"input": { - "service-name": "service-1", - "resource-reserve": "true", - "pce-metric": "hop-count", - "service-handler-header": { - "request-id": "request-1" - }, - "service-a-end": { - "node-id": "XPDRA01", - "service-rate": "100", - "service-format": "Ethernet", - "clli": "nodeA" - }, - "service-z-end": { - "node-id": "XPDRC01", - "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')) + response = test_utils.path_computation_request("request-1", "service-1", + {"node-id": "XPDRA01", "service-rate": "100", + "service-format": "Ethernet", "clli": "nodeA"}, + {"node-id": "XPDRC01", "service-rate": "100", + "service-format": "Ethernet", "clli": "nodeC"}) self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertIn('Path is calculated', @@ -141,34 +112,11 @@ class TransportPCEtesting(unittest.TestCase): # Path Computation success def test_05_path_computation_rdm_bi(self): - url = ("{}/operations/transportpce-pce:path-computation-request" - .format(self.restconf_baseurl)) - body = {"input": { - "service-name": "service-1", - "resource-reserve": "true", - "pce-metric": "hop-count", - "service-handler-header": { - "request-id": "request-1" - }, - "service-a-end": { - "node-id": "ROADMA01", - "service-rate": "100", - "service-format": "Ethernet", - "clli": "NodeA" - }, - "service-z-end": { - "node-id": "ROADMC01", - "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')) + response = test_utils.path_computation_request("request-1", "service-1", + {"node-id": "ROADMA01", "service-rate": "100", + "service-format": "Ethernet", "clli": "NodeA"}, + {"node-id": "ROADMC01", "service-rate": "100", + "service-format": "Ethernet", "clli": "NodeC"}) self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertIn('Path is calculated', @@ -177,47 +125,25 @@ class TransportPCEtesting(unittest.TestCase): # 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"} - response = requests.request( - "DELETE", url, headers=headers, auth=('admin', 'admin')) + response = test_utils.delete_request(test_utils.URL_CONFIG_ORDM_TOPO) 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"} - response = requests.request( - "GET", url, headers=headers, auth=('admin', 'admin')) - self.assertEqual(response.status_code, 404) + response = test_utils.get_ordm_topo_request("node/ROADMA01-SRG1") + self.assertEqual(response.status_code, requests.codes.conflict) 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)) - 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')) + response = test_utils.put_xmlrequest(test_utils.URL_CONFIG_ORDM_TOPO, self.simple_topo_uni_dir_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(self.restconf_baseurl)) - headers = {'content-type': 'application/json', - "Accept": "application/json"} - response = requests.request( - "GET", url, headers=headers, auth=('admin', 'admin')) + response = test_utils.get_ordm_topo_request("node/XPONDER-1-2") self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertEqual( @@ -227,12 +153,7 @@ 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"} - response = requests.request( - "GET", url, headers=headers, auth=('admin', 'admin')) + response = test_utils.get_ordm_topo_request("link/XPONDER-1-2XPDR-NW1-TX-toOpenROADM-1-2-SRG1-SRG1-PP1-RX") self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertEqual( @@ -242,34 +163,11 @@ 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)) - body = {"input": { - "service-name": "service-1", - "resource-reserve": "true", - "pce-metric": "hop-count", - "service-handler-header": { - "request-id": "request-1" - }, - "service-a-end": { - "node-id": "XPONDER-1-2", - "service-rate": "100", - "service-format": "Ethernet", - "clli": "ORANGE1" - }, - "service-z-end": { - "node-id": "XPONDER-3-2", - "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')) + response = test_utils.path_computation_request("request-1", "service-1", + {"node-id": "XPONDER-1-2", "service-rate": "100", + "service-format": "Ethernet", "clli": "ORANGE1"}, + {"node-id": "XPONDER-3-2", "service-rate": "100", + "service-format": "Ethernet", "clli": "ORANGE3"}) self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertIn('Path is calculated', @@ -278,34 +176,11 @@ class TransportPCEtesting(unittest.TestCase): # Path Computation success def test_12_path_computation_rdm_uni(self): - url = ("{}/operations/transportpce-pce:path-computation-request" - .format(self.restconf_baseurl)) - body = {"input": { - "service-name": "service1", - "resource-reserve": "true", - "service-handler-header": { - "request-id": "request1" - }, - "service-a-end": { - "service-rate": "100", - "service-format": "Ethernet", - "clli": "cll21", - "node-id": "OpenROADM-2-1" - }, - "service-z-end": { - "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')) + response = test_utils.path_computation_request("request1", "service1", + {"service-rate": "100", "service-format": "Ethernet", + "clli": "cll21", "node-id": "OpenROADM-2-1"}, + {"service-rate": "100", "service-format": "Ethernet", + "clli": "ncli22", "node-id": "OpenROADM-2-2"}) self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertIn('Path is calculated', @@ -318,55 +193,33 @@ class TransportPCEtesting(unittest.TestCase): 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'): + if atoz['id'] == '14': self.assertEqual(atoz['resource']['tp-id'], 'SRG1-PP1-TX') - if (ztoa['id'] == '0'): + if ztoa['id'] == '0': self.assertEqual(ztoa['resource']['tp-id'], 'SRG1-PP1-RX') time.sleep(5) # 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"} - response = requests.request( - "DELETE", url, headers=headers, auth=('admin', 'admin')) + response = test_utils.delete_request(test_utils.URL_CONFIG_ORDM_TOPO) 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"} - response = requests.request( - "GET", url, headers=headers, auth=('admin', 'admin')) - self.assertEqual(response.status_code, 404) + response = test_utils.get_ordm_topo_request("node/XPONDER-1-2") + self.assertEqual(response.status_code, requests.codes.conflict) time.sleep(1) # Load complex topology def test_15_load_complex_topology(self): - url = ("{}/config/ietf-network:networks/network/openroadm-topology" - .format(self.restconf_baseurl)) - 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')) + response = test_utils.put_xmlrequest(test_utils.URL_CONFIG_ORDM_TOPO, self.complex_topo_uni_dir_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(self.restconf_baseurl)) - headers = {'content-type': 'application/json', - "Accept": "application/json"} - response = requests.request( - "GET", url, headers=headers, auth=('admin', 'admin')) + response = test_utils.get_ordm_topo_request("node/XPONDER-3-2") self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertEqual( @@ -376,19 +229,8 @@ 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)) - 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')) + response = test_utils.post_request(test_utils.URL_PATH_COMPUTATION_REQUEST, + {"input": {"service-handler-header": {"request-id": "request-1"}}}) self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertIn('Service Name is not set', @@ -397,99 +239,54 @@ class TransportPCEtesting(unittest.TestCase): # Test1 success path computation def test_18_success1_path_computation(self): - url = ("{}/operations/transportpce-pce:path-computation-request" - .format(self.restconf_baseurl)) - body = {"input": { - "service-name": "service1", - "resource-reserve": "true", - "service-handler-header": { - "request-id": "request1" - }, - "service-a-end": { - "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": "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" - ] - } - }, - "soft-constraints": { - "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')) + response = test_utils.path_computation_request("request1", "service1", + {"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-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" + }}}, + {"customer-code": ["Some customer-code"], + "co-routing": {"existing-service": ["Some existing-service"]} + }, + {"customer-code": ["Some customer-code"], + "co-routing": {"existing-service": ["Some existing-service"]} + }, + "hop-count", {"locally-protected-links": "true"}) self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertIn('Path is calculated', @@ -498,34 +295,11 @@ class TransportPCEtesting(unittest.TestCase): # 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)) - 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-1-2", - "clli": "ORANGE1" - }, - "service-z-end": { - "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"} - response = requests.request( - "POST", url, data=json.dumps(body), headers=headers, - auth=('admin', 'admin')) + response = test_utils.path_computation_request("request 1", "service 1", + {"service-rate": "100", "service-format": "Ethernet", + "node-id": "XPONDER-1-2", "clli": "ORANGE1"}, + {"service-rate": "100", "service-format": "Ethernet", + "node-id": "XPONDER-3-2", "clli": "ORANGE3"}) self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertIn('Path is calculated', @@ -538,39 +312,12 @@ class TransportPCEtesting(unittest.TestCase): # 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)) - 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-1-2", - "clli": "ORANGE1" - }, - "service-z-end": { - "service-rate": "100", - "service-format": "Ethernet", - "node-id": "XPONDER-3-2", - "clli": "ORANGE3" - }, - "hard-constraints": { - "exclude_": { - "node-id": ["OpenROADM-2-1", "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')) + response = test_utils.path_computation_request("request 1", "service 1", + {"service-rate": "100", "service-format": "Ethernet", + "node-id": "XPONDER-1-2", "clli": "ORANGE1"}, + {"service-rate": "100", "service-format": "Ethernet", + "node-id": "XPONDER-3-2", "clli": "ORANGE3"}, + {"exclude_": {"node-id": ["OpenROADM-2-1", "OpenROADM-2-2"]}}) self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertIn('Path is calculated', @@ -583,34 +330,11 @@ class TransportPCEtesting(unittest.TestCase): # 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')) + response = test_utils.path_computation_request("request 1", "service 1", + {"service-rate": "100", "service-format": "Ethernet", + "node-id": "XPONDER-2-2", "clli": "ORANGE2"}, + {"service-rate": "100", "service-format": "Ethernet", + "node-id": "XPONDER-1-2", "clli": "ORANGE1"}) self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertIn('Path is calculated', @@ -621,54 +345,26 @@ class TransportPCEtesting(unittest.TestCase): 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): + 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')) + response = test_utils.del_oms_attr_request("OpenROADM-1-3-DEG2-to-OpenROADM-1-2-DEG2") 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"} - response = requests.request( - "POST", url, data=json.dumps(body), headers=headers, - auth=('admin', 'admin')) + response = test_utils.path_computation_request("request 1", "service 1", + {"service-rate": "100", "service-format": "Ethernet", + "node-id": "XPONDER-2-2", "clli": "ORANGE2"}, + {"service-rate": "100", "service-format": "Ethernet", + "node-id": "XPONDER-1-2", "clli": "ORANGE1"}) self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertIn('Path is calculated', @@ -679,32 +375,23 @@ class TransportPCEtesting(unittest.TestCase): 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): + 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_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"} - response = requests.request( - "DELETE", url, headers=headers, auth=('admin', 'admin')) + response = test_utils.delete_request(test_utils.URL_CONFIG_ORDM_TOPO) 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(self.restconf_baseurl)) - headers = {'content-type': 'application/json', - "Accept": "application/json"} - response = requests.request( - "GET", url, headers=headers, auth=('admin', 'admin')) - self.assertEqual(response.status_code, 404) + response = test_utils.get_ordm_topo_request("node/XPONDER-3-2") + self.assertEqual(response.status_code, requests.codes.conflict) time.sleep(1)