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
  11 # pylint: disable=no-member
  12 # pylint: disable=too-many-public-methods
  13 # some pylint false positives specific to tapi test
  14 # pylint: disable=unsubscriptable-object
  15 # pylint: disable=unsupported-assignment-operation
  16
  17 import os
  18 import sys
  19 import time
  20 import unittest
  21
  22 import requests
  23
  24 from common import test_utils
  25
  26
  27 CREATED_SUCCESSFULLY = 'Result message should contain Xponder Roadm Link created successfully'
  28
  29
  30 class TransportTapitesting(unittest.TestCase):
  31
  32     processes = None
  33
  34     @classmethod
  35     def setUpClass(cls):
  36         cls.init_failed = False
  37         os.environ['JAVA_MIN_MEM'] = '1024M'
  38         os.environ['JAVA_MAX_MEM'] = '4069M'
  39         cls.processes = test_utils.start_tpce()
  40         # TAPI feature is not installed by default in Karaf
  41         if "USE_LIGHTY" not in os.environ or os.environ['USE_LIGHTY'] != 'True':
  42             print("installing tapi feature...")
  43             result = test_utils.install_karaf_feature("odl-transportpce-tapi")
  44             if result.returncode != 0:
  45                 cls.init_failed = True
  46             print("Restarting OpenDaylight...")
  47             test_utils.shutdown_process(cls.processes[0])
  48             cls.processes[0] = test_utils.start_karaf()
  49             test_utils.process_list[0] = cls.processes[0]
  50             cls.init_failed = not test_utils.wait_until_log_contains(
  51                 test_utils.KARAF_LOG, test_utils.KARAF_OK_START_MSG, time_to_wait=60)
  52         if cls.init_failed:
  53             print("tapi installation feature failed...")
  54             test_utils.shutdown_process(cls.processes[0])
  55             sys.exit(2)
  56         cls.processes = test_utils.start_sims(['xpdra', 'roadma', 'roadmc', 'xpdrc', 'spdra'])
  57
  58     @classmethod
  59     def tearDownClass(cls):
  60         # pylint: disable=not-an-iterable
  61         for process in cls.processes:
  62             test_utils.shutdown_process(process)
  63         print("all processes killed")
  64
  65     def setUp(self):  # instruction executed before each test method
  66         if self.init_failed:
  67             self.fail('Feature installation failed')
  68         print("execution of {}".format(self.id().split(".")[-1]))
  69
  70     def test_00_connect_spdr_sa1(self):
  71         response = test_utils.mount_device("SPDR-SA1", 'spdra')
  72         self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
  73         time.sleep(10)
  74         # TODO replace connect and disconnect timers with test_utils.wait_until_log_contains
  75
  76     def test_01_connect_xpdra(self):
  77         response = test_utils.mount_device("XPDR-A1", 'xpdra')
  78         self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
  79         time.sleep(10)
  80
  81     def test_02_connect_xpdrc(self):
  82         response = test_utils.mount_device("XPDR-C1", 'xpdrc')
  83         self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
  84         time.sleep(10)
  85
  86     def test_03_connect_rdma(self):
  87         response = test_utils.mount_device("ROADM-A1", 'roadma')
  88         self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
  89         time.sleep(20)
  90
  91     def test_04_connect_rdmc(self):
  92         response = test_utils.mount_device("ROADM-C1", 'roadmc')
  93         self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
  94         time.sleep(20)
  95
  96     def test_05_connect_xprda_n1_to_roadma_pp1(self):
  97         response = test_utils.connect_xpdr_to_rdm_request("XPDR-A1", "1", "1",
  98                                                           "ROADM-A1", "1", "SRG1-PP1-TXRX")
  99         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 100         res = response.json()
 101         self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"],
 102                       CREATED_SUCCESSFULLY)
 103         time.sleep(2)
 104
 105     def test_06_connect_roadma_pp1_to_xpdra_n1(self):
 106         response = test_utils.connect_rdm_to_xpdr_request("XPDR-A1", "1", "1",
 107                                                           "ROADM-A1", "1", "SRG1-PP1-TXRX")
 108         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 109         res = response.json()
 110         self.assertIn('Roadm Xponder links created successfully', res["output"]["result"],
 111                       CREATED_SUCCESSFULLY)
 112         time.sleep(2)
 113
 114     def test_07_connect_xprdc_n1_to_roadmc_pp1(self):
 115         response = test_utils.connect_xpdr_to_rdm_request("XPDR-C1", "1", "1",
 116                                                           "ROADM-C1", "1", "SRG1-PP1-TXRX")
 117         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 118         res = response.json()
 119         self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"],
 120                       CREATED_SUCCESSFULLY)
 121         time.sleep(2)
 122
 123     def test_08_connect_roadmc_pp1_to_xpdrc_n1(self):
 124         response = test_utils.connect_rdm_to_xpdr_request("XPDR-C1", "1", "1",
 125                                                           "ROADM-C1", "1", "SRG1-PP1-TXRX")
 126         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 127         res = response.json()
 128         self.assertIn('Roadm Xponder links created successfully', res["output"]["result"],
 129                       CREATED_SUCCESSFULLY)
 130         time.sleep(2)
 131
 132     def test_09_connect_xprda_n2_to_roadma_pp2(self):
 133         response = test_utils.connect_xpdr_to_rdm_request("XPDR-A1", "1", "2",
 134                                                           "ROADM-A1", "1", "SRG1-PP2-TXRX")
 135         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 136         res = response.json()
 137         self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"],
 138                       CREATED_SUCCESSFULLY)
 139         time.sleep(2)
 140
 141     def test_10_connect_roadma_pp2_to_xpdra_n2(self):
 142         response = test_utils.connect_rdm_to_xpdr_request("XPDR-A1", "1", "2",
 143                                                           "ROADM-A1", "1", "SRG1-PP2-TXRX")
 144         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 145         res = response.json()
 146         self.assertIn('Roadm Xponder links created successfully', res["output"]["result"],
 147                       CREATED_SUCCESSFULLY)
 148         time.sleep(2)
 149
 150     def test_11_connect_xprdc_n2_to_roadmc_pp2(self):
 151         response = test_utils.connect_xpdr_to_rdm_request("XPDR-C1", "1", "2",
 152                                                           "ROADM-C1", "1", "SRG1-PP2-TXRX")
 153         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 154         res = response.json()
 155         self.assertIn('Xponder Roadm Link created successfully', res["output"]["result"],
 156                       CREATED_SUCCESSFULLY)
 157         time.sleep(2)
 158
 159     def test_12_connect_roadmc_pp2_to_xpdrc_n2(self):
 160         response = test_utils.connect_rdm_to_xpdr_request("XPDR-C1", "1", "2",
 161                                                           "ROADM-C1", "1", "SRG1-PP2-TXRX")
 162         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 163         res = response.json()
 164         self.assertIn('Roadm Xponder links created successfully', res["output"]["result"],
 165                       CREATED_SUCCESSFULLY)
 166         time.sleep(2)
 167
 168     def test_13_get_tapi_openroadm_topology(self):
 169         url = "{}/operations/tapi-topology:get-topology-details"
 170         data = {
 171             "tapi-topology:input": {
 172                 "tapi-topology:topology-id-or-name": "openroadm-topology"
 173             }
 174         }
 175
 176         response = test_utils.post_request(url, data)
 177         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 178         res = response.json()
 179         self.assertEqual(len(res["output"]["topology"]["node"]), 1, 'There should be 1 node')
 180         self.assertEqual(len(res["output"]["topology"]["node"][0]["owned-node-edge-point"]), 4,
 181                          'There should be 4 owned-node-edge-points')
 182
 183     def test_14_get_tapi_otn_topology(self):
 184         url = "{}/operations/tapi-topology:get-topology-details"
 185         data = {
 186             "tapi-topology:input": {
 187                 "tapi-topology:topology-id-or-name": "otn-topology"
 188             }
 189         }
 190
 191         response = test_utils.post_request(url, data)
 192         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 193         res = response.json()
 194         self.assertEqual(len(res["output"]["topology"]["node"]), 4, 'There should be 4 nodes')
 195         self.assertEqual(len(res["output"]["topology"]["link"]), 5, 'There should be 5 links')
 196         link_to_check = res["output"]["topology"]["link"][0]
 197         # get info from first link to do deeper check
 198         node1_uid = link_to_check["node-edge-point"][0]["node-uuid"]
 199         node2_uid = link_to_check["node-edge-point"][1]["node-uuid"]
 200         node_edge_point1_uid = link_to_check["node-edge-point"][0]["node-edge-point-uuid"]
 201         node_edge_point2_uid = link_to_check["node-edge-point"][1]["node-edge-point-uuid"]
 202         # get node associated to link info
 203         nodes = res["output"]["topology"]["node"]
 204         node1 = find_object_with_key(nodes, "uuid", node1_uid)
 205         self.assertIsNotNone(node1, 'Node with uuid ' + node1_uid + ' should not be null')
 206         node2 = find_object_with_key(nodes, "uuid", node2_uid)
 207         self.assertIsNotNone(node2, 'Node with uuid ' + node2_uid + ' should not be null')
 208         # get edge-point associated to nodes
 209         node1_edge_point = node1["owned-node-edge-point"]
 210         node2_edge_point = node2["owned-node-edge-point"]
 211         node_edge_point1 = find_object_with_key(node1_edge_point, "uuid", node_edge_point1_uid)
 212         self.assertIsNotNone(node_edge_point1, 'Node edge point  with uuid ' + node_edge_point1_uid + 'should not be '
 213                                                                                                       'null')
 214         node_edge_point2 = find_object_with_key(node2_edge_point, "uuid", node_edge_point2_uid)
 215         self.assertIsNotNone(node_edge_point2, 'Node edge point with uuid ' + node_edge_point2_uid + 'should not be '
 216                                                                                                      'null')
 217         self.assertEqual(len(node_edge_point1["name"]), 1, 'There should be 1 name')
 218         self.assertEqual(len(node_edge_point2["name"]), 1, 'There should be 1 name')
 219         if node_edge_point1["layer-protocol-name"] == 'ODU':
 220             self.assertIn('NodeEdgePoint_N', node_edge_point1["name"][0]["value-name"], 'Value name should be '
 221                           'NodeEdgePoint_NX')
 222         elif node_edge_point1["layer-protocol-name"] == 'PHOTONIC_MEDIA':
 223             self.assertIn('iNodeEdgePoint_', node_edge_point1["name"][0]["value-name"], 'Value name should be '
 224                           'iNodeEdgePoint_X')
 225         else:
 226             self.fail('Wrong layer protocol name')
 227
 228         if node_edge_point2["layer-protocol-name"] == 'ODU':
 229             self.assertIn('NodeEdgePoint_N', node_edge_point2["name"][0]["value-name"], 'Value name should be '
 230                           'NodeEdgePoint_NX')
 231         elif node_edge_point2["layer-protocol-name"] == 'PHOTONIC_MEDIA':
 232             self.assertIn('iNodeEdgePoint_', node_edge_point2["name"][0]["value-name"], 'Value name should be '
 233                           'iNodeEdgePoint_X')
 234         else:
 235             self.fail('Wrong layer protocol name')
 236
 237     def test_15_disconnect_xpdra(self):
 238         response = test_utils.unmount_device("XPDR-A1")
 239         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 240         time.sleep(10)
 241
 242     def test_16_disconnect_xpdrc(self):
 243         response = test_utils.unmount_device("XPDR-C1")
 244         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 245         time.sleep(10)
 246
 247     def test_17_disconnect_roadma(self):
 248         response = test_utils.unmount_device("ROADM-A1")
 249         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 250         time.sleep(10)
 251
 252     def test_18_disconnect_roadmc(self):
 253         response = test_utils.unmount_device("ROADM-C1")
 254         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 255         time.sleep(10)
 256
 257     def test_19_disconnect_spdr_sa1(self):
 258         response = test_utils.unmount_device("SPDR-SA1")
 259         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 260
 261
 262 def find_object_with_key(list_dicts, key, value):
 263     for dict_ in list_dicts:
 264         if dict_[key] == value:
 265             return dict_
 266     return None
 267
 268
 269 if __name__ == "__main__":
 270     unittest.main(verbosity=2)