tests/transportpce_tests/2.2.1/test_tapi.py

   1 #!/usr/bin/env python
   2 ##############################################################################
   3 # Copyright (c) 2020 Orange, Inc. and others.  All rights reserved.
   4 #
   5 # All rights reserved. This program and the accompanying materials
   6 # are made available under the terms of the Apache License, Version 2.0
   7 # which accompanies this distribution, and is available at
   8 # http://www.apache.org/licenses/LICENSE-2.0
   9 ##############################################################################
  10 import os
  11 import re
  12 import time
  13 import unittest
  14
  15 import requests
  16
  17 import test_utils
  18
  19 RESTCONF_BASE_URL = "http://localhost:8181/restconf"
  20
  21 CODE_SHOULD_BE_200 = 'Http status code should be 200'
  22
  23 CODE_SHOULD_BE_201 = 'Http status code should be 201'
  24
  25 CREATED_SUCCESSFULLY = 'Result message should contain Xponder Roadm Link created successfully'
  26
  27
  28 class TransportTapitesting(unittest.TestCase):
  29     odl_process = None
  30     honeynode_process1 = None
  31     honeynode_process2 = None
  32     honeynode_process3 = None
  33     honeynode_process4 = None
  34     honeynode_process5 = None
  35
  36     # START_IGNORE_XTESTING
  37
  38     @classmethod
  39     @unittest.skipIf("USE_LIGHTY" in os.environ and os.environ['USE_LIGHTY'] == 'True',
  40                      "not supported for lighty")
  41     def setUpClass(cls):
  42         cls.init_failed = False
  43         karaf_log = os.path.join(
  44             os.path.dirname(os.path.realpath(__file__)),
  45             "..", "..", "..", "karaf", "target", "assembly", "data", "log", "karaf.log")
  46         searched_expr = re.escape("Blueprint container for bundle "
  47                                   "org.opendaylight.netconf.restconf") + ".* was successfully created"
  48
  49         print("starting opendaylight...")
  50         cls.odl_process = test_utils.start_tpce()
  51         found = test_utils.wait_until_log_contains(karaf_log, searched_expr, time_to_wait=60)
  52         cls.init_failed = not found
  53         if not cls.init_failed:
  54             print("opendaylight started")
  55
  56             print("installing tapi feature...")
  57             result = test_utils.install_karaf_feature("odl-transportpce-tapi")
  58             if result.returncode != 0:
  59                 cls.init_failed = True
  60             print("Restarting opendaylight...")
  61             test_utils.shutdown_process(cls.odl_process)
  62             cls.odl_process = test_utils.start_tpce()
  63             found = test_utils.wait_until_log_contains(karaf_log, searched_expr, time_to_wait=60)
  64             cls.init_failed = not found
  65             if not cls.init_failed:
  66                 print("starting XPDRA...")
  67                 cls.honeynode_process1 = test_utils.start_xpdra_honeynode()
  68
  69                 print("starting ROADMA...")
  70                 cls.honeynode_process2 = test_utils.start_roadma_honeynode()
  71
  72                 print("starting ROADMC...")
  73                 cls.honeynode_process3 = test_utils.start_roadmc_honeynode()
  74
  75                 print("starting XPDRC...")
  76                 cls.honeynode_process4 = test_utils.start_xpdrc_honeynode()
  77
  78                 print("starting SPDRA...")
  79                 cls.honeynode_process5 = test_utils.start_spdra_honeynode()
  80                 print("all honeynodes started")
  81
  82     @classmethod
  83     def tearDownClass(cls):
  84         test_utils.shutdown_process(cls.odl_process)
  85         test_utils.shutdown_process(cls.honeynode_process1)
  86         test_utils.shutdown_process(cls.honeynode_process2)
  87         test_utils.shutdown_process(cls.honeynode_process3)
  88         test_utils.shutdown_process(cls.honeynode_process4)
  89         test_utils.shutdown_process(cls.honeynode_process5)
  90         print("all processes killed")
  91
  92     def setUp(self):  # instruction executed before each test method
  93         if self.init_failed:
  94             self.fail('Feature installation failed')
  95         print("execution of {}".format(self.id().split(".")[-1]))
  96
  97     # END_IGNORE_XTESTING
  98
  99     #  connect netconf devices
 100     def test_00_connect_spdr_sa1(self):
 101         url = ("{}/config/network-topology:"
 102                "network-topology/topology/topology-netconf/node/SPDR-SA1"
 103                .format(RESTCONF_BASE_URL))
 104         data = test_utils.generate_connect_data("SPDR-SA1", "17845")
 105         response = test_utils.put_request(url, data, 'admin', 'admin')
 106         self.assertEqual(response.status_code, requests.codes.created, CODE_SHOULD_BE_201)  # pylint: disable=no-member
 107         time.sleep(10)
 108
 109     def test_01_connect_xpdra(self):
 110         url = ("{}/config/network-topology:"
 111                "network-topology/topology/topology-netconf/node/XPDR-A1"
 112                .format(RESTCONF_BASE_URL))
 113         data = test_utils.generate_connect_data("XPDR-A1", "17840")
 114         response = test_utils.put_request(url, data, 'admin', 'admin')
 115         self.assertEqual(response.status_code, requests.codes.created, CODE_SHOULD_BE_201)  # pylint: disable=no-member
 116         time.sleep(10)
 117
 118     def test_02_connect_xpdrc(self):
 119         url = ("{}/config/network-topology:"
 120                "network-topology/topology/topology-netconf/node/XPDR-C1"
 121                .format(RESTCONF_BASE_URL))
 122         data = test_utils.generate_connect_data("XPDR-C1", "17844")
 123         response = test_utils.put_request(url, data, 'admin', 'admin')
 124         self.assertEqual(response.status_code, requests.codes.created, CODE_SHOULD_BE_201)  # pylint: disable=no-member
 125         time.sleep(10)
 126
 127     def test_03_connect_rdma(self):
 128         url = ("{}/config/network-topology:"
 129                "network-topology/topology/topology-netconf/node/ROADM-A1"
 130                .format(RESTCONF_BASE_URL))
 131         data = test_utils.generate_connect_data("ROADM-A1", "17841")
 132         response = test_utils.put_request(url, data, 'admin', 'admin')
 133         self.assertEqual(response.status_code, requests.codes.created, CODE_SHOULD_BE_201)  # pylint: disable=no-member
 134         time.sleep(20)
 135
 136     def test_04_connect_rdmc(self):
 137         url = ("{}/config/network-topology:"
 138                "network-topology/topology/topology-netconf/node/ROADM-C1"
 139                .format(RESTCONF_BASE_URL))
 140         data = test_utils.generate_connect_data("ROADM-C1", "17843")
 141         response = test_utils.put_request(url, data, 'admin', 'admin')
 142         self.assertEqual(response.status_code, requests.codes.created, CODE_SHOULD_BE_201)  # pylint: disable=no-member
 143         time.sleep(20)
 144
 145     def test_05_connect_xprda_n1_to_roadma_pp1(self):
 146         url = "{}/operations/transportpce-networkutils:init-xpdr-rdm-links".format(RESTCONF_BASE_URL)
 147         data = test_utils.generate_link_data("XPDR-A1", "1", "1", "ROADM-A1", "1", "SRG1-PP1-TXRX")
 148         response = test_utils.post_request(url, data, 'admin', 'admin')
 149         self.assertEqual(response.status_code, requests.codes.ok, CODE_SHOULD_BE_200)  # pylint: disable=no-member
 150         res = response.json()
 151         self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"],
 152                       CREATED_SUCCESSFULLY)
 153         time.sleep(2)
 154
 155     def test_06_connect_roadma_pp1_to_xpdra_n1(self):
 156         url = "{}/operations/transportpce-networkutils:init-rdm-xpdr-links".format(RESTCONF_BASE_URL)
 157         data = test_utils.generate_link_data("XPDR-A1", "1", "1", "ROADM-A1", "1", "SRG1-PP1-TXRX")
 158         response = test_utils.post_request(url, data, 'admin', 'admin')
 159         self.assertEqual(response.status_code, requests.codes.ok, CODE_SHOULD_BE_200)  # pylint: disable=no-member
 160         res = response.json()
 161         self.assertIn('Roadm Xponder links created successfully', res["output"]["result"],
 162                       CREATED_SUCCESSFULLY)
 163         time.sleep(2)
 164
 165     def test_07_connect_xprdc_n1_to_roadmc_pp1(self):
 166         url = "{}/operations/transportpce-networkutils:init-xpdr-rdm-links".format(RESTCONF_BASE_URL)
 167         data = test_utils.generate_link_data("XPDR-C1", "1", "1", "ROADM-C1", "1", "SRG1-PP1-TXRX")
 168         response = test_utils.post_request(url, data, 'admin', 'admin')
 169         self.assertEqual(response.status_code, requests.codes.ok, CODE_SHOULD_BE_200)  # pylint: disable=no-member
 170         res = response.json()
 171         self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"],
 172                       CREATED_SUCCESSFULLY)
 173         time.sleep(2)
 174
 175     def test_08_connect_roadmc_pp1_to_xpdrc_n1(self):
 176         url = "{}/operations/transportpce-networkutils:init-rdm-xpdr-links".format(RESTCONF_BASE_URL)
 177         data = test_utils.generate_link_data("XPDR-C1", "1", "1", "ROADM-C1", "1", "SRG1-PP1-TXRX")
 178         response = test_utils.post_request(url, data, 'admin', 'admin')
 179         self.assertEqual(response.status_code, requests.codes.ok, CODE_SHOULD_BE_200)  # pylint: disable=no-member
 180         res = response.json()
 181         self.assertIn('Roadm Xponder links created successfully', res["output"]["result"],
 182                       CREATED_SUCCESSFULLY)
 183         time.sleep(2)
 184
 185     def test_09_connect_xprda_n2_to_roadma_pp2(self):
 186         url = "{}/operations/transportpce-networkutils:init-xpdr-rdm-links".format(RESTCONF_BASE_URL)
 187         data = test_utils.generate_link_data("XPDR-A1", "1", "2", "ROADM-A1", "1", "SRG1-PP2-TXRX")
 188         response = test_utils.post_request(url, data, 'admin', 'admin')
 189         self.assertEqual(response.status_code, requests.codes.ok, CODE_SHOULD_BE_200)  # pylint: disable=no-member
 190         res = response.json()
 191         self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"],
 192                       CREATED_SUCCESSFULLY)
 193         time.sleep(2)
 194
 195     def test_10_connect_roadma_pp2_to_xpdra_n2(self):
 196         url = "{}/operations/transportpce-networkutils:init-rdm-xpdr-links".format(RESTCONF_BASE_URL)
 197         data = test_utils.generate_link_data("XPDR-A1", "1", "2", "ROADM-A1", "1", "SRG1-PP2-TXRX")
 198         response = test_utils.post_request(url, data, 'admin', 'admin')
 199         self.assertEqual(response.status_code, requests.codes.ok, CODE_SHOULD_BE_200)  # pylint: disable=no-member
 200         res = response.json()
 201         self.assertIn('Roadm Xponder links created successfully', res["output"]["result"],
 202                       CREATED_SUCCESSFULLY)
 203         time.sleep(2)
 204
 205     def test_11_connect_xprdc_n2_to_roadmc_pp2(self):
 206         url = "{}/operations/transportpce-networkutils:init-xpdr-rdm-links".format(RESTCONF_BASE_URL)
 207         data = test_utils.generate_link_data("XPDR-C1", "1", "2", "ROADM-C1", "1", "SRG1-PP2-TXRX")
 208         response = test_utils.post_request(url, data, 'admin', 'admin')
 209         self.assertEqual(response.status_code, requests.codes.ok, CODE_SHOULD_BE_200)  # pylint: disable=no-member
 210         res = response.json()
 211         self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"],
 212                       CREATED_SUCCESSFULLY)
 213         time.sleep(2)
 214
 215     def test_12_connect_roadmc_pp2_to_xpdrc_n2(self):
 216         url = "{}/operations/transportpce-networkutils:init-rdm-xpdr-links".format(RESTCONF_BASE_URL)
 217         data = test_utils.generate_link_data("XPDR-C1", "1", "2", "ROADM-C1", "1", "SRG1-PP2-TXRX")
 218         response = test_utils.post_request(url, data, 'admin', 'admin')
 219         self.assertEqual(response.status_code, requests.codes.ok, CODE_SHOULD_BE_200)  # pylint: disable=no-member
 220         res = response.json()
 221         self.assertIn('Roadm Xponder links created successfully', res["output"]["result"],
 222                       CREATED_SUCCESSFULLY)
 223         time.sleep(2)
 224
 225     def test_13_get_tapi_openroadm_topology(self):
 226         url = "{}/operations/tapi-topology:get-topology-details".format(RESTCONF_BASE_URL)
 227         data = {
 228             "tapi-topology:input": {
 229                 "tapi-topology:topology-id-or-name": "openroadm-topology"
 230             }
 231         }
 232
 233         response = test_utils.post_request(url, data, 'admin', 'admin')
 234         self.assertEqual(response.status_code, requests.codes.ok, CODE_SHOULD_BE_200)  # pylint: disable=no-member
 235         res = response.json()
 236         self.assertEqual(len(res["output"]["topology"]["node"]), 1, 'There should be 1 node')
 237         self.assertEqual(len(res["output"]["topology"]["node"][0]["owned-node-edge-point"]), 4,
 238                          'There should be 4 owned-node-edge-points')
 239
 240     def test_14_get_tapi_otn_topology(self):
 241         url = "{}/operations/tapi-topology:get-topology-details".format(RESTCONF_BASE_URL)
 242         data = {
 243             "tapi-topology:input": {
 244                 "tapi-topology:topology-id-or-name": "otn-topology"
 245             }
 246         }
 247
 248         response = test_utils.post_request(url, data, 'admin', 'admin')
 249         self.assertEqual(response.status_code, requests.codes.ok, CODE_SHOULD_BE_200)  # pylint: disable=no-member
 250         res = response.json()
 251         self.assertEqual(len(res["output"]["topology"]["node"]), 4, 'There should be 4 nodes')
 252         self.assertEqual(len(res["output"]["topology"]["link"]), 5, 'There should be 5 links')
 253         link_to_check = res["output"]["topology"]["link"][0]
 254         # get info from first link to do deeper check
 255         node1_uid = link_to_check["node-edge-point"][0]["node-uuid"]
 256         node2_uid = link_to_check["node-edge-point"][1]["node-uuid"]
 257         node_edge_point1_uid = link_to_check["node-edge-point"][0]["node-edge-point-uuid"]
 258         node_edge_point2_uid = link_to_check["node-edge-point"][1]["node-edge-point-uuid"]
 259         # get node associated to link info
 260         nodes = res["output"]["topology"]["node"]
 261         node1 = find_object_with_key(nodes, "uuid", node1_uid)
 262         self.assertIsNotNone(node1, 'Node with uuid ' + node1_uid + ' should not be null')
 263         node2 = find_object_with_key(nodes, "uuid", node2_uid)
 264         self.assertIsNotNone(node2, 'Node with uuid ' + node2_uid + ' should not be null')
 265         # get edge-point associated to nodes
 266         node1_edge_point = node1["owned-node-edge-point"]
 267         node2_edge_point = node2["owned-node-edge-point"]
 268         node_edge_point1 = find_object_with_key(node1_edge_point, "uuid", node_edge_point1_uid)
 269         self.assertIsNotNone(node_edge_point1, 'Node edge point  with uuid ' + node_edge_point1_uid + 'should not be '
 270                                                                                                       'null')
 271         node_edge_point2 = find_object_with_key(node2_edge_point, "uuid", node_edge_point2_uid)
 272         self.assertIsNotNone(node_edge_point2, 'Node edge point with uuid ' + node_edge_point2_uid + 'should not be '
 273                                                                                                      'null')
 274         self.assertEqual(len(node_edge_point1["name"]), 1, 'There should be 1 name')
 275         self.assertEqual(len(node_edge_point2["name"]), 1, 'There should be 1 name')
 276         if node_edge_point1["layer-protocol-name"] == 'ODU':
 277             self.assertIn('NodeEdgePoint_N', node_edge_point1["name"][0]["value-name"], 'Value name should be '
 278                           'NodeEdgePoint_NX')
 279         elif node_edge_point1["layer-protocol-name"] == 'PHOTONIC_MEDIA':
 280             self.assertIn('iNodeEdgePoint_', node_edge_point1["name"][0]["value-name"], 'Value name should be '
 281                           'iNodeEdgePoint_X')
 282         else:
 283             self.fail('Wrong layer protocol name')
 284
 285         if node_edge_point2["layer-protocol-name"] == 'ODU':
 286             self.assertIn('NodeEdgePoint_N', node_edge_point2["name"][0]["value-name"], 'Value name should be '
 287                           'NodeEdgePoint_NX')
 288         elif node_edge_point2["layer-protocol-name"] == 'PHOTONIC_MEDIA':
 289             self.assertIn('iNodeEdgePoint_', node_edge_point2["name"][0]["value-name"], 'Value name should be '
 290                           'iNodeEdgePoint_X')
 291         else:
 292             self.fail('Wrong layer protocol name')
 293
 294     def test_15_disconnect_xpdra(self):
 295         url = ("{}/config/network-topology:"
 296                "network-topology/topology/topology-netconf/node/XPDR-A1"
 297                .format(RESTCONF_BASE_URL))
 298
 299         response = test_utils.delete_request(url, 'admin', 'admin')
 300         self.assertEqual(response.status_code, requests.codes.ok, CODE_SHOULD_BE_200)  # pylint: disable=no-member
 301         time.sleep(10)
 302
 303     def test_16_disconnect_xpdrc(self):
 304         url = ("{}/config/network-topology:"
 305                "network-topology/topology/topology-netconf/node/XPDR-C1"
 306                .format(RESTCONF_BASE_URL))
 307
 308         response = test_utils.delete_request(url, 'admin', 'admin')
 309         self.assertEqual(response.status_code, requests.codes.ok, CODE_SHOULD_BE_200)  # pylint: disable=no-member
 310         time.sleep(10)
 311
 312     def test_17_disconnect_roadma(self):
 313         url = ("{}/config/network-topology:"
 314                "network-topology/topology/topology-netconf/node/ROADM-A1"
 315                .format(RESTCONF_BASE_URL))
 316
 317         response = test_utils.delete_request(url, 'admin', 'admin')
 318         self.assertEqual(response.status_code, requests.codes.ok, CODE_SHOULD_BE_200)  # pylint: disable=no-member
 319         time.sleep(10)
 320
 321     def test_18_disconnect_roadmc(self):
 322         url = ("{}/config/network-topology:"
 323                "network-topology/topology/topology-netconf/node/ROADM-C1"
 324                .format(RESTCONF_BASE_URL))
 325
 326         response = test_utils.delete_request(url, 'admin', 'admin')
 327         self.assertEqual(response.status_code, requests.codes.ok, CODE_SHOULD_BE_200)  # pylint: disable=no-member
 328         time.sleep(10)
 329
 330     def test_19_disconnect_spdr_sa1(self):
 331         url = ("{}/config/network-topology:"
 332                "network-topology/topology/topology-netconf/node/SPDR-SA1"
 333                .format(RESTCONF_BASE_URL))
 334         response = test_utils.delete_request(url, 'admin', 'admin')
 335         self.assertEqual(response.status_code, requests.codes.ok, CODE_SHOULD_BE_200)  # pylint: disable=no-member
 336
 337
 338 def find_object_with_key(list_dicts, key, value):
 339     for dict_ in list_dicts:
 340         if dict_[key] == value:
 341             return dict_
 342     return None
 343
 344
 345 if __name__ == "__main__":
 346     unittest.main(verbosity=2)