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