tests/transportpce_tests/1.2.1/test_pce.py

   1 #!/usr/bin/env python
   2 ##############################################################################
   3 # Copyright (c) 2017 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 unittest
  15 import json
  16 import os
  17 import sys
  18 import time
  19 import requests
  20 from common import test_utils
  21
  22
  23 class TransportPCEtesting(unittest.TestCase):
  24
  25     simple_topo_bi_dir_data = None
  26     simple_topo_uni_dir_data = None
  27     complex_topo_uni_dir_data = None
  28     processes = None
  29
  30     @classmethod
  31     def setUpClass(cls):
  32         try:
  33             sample_files_parsed = False
  34             TOPO_BI_DIR_FILE = os.path.join(os.path.dirname(os.path.realpath(__file__)),
  35                                             "..", "..", "sample_configs", "honeynode-topo.xml")
  36             with open(TOPO_BI_DIR_FILE, 'r') as topo_bi_dir:
  37                 cls.simple_topo_bi_dir_data = topo_bi_dir.read()
  38
  39             TOPO_UNI_DIR_FILE = os.path.join(os.path.dirname(os.path.realpath(__file__)),
  40                                              "..", "..", "sample_configs", "NW-simple-topology.xml")
  41
  42             with open(TOPO_UNI_DIR_FILE, 'r') as topo_uni_dir:
  43                 cls.simple_topo_uni_dir_data = topo_uni_dir.read()
  44
  45             TOPO_UNI_DIR_COMPLEX_FILE = os.path.join(os.path.dirname(os.path.realpath(__file__)),
  46                                                      "..", "..", "sample_configs", "NW-for-test-5-4.xml")
  47             with open(TOPO_UNI_DIR_COMPLEX_FILE, 'r') as topo_uni_dir_complex:
  48                 cls.complex_topo_uni_dir_data = topo_uni_dir_complex.read()
  49             sample_files_parsed = True
  50         except PermissionError as err:
  51             print("Permission Error when trying to read sample files\n", err)
  52             sys.exit(2)
  53         except FileNotFoundError as err:
  54             print("File Not found Error when trying to read sample files\n", err)
  55             sys.exit(2)
  56         except:
  57             print("Unexpected error when trying to read sample files\n", sys.exc_info()[0])
  58             sys.exit(2)
  59         finally:
  60             if sample_files_parsed:
  61                 print("sample files content loaded")
  62
  63         cls.processes = test_utils.start_tpce()
  64
  65     @classmethod
  66     def tearDownClass(cls):
  67         for process in cls.processes:
  68             test_utils.shutdown_process(process)
  69         print("all processes killed")
  70
  71     def setUp(self):  # instruction executed before each test method
  72         time.sleep(1)
  73
  74      # Load simple bidirectional topology
  75     def test_01_load_simple_topology_bi(self):
  76         response = test_utils.put_xmlrequest(test_utils.URL_CONFIG_ORDM_TOPO, self.simple_topo_bi_dir_data)
  77         self.assertEqual(response.status_code, requests.codes.ok)
  78         time.sleep(2)
  79
  80     # Get existing nodeId
  81     def test_02_get_nodeId(self):
  82         response = test_utils.get_ordm_topo_request("node/ROADMA01-SRG1")
  83         self.assertEqual(response.status_code, requests.codes.ok)
  84         res = response.json()
  85         self.assertEqual(
  86             res['node'][0]['node-id'], 'ROADMA01-SRG1')
  87         time.sleep(1)
  88
  89     # Get existing linkId
  90     def test_03_get_linkId(self):
  91         response = test_utils.get_ordm_topo_request("link/XPDRA01-XPDR1-XPDR1-NETWORK1toROADMA01-SRG1-SRG1-PP1-TXRX")
  92         self.assertEqual(response.status_code, requests.codes.ok)
  93         res = response.json()
  94         self.assertEqual(
  95             res['ietf-network-topology:link'][0]['link-id'],
  96             'XPDRA01-XPDR1-XPDR1-NETWORK1toROADMA01-SRG1-SRG1-PP1-TXRX')
  97         time.sleep(1)
  98
  99     # Path Computation success
 100     def test_04_path_computation_xpdr_bi(self):
 101         response = test_utils.path_computation_request("request-1", "service-1",
 102                                                        {"node-id": "XPDRA01", "service-rate": "100",
 103                                                            "service-format": "Ethernet", "clli": "nodeA"},
 104                                                        {"node-id": "XPDRC01", "service-rate": "100", "service-format": "Ethernet", "clli": "nodeC"})
 105         self.assertEqual(response.status_code, requests.codes.ok)
 106         res = response.json()
 107         self.assertIn('Path is calculated',
 108                       res['output']['configuration-response-common']['response-message'])
 109         time.sleep(5)
 110
 111     # Path Computation success
 112     def test_05_path_computation_rdm_bi(self):
 113         response = test_utils.path_computation_request("request-1", "service-1",
 114                                                        {"node-id": "ROADMA01", "service-rate": "100",
 115                                                            "service-format": "Ethernet", "clli": "NodeA"},
 116                                                        {"node-id": "ROADMC01", "service-rate": "100", "service-format": "Ethernet", "clli": "NodeC"})
 117         self.assertEqual(response.status_code, requests.codes.ok)
 118         res = response.json()
 119         self.assertIn('Path is calculated',
 120                       res['output']['configuration-response-common']['response-message'])
 121         time.sleep(5)
 122
 123     # Delete topology
 124     def test_06_delete_simple_topology_bi(self):
 125         response = test_utils.delete_request(test_utils.URL_CONFIG_ORDM_TOPO)
 126         self.assertEqual(response.status_code, requests.codes.ok)
 127         time.sleep(2)
 128
 129     # Test deleted topology
 130     def test_07_test_topology_simple_bi_deleted(self):
 131         response = test_utils.get_ordm_topo_request("node/ROADMA01-SRG1")
 132         self.assertEqual(response.status_code, requests.codes.conflict)
 133         time.sleep(1)
 134
 135     # Load simple bidirectional topology
 136     def test_08_load_simple_topology_uni(self):
 137         response = test_utils.put_xmlrequest(test_utils.URL_CONFIG_ORDM_TOPO, self.simple_topo_uni_dir_data)
 138         self.assertEqual(response.status_code, 201)
 139         time.sleep(2)
 140
 141     # Get existing nodeId
 142     def test_09_get_nodeId(self):
 143         response = test_utils.get_ordm_topo_request("node/XPONDER-1-2")
 144         self.assertEqual(response.status_code, requests.codes.ok)
 145         res = response.json()
 146         self.assertEqual(
 147             res['node'][0]['node-id'],
 148             'XPONDER-1-2')
 149         time.sleep(1)
 150
 151     # Get existing linkId
 152     def test_10_get_linkId(self):
 153         response = test_utils.get_ordm_topo_request("link/XPONDER-1-2XPDR-NW1-TX-toOpenROADM-1-2-SRG1-SRG1-PP1-RX")
 154         self.assertEqual(response.status_code, requests.codes.ok)
 155         res = response.json()
 156         self.assertEqual(
 157             res['ietf-network-topology:link'][0]['link-id'],
 158             'XPONDER-1-2XPDR-NW1-TX-toOpenROADM-1-2-SRG1-SRG1-PP1-RX')
 159         time.sleep(1)
 160
 161     # Path Computation success
 162     def test_11_path_computation_xpdr_uni(self):
 163         response = test_utils.path_computation_request("request-1", "service-1",
 164                                                        {"node-id": "XPONDER-1-2", "service-rate": "100",
 165                                                            "service-format": "Ethernet", "clli": "ORANGE1"},
 166                                                        {"node-id": "XPONDER-3-2", "service-rate": "100", "service-format": "Ethernet", "clli": "ORANGE3"})
 167         self.assertEqual(response.status_code, requests.codes.ok)
 168         res = response.json()
 169         self.assertIn('Path is calculated',
 170                       res['output']['configuration-response-common']['response-message'])
 171         time.sleep(5)
 172
 173     # Path Computation success
 174     def test_12_path_computation_rdm_uni(self):
 175         response = test_utils.path_computation_request("request1", "service1",
 176                                                        {"service-rate": "100", "service-format": "Ethernet",
 177                                                            "clli": "cll21", "node-id": "OpenROADM-2-1"},
 178                                                        {"service-rate": "100", "service-format": "Ethernet", "clli": "ncli22", "node-id": "OpenROADM-2-2"})
 179         self.assertEqual(response.status_code, requests.codes.ok)
 180         res = response.json()
 181         self.assertIn('Path is calculated',
 182                       res['output']['configuration-response-common']['response-message'])
 183         # ZtoA path test
 184         atozList = len(res['output']['response-parameters']['path-description']['aToZ-direction']['aToZ'])
 185         ztoaList = len(res['output']['response-parameters']['path-description']['zToA-direction']['zToA'])
 186         self.assertEqual(atozList, 15)
 187         self.assertEqual(ztoaList, 15)
 188         for i in range(0, 15):
 189             atoz = res['output']['response-parameters']['path-description']['aToZ-direction']['aToZ'][i]
 190             ztoa = res['output']['response-parameters']['path-description']['zToA-direction']['zToA'][i]
 191             if (atoz['id'] == '14'):
 192                 self.assertEqual(atoz['resource']['tp-id'], 'SRG1-PP1-TX')
 193             if (ztoa['id'] == '0'):
 194                 self.assertEqual(ztoa['resource']['tp-id'], 'SRG1-PP1-RX')
 195         time.sleep(5)
 196
 197     # Delete topology
 198     def test_13_delete_simple_topology(self):
 199         response = test_utils.delete_request(test_utils.URL_CONFIG_ORDM_TOPO)
 200         self.assertEqual(response.status_code, requests.codes.ok)
 201         time.sleep(2)
 202
 203     # Test deleted topology
 204     def test_14_test_topology_simple_deleted(self):
 205         response = test_utils.get_ordm_topo_request("node/XPONDER-1-2")
 206         self.assertEqual(response.status_code, requests.codes.conflict)
 207         time.sleep(1)
 208
 209     # Load complex topology
 210     def test_15_load_complex_topology(self):
 211         response = test_utils.put_xmlrequest(test_utils.URL_CONFIG_ORDM_TOPO, self.complex_topo_uni_dir_data)
 212         self.assertEqual(response.status_code, 201)
 213         time.sleep(2)
 214
 215     # Get existing nodeId
 216     def test_16_get_nodeId(self):
 217         response = test_utils.get_ordm_topo_request("node/XPONDER-3-2")
 218         self.assertEqual(response.status_code, requests.codes.ok)
 219         res = response.json()
 220         self.assertEqual(
 221             res['node'][0]['node-id'],
 222             'XPONDER-3-2')
 223         time.sleep(1)
 224
 225     # Test failed path computation
 226     def test_17_fail_path_computation(self):
 227         response = test_utils.post_request(test_utils.URL_PATH_COMPUTATION_REQUEST,
 228                                            {"input": {"service-handler-header": {"request-id": "request-1"}}})
 229         self.assertEqual(response.status_code, requests.codes.ok)
 230         res = response.json()
 231         self.assertIn('Service Name is not set',
 232                       res['output']['configuration-response-common']['response-message'])
 233         time.sleep(2)
 234
 235     # Test1 success path computation
 236     def test_18_success1_path_computation(self):
 237         response = test_utils.path_computation_request("request1", "service1",
 238                                                        {"service-format": "Ethernet", "service-rate": "100", "clli": "ORANGE2", "node-id": "XPONDER-2-2",
 239                                                         "tx-direction": {"port": {
 240                                                             "port-device-name": "Some port-device-name",
 241                                                             "port-type": "Some port-type",
 242                                                             "port-name": "Some port-name",
 243                                                             "port-rack": "Some port-rack",
 244                                                             "port-shelf": "Some port-shelf",
 245                                                             "port-slot": "Some port-slot",
 246                                                             "port-sub-slot": "Some port-sub-slot"
 247                                                         }},
 248                                                            "rx-direction": {"port": {
 249                                                                "port-device-name": "Some port-device-name",
 250                                                                "port-type": "Some port-type",
 251                                                                "port-name": "Some port-name",
 252                                                                "port-rack": "Some port-rack",
 253                                                                "port-shelf": "Some port-shelf",
 254                                                                "port-slot": "Some port-slot",
 255                                                                "port-sub-slot": "Some port-sub-slot"
 256                                                            }}},
 257                                                        {"service-format": "Ethernet", "service-rate": "100", "clli": "ORANGE1", "node-id": "XPONDER-1-2",
 258                                                            "tx-direction": {"port": {
 259                                                                "port-device-name": "Some port-device-name",
 260                                                                "port-type": "Some port-type",
 261                                                                "port-name": "Some port-name",
 262                                                                "port-rack": "Some port-rack",
 263                                                                "port-shelf": "Some port-shelf",
 264                                                                "port-slot": "Some port-slot",
 265                                                                "port-sub-slot": "Some port-sub-slot"
 266                                                            }},
 267                                                            "rx-direction": {"port": {
 268                                                                "port-device-name": "Some port-device-name",
 269                                                                "port-type": "Some port-type",
 270                                                                "port-name": "Some port-name",
 271                                                                "port-rack": "Some port-rack",
 272                                                                "port-shelf": "Some port-shelf",
 273                                                                "port-slot": "Some port-slot",
 274                                                                "port-sub-slot": "Some port-sub-slot"
 275                                                            }}},
 276                                                        {"customer-code": ["Some customer-code"],
 277                                                            "co-routing": {"existing-service": ["Some existing-service"]}},
 278                                                        {"customer-code": ["Some customer-code"],
 279                                                            "co-routing": {"existing-service": ["Some existing-service"]}},
 280                                                        "hop-count", {"locally-protected-links": "true"})
 281         self.assertEqual(response.status_code, requests.codes.ok)
 282         res = response.json()
 283         self.assertIn('Path is calculated',
 284                       res['output']['configuration-response-common']['response-message'])
 285         time.sleep(5)
 286
 287     # Test2 success path computation with path description
 288     def test_19_success2_path_computation(self):
 289         response = test_utils.path_computation_request("request 1", "service 1",
 290                                                        {"service-rate": "100", "service-format": "Ethernet",
 291                                                            "node-id": "XPONDER-1-2", "clli": "ORANGE1"},
 292                                                        {"service-rate": "100", "service-format": "Ethernet", "node-id": "XPONDER-3-2", "clli": "ORANGE3"})
 293         self.assertEqual(response.status_code, requests.codes.ok)
 294         res = response.json()
 295         self.assertIn('Path is calculated',
 296                       res['output']['configuration-response-common']['response-message'])
 297         self.assertEqual(5, res['output']['response-parameters']['path-description']
 298                          ['aToZ-direction']['aToZ-wavelength-number'])
 299         self.assertEqual(5, res['output']['response-parameters']['path-description']
 300                          ['zToA-direction']['zToA-wavelength-number'])
 301         time.sleep(5)
 302
 303     # Test3 success path computation with hard-constraints exclude
 304     def test_20_success3_path_computation(self):
 305         response = test_utils.path_computation_request("request 1", "service 1",
 306                                                        {"service-rate": "100", "service-format": "Ethernet",
 307                                                            "node-id": "XPONDER-1-2", "clli": "ORANGE1"},
 308                                                        {"service-rate": "100", "service-format": "Ethernet",
 309                                                            "node-id": "XPONDER-3-2", "clli": "ORANGE3"},
 310                                                        {"exclude_": {"node-id": ["OpenROADM-2-1", "OpenROADM-2-2"]}})
 311         self.assertEqual(response.status_code, requests.codes.ok)
 312         res = response.json()
 313         self.assertIn('Path is calculated',
 314                       res['output']['configuration-response-common']['response-message'])
 315         self.assertEqual(9, res['output']['response-parameters']['path-description']
 316                          ['aToZ-direction']['aToZ-wavelength-number'])
 317         self.assertEqual(9, res['output']['response-parameters']['path-description']
 318                          ['zToA-direction']['zToA-wavelength-number'])
 319         time.sleep(5)
 320
 321     # Path computation before deleting oms-attribute of the link :openroadm1-3 to openroadm1-2
 322     def test_21_path_computation_before_oms_attribute_deletion(self):
 323         response = test_utils.path_computation_request("request 1", "service 1",
 324                                                        {"service-rate": "100", "service-format": "Ethernet",
 325                                                            "node-id": "XPONDER-2-2", "clli": "ORANGE2"},
 326                                                        {"service-rate": "100", "service-format": "Ethernet", "node-id": "XPONDER-1-2", "clli": "ORANGE1"})
 327         self.assertEqual(response.status_code, requests.codes.ok)
 328         res = response.json()
 329         self.assertIn('Path is calculated',
 330                       res['output']['configuration-response-common']['response-message'])
 331         nbElmPath = len(res['output']['response-parameters']['path-description']
 332                         ['aToZ-direction']['aToZ'])
 333         self.assertEqual(31, nbElmPath)
 334         link = {"link-id": "OpenROADM-1-3-DEG2-to-OpenROADM-1-2-DEG2"}
 335         find = False
 336         for i in range(0, nbElmPath):
 337             resource_i = res['output']['response-parameters']['path-description']['aToZ-direction']['aToZ'][i]['resource']
 338             if(resource_i == link):
 339                 find = True
 340         self.assertEqual(find, True)
 341         time.sleep(5)
 342
 343     # Delete oms-attribute in the link :openroadm1-3 to openroadm1-2
 344     def test_22_delete_oms_attribute_in_openroadm13toopenroadm12_link(self):
 345         response = test_utils.del_oms_attr_request("OpenROADM-1-3-DEG2-to-OpenROADM-1-2-DEG2")
 346         self.assertEqual(response.status_code, requests.codes.ok)
 347         time.sleep(2)
 348
 349     # Path computation after deleting oms-attribute of the link :openroadm1-3 to openroadm1-2
 350     def test_23_path_computation_after_oms_attribute_deletion(self):
 351         response = test_utils.path_computation_request("request 1", "service 1",
 352                                                        {"service-rate": "100", "service-format": "Ethernet",
 353                                                            "node-id": "XPONDER-2-2", "clli": "ORANGE2"},
 354                                                        {"service-rate": "100", "service-format": "Ethernet", "node-id": "XPONDER-1-2", "clli": "ORANGE1"})
 355         self.assertEqual(response.status_code, requests.codes.ok)
 356         res = response.json()
 357         self.assertIn('Path is calculated',
 358                       res['output']['configuration-response-common']['response-message'])
 359         nbElmPath = len(res['output']['response-parameters']['path-description']
 360                         ['aToZ-direction']['aToZ'])
 361         self.assertEqual(47, nbElmPath)
 362         link = {"link-id": "OpenROADM-1-3-DEG2-to-OpenROADM-1-2-DEG2"}
 363         find = False
 364         for i in range(0, nbElmPath):
 365             resource_i = res['output']['response-parameters']['path-description']['aToZ-direction']['aToZ'][i]['resource']
 366             if (resource_i == link):
 367                 find = True
 368         self.assertNotEqual(find, True)
 369         time.sleep(5)
 370
 371     # Delete complex topology
 372     def test_24_delete_complex_topology(self):
 373         response = test_utils.delete_request(test_utils.URL_CONFIG_ORDM_TOPO)
 374         self.assertEqual(response.status_code, requests.codes.ok)
 375         time.sleep(2)
 376
 377     # Test deleted complex topology
 378     def test_25_test_topology_complex_deleted(self):
 379         response = test_utils.get_ordm_topo_request("node/XPONDER-3-2")
 380         self.assertEqual(response.status_code, requests.codes.conflict)
 381         time.sleep(1)
 382
 383
 384 if __name__ == "__main__":
 385     unittest.main(verbosity=2)