X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=tests%2Ftransportpce_tests%2F2.2.1%2Ftest_tapi.py;h=496a33e65d88d38f5eccd6a35b460f9e424cc340;hb=6fe500a1bf557eb44d22584ac6bf7c924f12c47f;hp=b5796895cf8ee552542adcf46a8d235030bab796;hpb=78881abcc31dc039c374bcc15266030e8dabcf87;p=transportpce.git diff --git a/tests/transportpce_tests/2.2.1/test_tapi.py b/tests/transportpce_tests/2.2.1/test_tapi.py index b5796895c..496a33e65 100644 --- a/tests/transportpce_tests/2.2.1/test_tapi.py +++ b/tests/transportpce_tests/2.2.1/test_tapi.py @@ -7,6 +7,13 @@ # which accompanies this distribution, and is available at # http://www.apache.org/licenses/LICENSE-2.0 ############################################################################## + +# pylint: disable=no-member +# pylint: disable=too-many-public-methods +# some pylint false positives specific to tapi test +# pylint: disable=unsubscriptable-object +# pylint: disable=unsupported-assignment-operation + import os import sys import time @@ -27,7 +34,8 @@ class TransportTapitesting(unittest.TestCase): @classmethod def setUpClass(cls): cls.init_failed = False - + os.environ['JAVA_MIN_MEM'] = '1024M' + os.environ['JAVA_MAX_MEM'] = '4069M' cls.processes = test_utils.start_tpce() # TAPI feature is not installed by default in Karaf if "USE_LIGHTY" not in os.environ or os.environ['USE_LIGHTY'] != 'True': @@ -37,17 +45,19 @@ class TransportTapitesting(unittest.TestCase): cls.init_failed = True print("Restarting OpenDaylight...") test_utils.shutdown_process(cls.processes[0]) - cls.processes = test_utils.start_tpce() + cls.processes[0] = test_utils.start_karaf() + test_utils.process_list[0] = cls.processes[0] cls.init_failed = not test_utils.wait_until_log_contains( test_utils.KARAF_LOG, test_utils.KARAF_OK_START_MSG, time_to_wait=60) if cls.init_failed: - print("tapi installaiton feature failed...") - test_utils.shutdown_process(cls.odl_processes[0]) + print("tapi installation feature failed...") + test_utils.shutdown_process(cls.processes[0]) sys.exit(2) - cls.processes = test_utils.start_sims(['xpdra', 'roadma', 'roadmc', 'xpdrc', 'spdrav2']) + cls.processes = test_utils.start_sims(['xpdra', 'roadma', 'roadmc', 'xpdrc', 'spdra']) @classmethod def tearDownClass(cls): + # pylint: disable=not-an-iterable for process in cls.processes: test_utils.shutdown_process(process) print("all processes killed") @@ -58,7 +68,7 @@ class TransportTapitesting(unittest.TestCase): print("execution of {}".format(self.id().split(".")[-1])) def test_00_connect_spdr_sa1(self): - response = test_utils.mount_device("SPDR-SA1", 'spdrav2') + response = test_utils.mount_device("SPDR-SA1", 'spdra') self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201) time.sleep(10) # TODO replace connect and disconnect timers with test_utils.wait_until_log_contains @@ -84,9 +94,8 @@ class TransportTapitesting(unittest.TestCase): time.sleep(20) def test_05_connect_xprda_n1_to_roadma_pp1(self): - url = "{}/operations/transportpce-networkutils:init-xpdr-rdm-links".format(test_utils.RESTCONF_BASE_URL) - data = test_utils.generate_link_data("XPDR-A1", "1", "1", "ROADM-A1", "1", "SRG1-PP1-TXRX") - response = test_utils.post_request(url, data, test_utils.ODL_LOGIN, test_utils.ODL_PWD) + response = test_utils.connect_xpdr_to_rdm_request("XPDR-A1", "1", "1", + "ROADM-A1", "1", "SRG1-PP1-TXRX") self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200) res = response.json() self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"], @@ -94,9 +103,8 @@ class TransportTapitesting(unittest.TestCase): time.sleep(2) def test_06_connect_roadma_pp1_to_xpdra_n1(self): - url = "{}/operations/transportpce-networkutils:init-rdm-xpdr-links".format(test_utils.RESTCONF_BASE_URL) - data = test_utils.generate_link_data("XPDR-A1", "1", "1", "ROADM-A1", "1", "SRG1-PP1-TXRX") - response = test_utils.post_request(url, data, test_utils.ODL_LOGIN, test_utils.ODL_PWD) + response = test_utils.connect_rdm_to_xpdr_request("XPDR-A1", "1", "1", + "ROADM-A1", "1", "SRG1-PP1-TXRX") self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200) res = response.json() self.assertIn('Roadm Xponder links created successfully', res["output"]["result"], @@ -104,9 +112,8 @@ class TransportTapitesting(unittest.TestCase): time.sleep(2) def test_07_connect_xprdc_n1_to_roadmc_pp1(self): - url = "{}/operations/transportpce-networkutils:init-xpdr-rdm-links".format(test_utils.RESTCONF_BASE_URL) - data = test_utils.generate_link_data("XPDR-C1", "1", "1", "ROADM-C1", "1", "SRG1-PP1-TXRX") - response = test_utils.post_request(url, data, test_utils.ODL_LOGIN, test_utils.ODL_PWD) + response = test_utils.connect_xpdr_to_rdm_request("XPDR-C1", "1", "1", + "ROADM-C1", "1", "SRG1-PP1-TXRX") self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200) res = response.json() self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"], @@ -114,9 +121,8 @@ class TransportTapitesting(unittest.TestCase): time.sleep(2) def test_08_connect_roadmc_pp1_to_xpdrc_n1(self): - url = "{}/operations/transportpce-networkutils:init-rdm-xpdr-links".format(test_utils.RESTCONF_BASE_URL) - data = test_utils.generate_link_data("XPDR-C1", "1", "1", "ROADM-C1", "1", "SRG1-PP1-TXRX") - response = test_utils.post_request(url, data, test_utils.ODL_LOGIN, test_utils.ODL_PWD) + response = test_utils.connect_rdm_to_xpdr_request("XPDR-C1", "1", "1", + "ROADM-C1", "1", "SRG1-PP1-TXRX") self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200) res = response.json() self.assertIn('Roadm Xponder links created successfully', res["output"]["result"], @@ -124,9 +130,8 @@ class TransportTapitesting(unittest.TestCase): time.sleep(2) def test_09_connect_xprda_n2_to_roadma_pp2(self): - url = "{}/operations/transportpce-networkutils:init-xpdr-rdm-links".format(test_utils.RESTCONF_BASE_URL) - data = test_utils.generate_link_data("XPDR-A1", "1", "2", "ROADM-A1", "1", "SRG1-PP2-TXRX") - response = test_utils.post_request(url, data, test_utils.ODL_LOGIN, test_utils.ODL_PWD) + response = test_utils.connect_xpdr_to_rdm_request("XPDR-A1", "1", "2", + "ROADM-A1", "1", "SRG1-PP2-TXRX") self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200) res = response.json() self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"], @@ -134,9 +139,8 @@ class TransportTapitesting(unittest.TestCase): time.sleep(2) def test_10_connect_roadma_pp2_to_xpdra_n2(self): - url = "{}/operations/transportpce-networkutils:init-rdm-xpdr-links".format(test_utils.RESTCONF_BASE_URL) - data = test_utils.generate_link_data("XPDR-A1", "1", "2", "ROADM-A1", "1", "SRG1-PP2-TXRX") - response = test_utils.post_request(url, data, test_utils.ODL_LOGIN, test_utils.ODL_PWD) + response = test_utils.connect_rdm_to_xpdr_request("XPDR-A1", "1", "2", + "ROADM-A1", "1", "SRG1-PP2-TXRX") self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200) res = response.json() self.assertIn('Roadm Xponder links created successfully', res["output"]["result"], @@ -144,9 +148,8 @@ class TransportTapitesting(unittest.TestCase): time.sleep(2) def test_11_connect_xprdc_n2_to_roadmc_pp2(self): - url = "{}/operations/transportpce-networkutils:init-xpdr-rdm-links".format(test_utils.RESTCONF_BASE_URL) - data = test_utils.generate_link_data("XPDR-C1", "1", "2", "ROADM-C1", "1", "SRG1-PP2-TXRX") - response = test_utils.post_request(url, data, test_utils.ODL_LOGIN, test_utils.ODL_PWD) + response = test_utils.connect_xpdr_to_rdm_request("XPDR-C1", "1", "2", + "ROADM-C1", "1", "SRG1-PP2-TXRX") self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200) res = response.json() self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"], @@ -154,9 +157,8 @@ class TransportTapitesting(unittest.TestCase): time.sleep(2) def test_12_connect_roadmc_pp2_to_xpdrc_n2(self): - url = "{}/operations/transportpce-networkutils:init-rdm-xpdr-links".format(test_utils.RESTCONF_BASE_URL) - data = test_utils.generate_link_data("XPDR-C1", "1", "2", "ROADM-C1", "1", "SRG1-PP2-TXRX") - response = test_utils.post_request(url, data, test_utils.ODL_LOGIN, test_utils.ODL_PWD) + response = test_utils.connect_rdm_to_xpdr_request("XPDR-C1", "1", "2", + "ROADM-C1", "1", "SRG1-PP2-TXRX") self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200) res = response.json() self.assertIn('Roadm Xponder links created successfully', res["output"]["result"], @@ -164,14 +166,14 @@ class TransportTapitesting(unittest.TestCase): time.sleep(2) def test_13_get_tapi_openroadm_topology(self): - url = "{}/operations/tapi-topology:get-topology-details".format(test_utils.RESTCONF_BASE_URL) + url = "{}/operations/tapi-topology:get-topology-details" data = { "tapi-topology:input": { "tapi-topology:topology-id-or-name": "openroadm-topology" } } - response = test_utils.post_request(url, data, test_utils.ODL_LOGIN, test_utils.ODL_PWD) + response = test_utils.post_request(url, data) self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200) res = response.json() self.assertEqual(len(res["output"]["topology"]["node"]), 1, 'There should be 1 node') @@ -179,14 +181,14 @@ class TransportTapitesting(unittest.TestCase): 'There should be 4 owned-node-edge-points') def test_14_get_tapi_otn_topology(self): - url = "{}/operations/tapi-topology:get-topology-details".format(test_utils.RESTCONF_BASE_URL) + url = "{}/operations/tapi-topology:get-topology-details" data = { "tapi-topology:input": { "tapi-topology:topology-id-or-name": "otn-topology" } } - response = test_utils.post_request(url, data, test_utils.ODL_LOGIN, test_utils.ODL_PWD) + response = test_utils.post_request(url, data) self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200) res = response.json() self.assertEqual(len(res["output"]["topology"]["node"]), 4, 'There should be 4 nodes')