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