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