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