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=159cfbd038ccea62064b63e4e8ced9d1038409cf;hpb=87051d0370d613dc84e25af7dc657431164374b2;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 159cfbd03..496a33e65 100644 --- a/tests/transportpce_tests/2.2.1/test_tapi.py +++ b/tests/transportpce_tests/2.2.1/test_tapi.py @@ -7,86 +7,59 @@ # 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 re +import sys import time import unittest import requests -import test_utils - -RESTCONF_BASE_URL = "http://localhost:8181/restconf" - -CODE_SHOULD_BE_200 = 'Http status code should be 200' +from common import test_utils -CODE_SHOULD_BE_201 = 'Http status code should be 201' CREATED_SUCCESSFULLY = 'Result message should contain Xponder Roadm Link created successfully' class TransportTapitesting(unittest.TestCase): - odl_process = None - honeynode_process1 = None - honeynode_process2 = None - honeynode_process3 = None - honeynode_process4 = None - honeynode_process5 = None - # START_IGNORE_XTESTING + processes = None @classmethod - @unittest.skipIf("USE_LIGHTY" in os.environ and os.environ['USE_LIGHTY'] == 'True', - "not supported for lighty") def setUpClass(cls): cls.init_failed = False - karaf_log = os.path.join( - os.path.dirname(os.path.realpath(__file__)), - "..", "..", "..", "karaf", "target", "assembly", "data", "log", "karaf.log") - searched_expr = re.escape("Blueprint container for bundle " - "org.opendaylight.netconf.restconf") + ".* was successfully created" - - print("starting opendaylight...") - cls.odl_process = test_utils.start_tpce() - found = test_utils.wait_until_log_contains(karaf_log, searched_expr, time_to_wait=60) - cls.init_failed = not found - if not cls.init_failed: - print("opendaylight started") - + 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': print("installing tapi feature...") result = test_utils.install_karaf_feature("odl-transportpce-tapi") if result.returncode != 0: cls.init_failed = True - print("Restarting opendaylight...") - test_utils.shutdown_process(cls.odl_process) - cls.odl_process = test_utils.start_tpce() - found = test_utils.wait_until_log_contains(karaf_log, searched_expr, time_to_wait=60) - cls.init_failed = not found - if not cls.init_failed: - print("starting XPDRA...") - cls.honeynode_process1 = test_utils.start_xpdra_honeynode() - - print("starting ROADMA...") - cls.honeynode_process2 = test_utils.start_roadma_honeynode() - - print("starting ROADMC...") - cls.honeynode_process3 = test_utils.start_roadmc_honeynode() - - print("starting XPDRC...") - cls.honeynode_process4 = test_utils.start_xpdrc_honeynode() - - print("starting SPDRA...") - cls.honeynode_process5 = test_utils.start_spdra_honeynode() - print("all honeynodes started") + print("Restarting OpenDaylight...") + test_utils.shutdown_process(cls.processes[0]) + 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 installation feature failed...") + test_utils.shutdown_process(cls.processes[0]) + sys.exit(2) + cls.processes = test_utils.start_sims(['xpdra', 'roadma', 'roadmc', 'xpdrc', 'spdra']) @classmethod def tearDownClass(cls): - test_utils.shutdown_process(cls.odl_process) - test_utils.shutdown_process(cls.honeynode_process1) - test_utils.shutdown_process(cls.honeynode_process2) - test_utils.shutdown_process(cls.honeynode_process3) - test_utils.shutdown_process(cls.honeynode_process4) - test_utils.shutdown_process(cls.honeynode_process5) + # 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 @@ -94,159 +67,129 @@ class TransportTapitesting(unittest.TestCase): self.fail('Feature installation failed') print("execution of {}".format(self.id().split(".")[-1])) - # END_IGNORE_XTESTING - - # connect netconf devices def test_00_connect_spdr_sa1(self): - url = ("{}/config/network-topology:" - "network-topology/topology/topology-netconf/node/SPDR-SA1" - .format(RESTCONF_BASE_URL)) - data = test_utils.generate_connect_data("SPDR-SA1", "17845") - response = test_utils.put_request(url, data, 'admin', 'admin') - self.assertEqual(response.status_code, requests.codes.created, CODE_SHOULD_BE_201) # pylint: disable=no-member + 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 def test_01_connect_xpdra(self): - url = ("{}/config/network-topology:" - "network-topology/topology/topology-netconf/node/XPDR-A1" - .format(RESTCONF_BASE_URL)) - data = test_utils.generate_connect_data("XPDR-A1", "17840") - response = test_utils.put_request(url, data, 'admin', 'admin') - self.assertEqual(response.status_code, requests.codes.created, CODE_SHOULD_BE_201) # pylint: disable=no-member + response = test_utils.mount_device("XPDR-A1", 'xpdra') + self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201) time.sleep(10) def test_02_connect_xpdrc(self): - url = ("{}/config/network-topology:" - "network-topology/topology/topology-netconf/node/XPDR-C1" - .format(RESTCONF_BASE_URL)) - data = test_utils.generate_connect_data("XPDR-C1", "17844") - response = test_utils.put_request(url, data, 'admin', 'admin') - self.assertEqual(response.status_code, requests.codes.created, CODE_SHOULD_BE_201) # pylint: disable=no-member + response = test_utils.mount_device("XPDR-C1", 'xpdrc') + self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201) time.sleep(10) def test_03_connect_rdma(self): - url = ("{}/config/network-topology:" - "network-topology/topology/topology-netconf/node/ROADM-A1" - .format(RESTCONF_BASE_URL)) - data = test_utils.generate_connect_data("ROADM-A1", "17841") - response = test_utils.put_request(url, data, 'admin', 'admin') - self.assertEqual(response.status_code, requests.codes.created, CODE_SHOULD_BE_201) # pylint: disable=no-member + response = test_utils.mount_device("ROADM-A1", 'roadma') + self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201) time.sleep(20) def test_04_connect_rdmc(self): - url = ("{}/config/network-topology:" - "network-topology/topology/topology-netconf/node/ROADM-C1" - .format(RESTCONF_BASE_URL)) - data = test_utils.generate_connect_data("ROADM-C1", "17843") - response = test_utils.put_request(url, data, 'admin', 'admin') - self.assertEqual(response.status_code, requests.codes.created, CODE_SHOULD_BE_201) # pylint: disable=no-member + response = test_utils.mount_device("ROADM-C1", 'roadmc') + self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201) time.sleep(20) def test_05_connect_xprda_n1_to_roadma_pp1(self): - url = "{}/operations/transportpce-networkutils:init-xpdr-rdm-links".format(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, 'admin', 'admin') - self.assertEqual(response.status_code, requests.codes.ok, CODE_SHOULD_BE_200) # pylint: disable=no-member + 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"], CREATED_SUCCESSFULLY) time.sleep(2) def test_06_connect_roadma_pp1_to_xpdra_n1(self): - url = "{}/operations/transportpce-networkutils:init-rdm-xpdr-links".format(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, 'admin', 'admin') - self.assertEqual(response.status_code, requests.codes.ok, CODE_SHOULD_BE_200) # pylint: disable=no-member + 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"], CREATED_SUCCESSFULLY) time.sleep(2) def test_07_connect_xprdc_n1_to_roadmc_pp1(self): - url = "{}/operations/transportpce-networkutils:init-xpdr-rdm-links".format(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, 'admin', 'admin') - self.assertEqual(response.status_code, requests.codes.ok, CODE_SHOULD_BE_200) # pylint: disable=no-member + 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"], CREATED_SUCCESSFULLY) time.sleep(2) def test_08_connect_roadmc_pp1_to_xpdrc_n1(self): - url = "{}/operations/transportpce-networkutils:init-rdm-xpdr-links".format(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, 'admin', 'admin') - self.assertEqual(response.status_code, requests.codes.ok, CODE_SHOULD_BE_200) # pylint: disable=no-member + 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"], CREATED_SUCCESSFULLY) time.sleep(2) def test_09_connect_xprda_n2_to_roadma_pp2(self): - url = "{}/operations/transportpce-networkutils:init-xpdr-rdm-links".format(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, 'admin', 'admin') - self.assertEqual(response.status_code, requests.codes.ok, CODE_SHOULD_BE_200) # pylint: disable=no-member + 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"], CREATED_SUCCESSFULLY) time.sleep(2) def test_10_connect_roadma_pp2_to_xpdra_n2(self): - url = "{}/operations/transportpce-networkutils:init-rdm-xpdr-links".format(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, 'admin', 'admin') - self.assertEqual(response.status_code, requests.codes.ok, CODE_SHOULD_BE_200) # pylint: disable=no-member + 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"], CREATED_SUCCESSFULLY) time.sleep(2) def test_11_connect_xprdc_n2_to_roadmc_pp2(self): - url = "{}/operations/transportpce-networkutils:init-xpdr-rdm-links".format(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, 'admin', 'admin') - self.assertEqual(response.status_code, requests.codes.ok, CODE_SHOULD_BE_200) # pylint: disable=no-member + 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"], CREATED_SUCCESSFULLY) time.sleep(2) def test_12_connect_roadmc_pp2_to_xpdrc_n2(self): - url = "{}/operations/transportpce-networkutils:init-rdm-xpdr-links".format(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, 'admin', 'admin') - self.assertEqual(response.status_code, requests.codes.ok, CODE_SHOULD_BE_200) # pylint: disable=no-member + 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"], CREATED_SUCCESSFULLY) time.sleep(2) def test_13_get_tapi_openroadm_topology(self): - url = "{}/operations/tapi-topology:get-topology-details".format(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, 'admin', 'admin') - self.assertEqual(response.status_code, requests.codes.ok, CODE_SHOULD_BE_200) # pylint: disable=no-member + 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') self.assertEqual(len(res["output"]["topology"]["node"][0]["owned-node-edge-point"]), 4, 'There should be 4 owned-node-edge-points') def test_14_get_tapi_otn_topology(self): - url = "{}/operations/tapi-topology:get-topology-details".format(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, 'admin', 'admin') - self.assertEqual(response.status_code, requests.codes.ok, CODE_SHOULD_BE_200) # pylint: disable=no-member + 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') self.assertEqual(len(res["output"]["topology"]["link"]), 5, 'There should be 5 links') @@ -292,47 +235,28 @@ class TransportTapitesting(unittest.TestCase): self.fail('Wrong layer protocol name') def test_15_disconnect_xpdra(self): - url = ("{}/config/network-topology:" - "network-topology/topology/topology-netconf/node/XPDR-A1" - .format(RESTCONF_BASE_URL)) - - response = test_utils.delete_request(url, 'admin', 'admin') - self.assertEqual(response.status_code, requests.codes.ok, CODE_SHOULD_BE_200) # pylint: disable=no-member + response = test_utils.unmount_device("XPDR-A1") + self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200) time.sleep(10) def test_16_disconnect_xpdrc(self): - url = ("{}/config/network-topology:" - "network-topology/topology/topology-netconf/node/XPDR-C1" - .format(RESTCONF_BASE_URL)) - - response = test_utils.delete_request(url, 'admin', 'admin') - self.assertEqual(response.status_code, requests.codes.ok, CODE_SHOULD_BE_200) # pylint: disable=no-member + response = test_utils.unmount_device("XPDR-C1") + self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200) time.sleep(10) def test_17_disconnect_roadma(self): - url = ("{}/config/network-topology:" - "network-topology/topology/topology-netconf/node/ROADM-A1" - .format(RESTCONF_BASE_URL)) - - response = test_utils.delete_request(url, 'admin', 'admin') - self.assertEqual(response.status_code, requests.codes.ok, CODE_SHOULD_BE_200) # pylint: disable=no-member + response = test_utils.unmount_device("ROADM-A1") + self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200) time.sleep(10) def test_18_disconnect_roadmc(self): - url = ("{}/config/network-topology:" - "network-topology/topology/topology-netconf/node/ROADM-C1" - .format(RESTCONF_BASE_URL)) - - response = test_utils.delete_request(url, 'admin', 'admin') - self.assertEqual(response.status_code, requests.codes.ok, CODE_SHOULD_BE_200) # pylint: disable=no-member + response = test_utils.unmount_device("ROADM-C1") + self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200) time.sleep(10) def test_19_disconnect_spdr_sa1(self): - url = ("{}/config/network-topology:" - "network-topology/topology/topology-netconf/node/SPDR-SA1" - .format(RESTCONF_BASE_URL)) - response = test_utils.delete_request(url, 'admin', 'admin') - self.assertEqual(response.status_code, requests.codes.ok, CODE_SHOULD_BE_200) # pylint: disable=no-member + response = test_utils.unmount_device("SPDR-SA1") + self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200) def find_object_with_key(list_dicts, key, value):