X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=tests%2Ftransportpce_tests%2F2.2.1%2Ftest_tapi.py;h=9467034383e8a8feb51d7db1c7c506b72975b072;hb=19a9408ee823754859467888a5d2823bfd8655ab;hp=ed2ec846f05c237b66ec11f40af0d126a78d9ac2;hpb=866ecc93841cc9601d7974071e288adc89dbacf1;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 ed2ec846f..946703438 100644 --- a/tests/transportpce_tests/2.2.1/test_tapi.py +++ b/tests/transportpce_tests/2.2.1/test_tapi.py @@ -8,83 +8,49 @@ # http://www.apache.org/licenses/LICENSE-2.0 ############################################################################## import os -import re +import sys import time import unittest import requests -import test_utils +from common import test_utils -RESTCONF_BASE_URL = "http://localhost:8181/restconf" - -CODE_SHOULD_BE_200 = 'Http status code should be 200' - -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 def setUpClass(cls): cls.init_failed = False - print("starting opendaylight...") - cls.odl_process = test_utils.start_tpce() + 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': - 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" - 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") - 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("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.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) + for process in cls.processes: + test_utils.shutdown_process(process) print("all processes killed") def setUp(self): # instruction executed before each test method @@ -92,159 +58,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') @@ -290,47 +226,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):