#!/usr/bin/env python ############################################################################## # Copyright (c) 2020 Orange, Inc. and others. All rights reserved. # # All rights reserved. This program and the accompanying materials # are made available under the terms of the Apache License, Version 2.0 # which accompanies this distribution, and is available at # http://www.apache.org/licenses/LICENSE-2.0 ############################################################################## import os import re 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' 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 @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") 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") @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) print("all processes killed") def setUp(self): # instruction executed before each test method if self.init_failed: 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 time.sleep(10) 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 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 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 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 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 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 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 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 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 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 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 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 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) 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 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) 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 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') link_to_check = res["output"]["topology"]["link"][0] # get info from first link to do deeper check node1_uid = link_to_check["node-edge-point"][0]["node-uuid"] node2_uid = link_to_check["node-edge-point"][1]["node-uuid"] node_edge_point1_uid = link_to_check["node-edge-point"][0]["node-edge-point-uuid"] node_edge_point2_uid = link_to_check["node-edge-point"][1]["node-edge-point-uuid"] # get node associated to link info nodes = res["output"]["topology"]["node"] node1 = find_object_with_key(nodes, "uuid", node1_uid) self.assertIsNotNone(node1, 'Node with uuid ' + node1_uid + ' should not be null') node2 = find_object_with_key(nodes, "uuid", node2_uid) self.assertIsNotNone(node2, 'Node with uuid ' + node2_uid + ' should not be null') # get edge-point associated to nodes node1_edge_point = node1["owned-node-edge-point"] node2_edge_point = node2["owned-node-edge-point"] node_edge_point1 = find_object_with_key(node1_edge_point, "uuid", node_edge_point1_uid) self.assertIsNotNone(node_edge_point1, 'Node edge point with uuid ' + node_edge_point1_uid + 'should not be ' 'null') node_edge_point2 = find_object_with_key(node2_edge_point, "uuid", node_edge_point2_uid) self.assertIsNotNone(node_edge_point2, 'Node edge point with uuid ' + node_edge_point2_uid + 'should not be ' 'null') self.assertEqual(len(node_edge_point1["name"]), 1, 'There should be 1 name') self.assertEqual(len(node_edge_point2["name"]), 1, 'There should be 1 name') if node_edge_point1["layer-protocol-name"] == 'ODU': self.assertIn('NodeEdgePoint_N', node_edge_point1["name"][0]["value-name"], 'Value name should be ' 'NodeEdgePoint_NX') elif node_edge_point1["layer-protocol-name"] == 'PHOTONIC_MEDIA': self.assertIn('iNodeEdgePoint_', node_edge_point1["name"][0]["value-name"], 'Value name should be ' 'iNodeEdgePoint_X') else: self.fail('Wrong layer protocol name') if node_edge_point2["layer-protocol-name"] == 'ODU': self.assertIn('NodeEdgePoint_N', node_edge_point2["name"][0]["value-name"], 'Value name should be ' 'NodeEdgePoint_NX') elif node_edge_point2["layer-protocol-name"] == 'PHOTONIC_MEDIA': self.assertIn('iNodeEdgePoint_', node_edge_point2["name"][0]["value-name"], 'Value name should be ' 'iNodeEdgePoint_X') else: 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 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 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 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 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 def find_object_with_key(list_dicts, key, value): for dict_ in list_dicts: if dict_[key] == value: return dict_ return None if __name__ == "__main__": unittest.main(verbosity=2)