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         # TAPI feature is not installed by default in Karaf
  44         if "USE_LIGHTY" not in os.environ or os.environ['USE_LIGHTY'] != 'True':
  45             print("installing tapi feature...")
  46             result = test_utils.install_karaf_feature("odl-transportpce-tapi")
  47             if result.returncode != 0:
  48                 cls.init_failed = True
  49             print("Restarting opendaylight...")
  50             test_utils.shutdown_process(cls.odl_process)
  51             cls.odl_process = test_utils.start_tpce()
  52             cls.init_failed = not test_utils.wait_until_log_contains(
  53                 test_utils.karaf_log, test_utils.KARAF_OK_START_MSG, time_to_wait=60)
  54         if cls.init_failed:
  55             print("tapi installaiton feature failed...")
  56             test_utils.shutdown_process(cls.odl_process)
  57             exit(2)
  58         cls.sim_process1 = test_utils.start_sim('xpdra')
  59         cls.sim_process2 = test_utils.start_sim('roadma')
  60         cls.sim_process3 = test_utils.start_sim('roadmc')
  61         cls.sim_process4 = test_utils.start_sim('xpdrc')
  62         cls.sim_process5 = test_utils.start_sim('spdrav2')
  63
  64     @classmethod
  65     def tearDownClass(cls):
  66         test_utils.shutdown_process(cls.odl_process)
  67         test_utils.shutdown_process(cls.sim_process1)
  68         test_utils.shutdown_process(cls.sim_process2)
  69         test_utils.shutdown_process(cls.sim_process3)
  70         test_utils.shutdown_process(cls.sim_process4)
  71         test_utils.shutdown_process(cls.sim_process5)
  72         print("all processes killed")
  73
  74     def setUp(self):  # instruction executed before each test method
  75         if self.init_failed:
  76             self.fail('Feature installation failed')
  77         print("execution of {}".format(self.id().split(".")[-1]))
  78
  79     # END_IGNORE_XTESTING
  80
  81     #  connect netconf devices
  82     def test_00_connect_spdr_sa1(self):
  83         url = ("{}/config/network-topology:"
  84                "network-topology/topology/topology-netconf/node/SPDR-SA1"
  85                .format(RESTCONF_BASE_URL))
  86         data = test_utils.generate_connect_data("SPDR-SA1", test_utils.sims['spdrav2']['port'])
  87         response = test_utils.put_request(url, data, 'admin', 'admin')
  88         self.assertEqual(response.status_code, requests.codes.created, CODE_SHOULD_BE_201)  # pylint: disable=no-member
  89         time.sleep(10)
  90
  91     def test_01_connect_xpdra(self):
  92         url = ("{}/config/network-topology:"
  93                "network-topology/topology/topology-netconf/node/XPDR-A1"
  94                .format(RESTCONF_BASE_URL))
  95         data = test_utils.generate_connect_data("XPDR-A1", test_utils.sims['xpdra']['port'])
  96         response = test_utils.put_request(url, data, 'admin', 'admin')
  97         self.assertEqual(response.status_code, requests.codes.created, CODE_SHOULD_BE_201)  # pylint: disable=no-member
  98         time.sleep(10)
  99
 100     def test_02_connect_xpdrc(self):
 101         url = ("{}/config/network-topology:"
 102                "network-topology/topology/topology-netconf/node/XPDR-C1"
 103                .format(RESTCONF_BASE_URL))
 104         data = test_utils.generate_connect_data("XPDR-C1", test_utils.sims['xpdrc']['port'])
 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_03_connect_rdma(self):
 110         url = ("{}/config/network-topology:"
 111                "network-topology/topology/topology-netconf/node/ROADM-A1"
 112                .format(RESTCONF_BASE_URL))
 113         data = test_utils.generate_connect_data("ROADM-A1", test_utils.sims['roadma']['port'])
 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(20)
 117
 118     def test_04_connect_rdmc(self):
 119         url = ("{}/config/network-topology:"
 120                "network-topology/topology/topology-netconf/node/ROADM-C1"
 121                .format(RESTCONF_BASE_URL))
 122         data = test_utils.generate_connect_data("ROADM-C1", test_utils.sims['roadmc']['port'])
 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(20)
 126
 127     def test_05_connect_xprda_n1_to_roadma_pp1(self):
 128         url = "{}/operations/transportpce-networkutils:init-xpdr-rdm-links".format(RESTCONF_BASE_URL)
 129         data = test_utils.generate_link_data("XPDR-A1", "1", "1", "ROADM-A1", "1", "SRG1-PP1-TXRX")
 130         response = test_utils.post_request(url, data, 'admin', 'admin')
 131         self.assertEqual(response.status_code, requests.codes.ok, CODE_SHOULD_BE_200)  # pylint: disable=no-member
 132         res = response.json()
 133         self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"],
 134                       CREATED_SUCCESSFULLY)
 135         time.sleep(2)
 136
 137     def test_06_connect_roadma_pp1_to_xpdra_n1(self):
 138         url = "{}/operations/transportpce-networkutils:init-rdm-xpdr-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('Roadm Xponder links created successfully', res["output"]["result"],
 144                       CREATED_SUCCESSFULLY)
 145         time.sleep(2)
 146
 147     def test_07_connect_xprdc_n1_to_roadmc_pp1(self):
 148         url = "{}/operations/transportpce-networkutils:init-xpdr-rdm-links".format(RESTCONF_BASE_URL)
 149         data = test_utils.generate_link_data("XPDR-C1", "1", "1", "ROADM-C1", "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('Xponder Roadm Link created successfully', res["output"]["result"],
 154                       CREATED_SUCCESSFULLY)
 155         time.sleep(2)
 156
 157     def test_08_connect_roadmc_pp1_to_xpdrc_n1(self):
 158         url = "{}/operations/transportpce-networkutils:init-rdm-xpdr-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('Roadm Xponder links created successfully', res["output"]["result"],
 164                       CREATED_SUCCESSFULLY)
 165         time.sleep(2)
 166
 167     def test_09_connect_xprda_n2_to_roadma_pp2(self):
 168         url = "{}/operations/transportpce-networkutils:init-xpdr-rdm-links".format(RESTCONF_BASE_URL)
 169         data = test_utils.generate_link_data("XPDR-A1", "1", "2", "ROADM-A1", "1", "SRG1-PP2-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('Xponder Roadm Link created successfully', res["output"]["result"],
 174                       CREATED_SUCCESSFULLY)
 175         time.sleep(2)
 176
 177     def test_10_connect_roadma_pp2_to_xpdra_n2(self):
 178         url = "{}/operations/transportpce-networkutils:init-rdm-xpdr-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('Roadm Xponder links created successfully', res["output"]["result"],
 184                       CREATED_SUCCESSFULLY)
 185         time.sleep(2)
 186
 187     def test_11_connect_xprdc_n2_to_roadmc_pp2(self):
 188         url = "{}/operations/transportpce-networkutils:init-xpdr-rdm-links".format(RESTCONF_BASE_URL)
 189         data = test_utils.generate_link_data("XPDR-C1", "1", "2", "ROADM-C1", "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('Xponder Roadm Link created successfully', res["output"]["result"],
 194                       CREATED_SUCCESSFULLY)
 195         time.sleep(2)
 196
 197     def test_12_connect_roadmc_pp2_to_xpdrc_n2(self):
 198         url = "{}/operations/transportpce-networkutils:init-rdm-xpdr-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('Roadm Xponder links created successfully', res["output"]["result"],
 204                       CREATED_SUCCESSFULLY)
 205         time.sleep(2)
 206
 207     def test_13_get_tapi_openroadm_topology(self):
 208         url = "{}/operations/tapi-topology:get-topology-details".format(RESTCONF_BASE_URL)
 209         data = {
 210             "tapi-topology:input": {
 211                 "tapi-topology:topology-id-or-name": "openroadm-topology"
 212             }
 213         }
 214
 215         response = test_utils.post_request(url, data, 'admin', 'admin')
 216         self.assertEqual(response.status_code, requests.codes.ok, CODE_SHOULD_BE_200)  # pylint: disable=no-member
 217         res = response.json()
 218         self.assertEqual(len(res["output"]["topology"]["node"]), 1, 'There should be 1 node')
 219         self.assertEqual(len(res["output"]["topology"]["node"][0]["owned-node-edge-point"]), 4,
 220                          'There should be 4 owned-node-edge-points')
 221
 222     def test_14_get_tapi_otn_topology(self):
 223         url = "{}/operations/tapi-topology:get-topology-details".format(RESTCONF_BASE_URL)
 224         data = {
 225             "tapi-topology:input": {
 226                 "tapi-topology:topology-id-or-name": "otn-topology"
 227             }
 228         }
 229
 230         response = test_utils.post_request(url, data, 'admin', 'admin')
 231         self.assertEqual(response.status_code, requests.codes.ok, CODE_SHOULD_BE_200)  # pylint: disable=no-member
 232         res = response.json()
 233         self.assertEqual(len(res["output"]["topology"]["node"]), 4, 'There should be 4 nodes')
 234         self.assertEqual(len(res["output"]["topology"]["link"]), 5, 'There should be 5 links')
 235         link_to_check = res["output"]["topology"]["link"][0]
 236         # get info from first link to do deeper check
 237         node1_uid = link_to_check["node-edge-point"][0]["node-uuid"]
 238         node2_uid = link_to_check["node-edge-point"][1]["node-uuid"]
 239         node_edge_point1_uid = link_to_check["node-edge-point"][0]["node-edge-point-uuid"]
 240         node_edge_point2_uid = link_to_check["node-edge-point"][1]["node-edge-point-uuid"]
 241         # get node associated to link info
 242         nodes = res["output"]["topology"]["node"]
 243         node1 = find_object_with_key(nodes, "uuid", node1_uid)
 244         self.assertIsNotNone(node1, 'Node with uuid ' + node1_uid + ' should not be null')
 245         node2 = find_object_with_key(nodes, "uuid", node2_uid)
 246         self.assertIsNotNone(node2, 'Node with uuid ' + node2_uid + ' should not be null')
 247         # get edge-point associated to nodes
 248         node1_edge_point = node1["owned-node-edge-point"]
 249         node2_edge_point = node2["owned-node-edge-point"]
 250         node_edge_point1 = find_object_with_key(node1_edge_point, "uuid", node_edge_point1_uid)
 251         self.assertIsNotNone(node_edge_point1, 'Node edge point  with uuid ' + node_edge_point1_uid + 'should not be '
 252                                                                                                       'null')
 253         node_edge_point2 = find_object_with_key(node2_edge_point, "uuid", node_edge_point2_uid)
 254         self.assertIsNotNone(node_edge_point2, 'Node edge point with uuid ' + node_edge_point2_uid + 'should not be '
 255                                                                                                      'null')
 256         self.assertEqual(len(node_edge_point1["name"]), 1, 'There should be 1 name')
 257         self.assertEqual(len(node_edge_point2["name"]), 1, 'There should be 1 name')
 258         if node_edge_point1["layer-protocol-name"] == 'ODU':
 259             self.assertIn('NodeEdgePoint_N', node_edge_point1["name"][0]["value-name"], 'Value name should be '
 260                           'NodeEdgePoint_NX')
 261         elif node_edge_point1["layer-protocol-name"] == 'PHOTONIC_MEDIA':
 262             self.assertIn('iNodeEdgePoint_', node_edge_point1["name"][0]["value-name"], 'Value name should be '
 263                           'iNodeEdgePoint_X')
 264         else:
 265             self.fail('Wrong layer protocol name')
 266
 267         if node_edge_point2["layer-protocol-name"] == 'ODU':
 268             self.assertIn('NodeEdgePoint_N', node_edge_point2["name"][0]["value-name"], 'Value name should be '
 269                           'NodeEdgePoint_NX')
 270         elif node_edge_point2["layer-protocol-name"] == 'PHOTONIC_MEDIA':
 271             self.assertIn('iNodeEdgePoint_', node_edge_point2["name"][0]["value-name"], 'Value name should be '
 272                           'iNodeEdgePoint_X')
 273         else:
 274             self.fail('Wrong layer protocol name')
 275
 276     def test_15_disconnect_xpdra(self):
 277         url = ("{}/config/network-topology:"
 278                "network-topology/topology/topology-netconf/node/XPDR-A1"
 279                .format(RESTCONF_BASE_URL))
 280
 281         response = test_utils.delete_request(url, 'admin', 'admin')
 282         self.assertEqual(response.status_code, requests.codes.ok, CODE_SHOULD_BE_200)  # pylint: disable=no-member
 283         time.sleep(10)
 284
 285     def test_16_disconnect_xpdrc(self):
 286         url = ("{}/config/network-topology:"
 287                "network-topology/topology/topology-netconf/node/XPDR-C1"
 288                .format(RESTCONF_BASE_URL))
 289
 290         response = test_utils.delete_request(url, 'admin', 'admin')
 291         self.assertEqual(response.status_code, requests.codes.ok, CODE_SHOULD_BE_200)  # pylint: disable=no-member
 292         time.sleep(10)
 293
 294     def test_17_disconnect_roadma(self):
 295         url = ("{}/config/network-topology:"
 296                "network-topology/topology/topology-netconf/node/ROADM-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_18_disconnect_roadmc(self):
 304         url = ("{}/config/network-topology:"
 305                "network-topology/topology/topology-netconf/node/ROADM-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_19_disconnect_spdr_sa1(self):
 313         url = ("{}/config/network-topology:"
 314                "network-topology/topology/topology-netconf/node/SPDR-SA1"
 315                .format(RESTCONF_BASE_URL))
 316         response = test_utils.delete_request(url, 'admin', 'admin')
 317         self.assertEqual(response.status_code, requests.codes.ok, CODE_SHOULD_BE_200)  # pylint: disable=no-member
 318
 319
 320 def find_object_with_key(list_dicts, key, value):
 321     for dict_ in list_dicts:
 322         if dict_[key] == value:
 323             return dict_
 324     return None
 325
 326
 327 if __name__ == "__main__":
 328     unittest.main(verbosity=2)