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