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