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