X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=tests%2Ftransportpce_tests%2F1.2.1%2Ftest_pce.py;h=a1b35323b03b679e47eb5b6d7321ebd9f8bc2609;hb=19a9408ee823754859467888a5d2823bfd8655ab;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..a1b35323b 100644 --- a/tests/transportpce_tests/1.2.1/test_pce.py +++ b/tests/transportpce_tests/1.2.1/test_pce.py @@ -8,80 +8,66 @@ # 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" + 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" + 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() + processes = None + @classmethod - def setUpClass(cls): # a class method called before tests in an individual class run. + def setUpClass(cls): cls._get_file() - 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) # 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')) + url = "{}/config/ietf-network:networks/network/openroadm-topology" + data = self.simple_topo_bi_dir_data + response = test_utils.put_xmlrequest(url, 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')) + url = "{}/config/ietf-network:networks/network/openroadm-topology/node/ROADMA01-SRG1" + response = test_utils.get_request(url) self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertEqual( @@ -90,12 +76,8 @@ 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')) + url = "{}/config/ietf-network:networks/network/openroadm-topology/link/XPDRA01-XPDR1-XPDR1-NETWORK1toROADMA01-SRG1-SRG1-PP1-TXRX" + response = test_utils.get_request(url) self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertEqual( @@ -105,9 +87,8 @@ 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": { + url = "{}/operations/transportpce-pce:path-computation-request" + data = {"input": { "service-name": "service-1", "resource-reserve": "true", "pce-metric": "hop-count", @@ -128,11 +109,7 @@ class TransportPCEtesting(unittest.TestCase): } } } - 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(url, data) self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertIn('Path is calculated', @@ -141,9 +118,8 @@ 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": { + url = "{}/operations/transportpce-pce:path-computation-request" + data = {"input": { "service-name": "service-1", "resource-reserve": "true", "pce-metric": "hop-count", @@ -164,11 +140,7 @@ class TransportPCEtesting(unittest.TestCase): } } } - 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(url, data) self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertIn('Path is calculated', @@ -177,47 +149,30 @@ 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')) + url = "{}/config/ietf-network:networks/network/openroadm-topology" + response = test_utils.delete_request(url) 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')) + url = "{}/config/ietf-network:networks/network/openroadm-topology/node/ROADMA01-SRG1" + response = test_utils.get_request(url) 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)) - 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')) + url = "{}/config/ietf-network:networks/network/openroadm-topology" + data = self.simple_topo_uni_dir_data + response = test_utils.put_xmlrequest(url, 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')) + url = "{}/config/ietf-network:networks/network/openroadm-topology/node/XPONDER-1-2" + response = test_utils.get_request(url) self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertEqual( @@ -227,12 +182,8 @@ 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')) + url = "{}/config/ietf-network:networks/network/openroadm-topology/link/XPONDER-1-2XPDR-NW1-TX-toOpenROADM-1-2-SRG1-SRG1-PP1-RX" + response = test_utils.get_request(url) self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertEqual( @@ -242,9 +193,8 @@ 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": { + url = "{}/operations/transportpce-pce:path-computation-request" + data = {"input": { "service-name": "service-1", "resource-reserve": "true", "pce-metric": "hop-count", @@ -265,11 +215,7 @@ class TransportPCEtesting(unittest.TestCase): } } } - 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(url, data) self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertIn('Path is calculated', @@ -278,9 +224,8 @@ 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": { + url = "{}/operations/transportpce-pce:path-computation-request" + data = {"input": { "service-name": "service1", "resource-reserve": "true", "service-handler-header": { @@ -301,11 +246,7 @@ 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')) + response = test_utils.post_request(url, data) self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertIn('Path is calculated', @@ -326,47 +267,30 @@ 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"} - response = requests.request( - "DELETE", url, headers=headers, auth=('admin', 'admin')) + url = "{}/config/ietf-network:networks/network/openroadm-topology" + response = test_utils.delete_request(url) 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')) + url = "{}/config/ietf-network:networks/network/openroadm-topology/node/XPONDER-1-2" + response = test_utils.get_request(url) self.assertEqual(response.status_code, 404) 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')) + url = "{}/config/ietf-network:networks/network/openroadm-topology" + data = self.complex_topo_uni_dir_data + response = test_utils.put_xmlrequest(url, 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')) + url = "{}/config/ietf-network:networks/network/openroadm-topology/node/XPONDER-3-2" + response = test_utils.get_request(url) self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertEqual( @@ -376,19 +300,14 @@ 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": { + url = "{}/operations/transportpce-pce:path-computation-request" + data = {"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(url, data) self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertIn('Service Name is not set', @@ -397,9 +316,8 @@ 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": { + url = "{}/operations/transportpce-pce:path-computation-request" + data = {"input": { "service-name": "service1", "resource-reserve": "true", "service-handler-header": { @@ -485,11 +403,7 @@ class TransportPCEtesting(unittest.TestCase): "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.post_request(url, data) self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertIn('Path is calculated', @@ -498,9 +412,8 @@ 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": { + url = "{}/operations/transportpce-pce:path-computation-request" + data = {"input": { "service-name": "service 1", "resource-reserve": "true", "service-handler-header": { @@ -521,11 +434,7 @@ 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')) + response = test_utils.post_request(url, data) self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertIn('Path is calculated', @@ -538,9 +447,8 @@ 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": { + url = "{}/operations/transportpce-pce:path-computation-request" + data = {"input": { "service-name": "service 1", "resource-reserve": "true", "service-handler-header": { @@ -566,11 +474,7 @@ 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')) + response = test_utils.post_request(url, data) self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertIn('Path is calculated', @@ -583,9 +487,8 @@ 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": { + url = "{}/operations/transportpce-pce:path-computation-request" + data = {"input": { "service-name": "service 1", "resource-reserve": "true", "service-handler-header": { @@ -606,11 +509,7 @@ 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')) + response = test_utils.post_request(url, data) self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertIn('Path is calculated', @@ -631,19 +530,15 @@ class TransportPCEtesting(unittest.TestCase): 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.delete_request(url) 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": { + url = "{}/operations/transportpce-pce:path-computation-request" + data = {"input": { "service-name": "service 1", "resource-reserve": "true", "service-handler-header": { @@ -664,11 +559,7 @@ 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')) + response = test_utils.post_request(url, data) self.assertEqual(response.status_code, requests.codes.ok) res = response.json() self.assertIn('Path is calculated', @@ -687,23 +578,15 @@ class TransportPCEtesting(unittest.TestCase): # 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')) + url = "{}/config/ietf-network:networks/network/openroadm-topology" + response = test_utils.delete_request(url) 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')) + url = "{}/config/ietf-network:networks/network/openroadm-topology/node/XPONDER-3-2" + response = test_utils.get_request(url) self.assertEqual(response.status_code, 404) time.sleep(1)