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 import unittest
  12 import json
  13 import os
  14 import sys
  15 import time
  16 import requests
  17 from common import test_utils
  18
  19
  20 class TransportPCEtesting(unittest.TestCase):
  21
  22     simple_topo_bi_dir_data = None
  23     simple_topo_uni_dir_data = None
  24     complex_topo_uni_dir_data = None
  25     processes = None
  26
  27     @classmethod
  28     def setUpClass(cls):
  29         try:
  30             sample_files_parsed = False
  31             TOPO_BI_DIR_FILE = os.path.join(os.path.dirname(os.path.realpath(__file__)),
  32                                             "..", "..", "sample_configs", "honeynode-topo.xml")
  33             with open(TOPO_BI_DIR_FILE, 'r') as topo_bi_dir:
  34                 cls.simple_topo_bi_dir_data = topo_bi_dir.read()
  35
  36             TOPO_UNI_DIR_FILE = os.path.join(os.path.dirname(os.path.realpath(__file__)),
  37                                              "..", "..", "sample_configs", "NW-simple-topology.xml")
  38
  39             with open(TOPO_UNI_DIR_FILE, 'r') as topo_uni_dir:
  40                 cls.simple_topo_uni_dir_data = topo_uni_dir.read()
  41
  42             TOPO_UNI_DIR_COMPLEX_FILE = os.path.join(os.path.dirname(os.path.realpath(__file__)),
  43                                                      "..", "..", "sample_configs", "NW-for-test-5-4.xml")
  44             with open(TOPO_UNI_DIR_COMPLEX_FILE, 'r') as topo_uni_dir_complex:
  45                 cls.complex_topo_uni_dir_data = topo_uni_dir_complex.read()
  46             sample_files_parsed = True
  47         except PermissionError as err:
  48             print("Permission Error when trying to read sample files\n", err)
  49             sys.exit(2)
  50         except FileNotFoundError as err:
  51             print("File Not found Error when trying to read sample files\n", err)
  52             sys.exit(2)
  53         except:
  54             print("Unexpected error when trying to read sample files\n", sys.exc_info()[0])
  55             sys.exit(2)
  56         finally:
  57             if sample_files_parsed:
  58                 print("sample files content loaded")
  59
  60         cls.processes = test_utils.start_tpce()
  61
  62     @classmethod
  63     def tearDownClass(cls):
  64         for process in cls.processes:
  65             test_utils.shutdown_process(process)
  66         print("all processes killed")
  67
  68     def setUp(self):  # instruction executed before each test method
  69         time.sleep(1)
  70
  71      # Load simple bidirectional topology
  72     def test_01_load_simple_topology_bi(self):
  73         response = test_utils.put_xmlrequest(test_utils.URL_CONFIG_ORDM_TOPO, self.simple_topo_bi_dir_data)
  74         self.assertEqual(response.status_code, requests.codes.ok)
  75         time.sleep(2)
  76
  77     # Get existing nodeId
  78     def test_02_get_nodeId(self):
  79         response = test_utils.get_ordm_topo_request("node/ROADMA01-SRG1")
  80         self.assertEqual(response.status_code, requests.codes.ok)
  81         res = response.json()
  82         self.assertEqual(
  83             res['node'][0]['node-id'], 'ROADMA01-SRG1')
  84         time.sleep(1)
  85
  86     # Get existing linkId
  87     def test_03_get_linkId(self):
  88         response = test_utils.get_ordm_topo_request("link/XPDRA01-XPDR1-XPDR1-NETWORK1toROADMA01-SRG1-SRG1-PP1-TXRX")
  89         self.assertEqual(response.status_code, requests.codes.ok)
  90         res = response.json()
  91         self.assertEqual(
  92             res['ietf-network-topology:link'][0]['link-id'],
  93             'XPDRA01-XPDR1-XPDR1-NETWORK1toROADMA01-SRG1-SRG1-PP1-TXRX')
  94         time.sleep(1)
  95
  96     # Path Computation success
  97     def test_04_path_computation_xpdr_bi(self):
  98         url = "{}/operations/transportpce-pce:path-computation-request"
  99         data = {"input": {
 100                 "service-name": "service-1",
 101                 "resource-reserve": "true",
 102                 "pce-metric": "hop-count",
 103                 "service-handler-header": {
 104                     "request-id": "request-1"
 105                 },
 106                 "service-a-end": {
 107                     "node-id": "XPDRA01",
 108                     "service-rate": "100",
 109                     "service-format": "Ethernet",
 110                     "clli": "nodeA"
 111                 },
 112                 "service-z-end": {
 113                     "node-id": "XPDRC01",
 114                     "service-rate": "100",
 115                     "service-format": "Ethernet",
 116                     "clli": "nodeC"
 117                 }
 118                 }
 119                 }
 120         response = test_utils.post_request(url, data)
 121         self.assertEqual(response.status_code, requests.codes.ok)
 122         res = response.json()
 123         self.assertIn('Path is calculated',
 124                       res['output']['configuration-response-common']['response-message'])
 125         time.sleep(5)
 126
 127     # Path Computation success
 128     def test_05_path_computation_rdm_bi(self):
 129         url = "{}/operations/transportpce-pce:path-computation-request"
 130         data = {"input": {
 131                 "service-name": "service-1",
 132                 "resource-reserve": "true",
 133                 "pce-metric": "hop-count",
 134                 "service-handler-header": {
 135                     "request-id": "request-1"
 136                 },
 137                 "service-a-end": {
 138                     "node-id": "ROADMA01",
 139                     "service-rate": "100",
 140                     "service-format": "Ethernet",
 141                     "clli": "NodeA"
 142                 },
 143                 "service-z-end": {
 144                     "node-id": "ROADMC01",
 145                     "service-rate": "100",
 146                     "service-format": "Ethernet",
 147                     "clli": "NodeC"
 148                 }
 149                 }
 150                 }
 151         response = test_utils.post_request(url, data)
 152         self.assertEqual(response.status_code, requests.codes.ok)
 153         res = response.json()
 154         self.assertIn('Path is calculated',
 155                       res['output']['configuration-response-common']['response-message'])
 156         time.sleep(5)
 157
 158     # Delete topology
 159     def test_06_delete_simple_topology_bi(self):
 160         response = test_utils.delete_request(test_utils.URL_CONFIG_ORDM_TOPO)
 161         self.assertEqual(response.status_code, requests.codes.ok)
 162         time.sleep(2)
 163
 164     # Test deleted topology
 165     def test_07_test_topology_simple_bi_deleted(self):
 166         response = test_utils.get_ordm_topo_request("node/ROADMA01-SRG1")
 167         self.assertEqual(response.status_code, 404)
 168         time.sleep(1)
 169
 170     # Load simple bidirectional topology
 171     def test_08_load_simple_topology_uni(self):
 172         response = test_utils.put_xmlrequest(test_utils.URL_CONFIG_ORDM_TOPO, self.simple_topo_uni_dir_data)
 173         self.assertEqual(response.status_code, 201)
 174         time.sleep(2)
 175
 176     # Get existing nodeId
 177     def test_09_get_nodeId(self):
 178         response = test_utils.get_ordm_topo_request("node/XPONDER-1-2")
 179         self.assertEqual(response.status_code, requests.codes.ok)
 180         res = response.json()
 181         self.assertEqual(
 182             res['node'][0]['node-id'],
 183             'XPONDER-1-2')
 184         time.sleep(1)
 185
 186     # Get existing linkId
 187     def test_10_get_linkId(self):
 188         response = test_utils.get_ordm_topo_request("link/XPONDER-1-2XPDR-NW1-TX-toOpenROADM-1-2-SRG1-SRG1-PP1-RX")
 189         self.assertEqual(response.status_code, requests.codes.ok)
 190         res = response.json()
 191         self.assertEqual(
 192             res['ietf-network-topology:link'][0]['link-id'],
 193             'XPONDER-1-2XPDR-NW1-TX-toOpenROADM-1-2-SRG1-SRG1-PP1-RX')
 194         time.sleep(1)
 195
 196     # Path Computation success
 197     def test_11_path_computation_xpdr_uni(self):
 198         url = "{}/operations/transportpce-pce:path-computation-request"
 199         data = {"input": {
 200                 "service-name": "service-1",
 201                 "resource-reserve": "true",
 202                 "pce-metric": "hop-count",
 203                 "service-handler-header": {
 204                     "request-id": "request-1"
 205                 },
 206                 "service-a-end": {
 207                     "node-id": "XPONDER-1-2",
 208                     "service-rate": "100",
 209                     "service-format": "Ethernet",
 210                     "clli": "ORANGE1"
 211                 },
 212                 "service-z-end": {
 213                     "node-id": "XPONDER-3-2",
 214                     "service-rate": "100",
 215                     "service-format": "Ethernet",
 216                     "clli": "ORANGE3"
 217                 }
 218                 }
 219                 }
 220         response = test_utils.post_request(url, data)
 221         self.assertEqual(response.status_code, requests.codes.ok)
 222         res = response.json()
 223         self.assertIn('Path is calculated',
 224                       res['output']['configuration-response-common']['response-message'])
 225         time.sleep(5)
 226
 227     # Path Computation success
 228     def test_12_path_computation_rdm_uni(self):
 229         url = "{}/operations/transportpce-pce:path-computation-request"
 230         data = {"input": {
 231                 "service-name": "service1",
 232                 "resource-reserve": "true",
 233                 "service-handler-header": {
 234                     "request-id": "request1"
 235                 },
 236                 "service-a-end": {
 237                     "service-rate": "100",
 238                     "service-format": "Ethernet",
 239                     "clli": "cll21",
 240                     "node-id": "OpenROADM-2-1"
 241                 },
 242                 "service-z-end": {
 243                     "service-rate": "100",
 244                     "service-format": "Ethernet",
 245                     "clli": "ncli22",
 246                     "node-id": "OpenROADM-2-2"
 247                 },
 248                 "pce-metric": "hop-count"
 249                 }
 250                 }
 251         response = test_utils.post_request(url, data)
 252         self.assertEqual(response.status_code, requests.codes.ok)
 253         res = response.json()
 254         self.assertIn('Path is calculated',
 255                       res['output']['configuration-response-common']['response-message'])
 256         # ZtoA path test
 257         atozList = len(res['output']['response-parameters']['path-description']['aToZ-direction']['aToZ'])
 258         ztoaList = len(res['output']['response-parameters']['path-description']['zToA-direction']['zToA'])
 259         self.assertEqual(atozList, 15)
 260         self.assertEqual(ztoaList, 15)
 261         for i in range(0, 15):
 262             atoz = res['output']['response-parameters']['path-description']['aToZ-direction']['aToZ'][i]
 263             ztoa = res['output']['response-parameters']['path-description']['zToA-direction']['zToA'][i]
 264             if (atoz['id'] == '14'):
 265                 self.assertEqual(atoz['resource']['tp-id'], 'SRG1-PP1-TX')
 266             if (ztoa['id'] == '0'):
 267                 self.assertEqual(ztoa['resource']['tp-id'], 'SRG1-PP1-RX')
 268         time.sleep(5)
 269
 270     # Delete topology
 271     def test_13_delete_simple_topology(self):
 272         response = test_utils.delete_request(test_utils.URL_CONFIG_ORDM_TOPO)
 273         self.assertEqual(response.status_code, requests.codes.ok)
 274         time.sleep(2)
 275
 276     # Test deleted topology
 277     def test_14_test_topology_simple_deleted(self):
 278         response = test_utils.get_ordm_topo_request("node/XPONDER-1-2")
 279         self.assertEqual(response.status_code, 404)
 280         time.sleep(1)
 281
 282     # Load complex topology
 283     def test_15_load_complex_topology(self):
 284         response = test_utils.put_xmlrequest(test_utils.URL_CONFIG_ORDM_TOPO, self.complex_topo_uni_dir_data)
 285         self.assertEqual(response.status_code, 201)
 286         time.sleep(2)
 287
 288     # Get existing nodeId
 289     def test_16_get_nodeId(self):
 290         response = test_utils.get_ordm_topo_request("node/XPONDER-3-2")
 291         self.assertEqual(response.status_code, requests.codes.ok)
 292         res = response.json()
 293         self.assertEqual(
 294             res['node'][0]['node-id'],
 295             'XPONDER-3-2')
 296         time.sleep(1)
 297
 298     # Test failed path computation
 299     def test_17_fail_path_computation(self):
 300         url = "{}/operations/transportpce-pce:path-computation-request"
 301         data = {"input": {
 302                 "service-handler-header": {
 303                     "request-id": "request-1"
 304                 }
 305                 }
 306                 }
 307         response = test_utils.post_request(url, data)
 308         self.assertEqual(response.status_code, requests.codes.ok)
 309         res = response.json()
 310         self.assertIn('Service Name is not set',
 311                       res['output']['configuration-response-common']['response-message'])
 312         time.sleep(2)
 313
 314     # Test1 success path computation
 315     def test_18_success1_path_computation(self):
 316         url = "{}/operations/transportpce-pce:path-computation-request"
 317         data = {"input": {
 318                 "service-name": "service1",
 319                 "resource-reserve": "true",
 320                 "service-handler-header": {
 321                     "request-id": "request1"
 322                 },
 323                 "service-a-end": {
 324                     "service-format": "Ethernet",
 325                     "service-rate": "100",
 326                     "clli": "ORANGE2",
 327                     "node-id": "XPONDER-2-2",
 328                     "tx-direction": {
 329                         "port": {
 330                             "port-device-name": "Some port-device-name",
 331                             "port-type": "Some port-type",
 332                             "port-name": "Some port-name",
 333                             "port-rack": "Some port-rack",
 334                             "port-shelf": "Some port-shelf",
 335                             "port-slot": "Some port-slot",
 336                             "port-sub-slot": "Some port-sub-slot"
 337                         }
 338                     },
 339                     "rx-direction": {
 340                         "port": {
 341                             "port-device-name": "Some port-device-name",
 342                             "port-type": "Some port-type",
 343                             "port-name": "Some port-name",
 344                             "port-rack": "Some port-rack",
 345                             "port-shelf": "Some port-shelf",
 346                             "port-slot": "Some port-slot",
 347                             "port-sub-slot": "Some port-sub-slot"
 348                         }
 349                     }
 350                 },
 351                 "service-z-end": {
 352                     "service-format": "Ethernet",
 353                     "service-rate": "100",
 354                     "clli": "ORANGE1",
 355                     "node-id": "XPONDER-1-2",
 356                     "tx-direction": {
 357                         "port": {
 358                             "port-device-name": "Some port-device-name",
 359                             "port-type": "Some port-type",
 360                             "port-name": "Some port-name",
 361                             "port-rack": "Some port-rack",
 362                             "port-shelf": "Some port-shelf",
 363                             "port-slot": "Some port-slot",
 364                             "port-sub-slot": "Some port-sub-slot"
 365                         }
 366                     },
 367                     "rx-direction": {
 368                         "port": {
 369                             "port-device-name": "Some port-device-name",
 370                             "port-type": "Some port-type",
 371                             "port-name": "Some port-name",
 372                             "port-rack": "Some port-rack",
 373                             "port-shelf": "Some port-shelf",
 374                             "port-slot": "Some port-slot",
 375                             "port-sub-slot": "Some port-sub-slot"
 376                         }
 377                     }
 378                 },
 379                 "hard-constraints": {
 380                     "customer-code": [
 381                         "Some customer-code"
 382                     ],
 383                     "co-routing": {
 384                         "existing-service": [
 385                             "Some existing-service"
 386                         ]
 387                     }
 388                 },
 389                 "soft-constraints": {
 390                     "customer-code": [
 391                         "Some customer-code"
 392                     ],
 393                     "co-routing": {
 394                         "existing-service": [
 395                             "Some existing-service"
 396                         ]
 397                     }
 398                 },
 399                 "pce-metric": "hop-count",
 400                 "locally-protected-links": "true"
 401                 }
 402                 }
 403         response = test_utils.post_request(url, data)
 404         self.assertEqual(response.status_code, requests.codes.ok)
 405         res = response.json()
 406         self.assertIn('Path is calculated',
 407                       res['output']['configuration-response-common']['response-message'])
 408         time.sleep(5)
 409
 410     # Test2 success path computation with path description
 411     def test_19_success2_path_computation(self):
 412         url = "{}/operations/transportpce-pce:path-computation-request"
 413         data = {"input": {
 414                 "service-name": "service 1",
 415                 "resource-reserve": "true",
 416                 "service-handler-header": {
 417                     "request-id": "request 1"
 418                 },
 419                 "service-a-end": {
 420                     "service-rate": "100",
 421                     "service-format": "Ethernet",
 422                     "node-id": "XPONDER-1-2",
 423                     "clli": "ORANGE1"
 424                 },
 425                 "service-z-end": {
 426                     "service-rate": "100",
 427                     "service-format": "Ethernet",
 428                     "node-id": "XPONDER-3-2",
 429                     "clli": "ORANGE3"
 430                 },
 431                 "pce-metric": "hop-count"
 432                 }
 433                 }
 434         response = test_utils.post_request(url, data)
 435         self.assertEqual(response.status_code, requests.codes.ok)
 436         res = response.json()
 437         self.assertIn('Path is calculated',
 438                       res['output']['configuration-response-common']['response-message'])
 439         self.assertEqual(5, res['output']['response-parameters']['path-description']
 440                          ['aToZ-direction']['aToZ-wavelength-number'])
 441         self.assertEqual(5, res['output']['response-parameters']['path-description']
 442                          ['zToA-direction']['zToA-wavelength-number'])
 443         time.sleep(5)
 444
 445     # Test3 success path computation with hard-constraints exclude
 446     def test_20_success3_path_computation(self):
 447         url = "{}/operations/transportpce-pce:path-computation-request"
 448         data = {"input": {
 449                 "service-name": "service 1",
 450                 "resource-reserve": "true",
 451                 "service-handler-header": {
 452                     "request-id": "request 1"
 453                 },
 454                 "service-a-end": {
 455                     "service-rate": "100",
 456                     "service-format": "Ethernet",
 457                     "node-id": "XPONDER-1-2",
 458                     "clli": "ORANGE1"
 459                 },
 460                 "service-z-end": {
 461                     "service-rate": "100",
 462                     "service-format": "Ethernet",
 463                     "node-id": "XPONDER-3-2",
 464                     "clli": "ORANGE3"
 465                 },
 466                 "hard-constraints": {
 467                     "exclude_": {
 468                         "node-id": ["OpenROADM-2-1", "OpenROADM-2-2"]
 469                     }
 470                 },
 471                 "pce-metric": "hop-count"
 472                 }
 473                 }
 474         response = test_utils.post_request(url, data)
 475         self.assertEqual(response.status_code, requests.codes.ok)
 476         res = response.json()
 477         self.assertIn('Path is calculated',
 478                       res['output']['configuration-response-common']['response-message'])
 479         self.assertEqual(9, res['output']['response-parameters']['path-description']
 480                          ['aToZ-direction']['aToZ-wavelength-number'])
 481         self.assertEqual(9, res['output']['response-parameters']['path-description']
 482                          ['zToA-direction']['zToA-wavelength-number'])
 483         time.sleep(5)
 484
 485     # Path computation before deleting oms-attribute of the link :openroadm1-3 to openroadm1-2
 486     def test_21_path_computation_before_oms_attribute_deletion(self):
 487         url = "{}/operations/transportpce-pce:path-computation-request"
 488         data = {"input": {
 489                 "service-name": "service 1",
 490                 "resource-reserve": "true",
 491                 "service-handler-header": {
 492                     "request-id": "request 1"
 493                 },
 494                 "service-a-end": {
 495                     "service-rate": "100",
 496                     "service-format": "Ethernet",
 497                     "node-id": "XPONDER-2-2",
 498                     "clli": "ORANGE2"
 499                 },
 500                 "service-z-end": {
 501                     "service-rate": "100",
 502                     "service-format": "Ethernet",
 503                     "node-id": "XPONDER-1-2",
 504                     "clli": "ORANGE1"
 505                 },
 506                 "pce-metric": "hop-count"
 507                 }
 508                 }
 509         response = test_utils.post_request(url, data)
 510         self.assertEqual(response.status_code, requests.codes.ok)
 511         res = response.json()
 512         self.assertIn('Path is calculated',
 513                       res['output']['configuration-response-common']['response-message'])
 514         nbElmPath = len(res['output']['response-parameters']['path-description']
 515                         ['aToZ-direction']['aToZ'])
 516         self.assertEqual(31, nbElmPath)
 517         link = {"link-id": "OpenROADM-1-3-DEG2-to-OpenROADM-1-2-DEG2"}
 518         find = False
 519         for i in range(0, nbElmPath):
 520             resource_i = res['output']['response-parameters']['path-description']['aToZ-direction']['aToZ'][i]['resource']
 521             if(resource_i == link):
 522                 find = True
 523         self.assertEqual(find, True)
 524         time.sleep(5)
 525
 526     # Delete oms-attribute in the link :openroadm1-3 to openroadm1-2
 527     def test_22_delete_oms_attribute_in_openroadm13toopenroadm12_link(self):
 528         response = test_utils.del_oms_attr_request("OpenROADM-1-3-DEG2-to-OpenROADM-1-2-DEG2")
 529         self.assertEqual(response.status_code, requests.codes.ok)
 530         time.sleep(2)
 531
 532     # Path computation after deleting oms-attribute of the link :openroadm1-3 to openroadm1-2
 533     def test_23_path_computation_after_oms_attribute_deletion(self):
 534         url = "{}/operations/transportpce-pce:path-computation-request"
 535         data = {"input": {
 536                 "service-name": "service 1",
 537                 "resource-reserve": "true",
 538                 "service-handler-header": {
 539                     "request-id": "request 1"
 540                 },
 541                 "service-a-end": {
 542                     "service-rate": "100",
 543                     "service-format": "Ethernet",
 544                     "node-id": "XPONDER-2-2",
 545                     "clli": "ORANGE2"
 546                 },
 547                 "service-z-end": {
 548                     "service-rate": "100",
 549                     "service-format": "Ethernet",
 550                     "node-id": "XPONDER-1-2",
 551                     "clli": "ORANGE1"
 552                 },
 553                 "pce-metric": "hop-count"
 554                 }
 555                 }
 556         response = test_utils.post_request(url, data)
 557         self.assertEqual(response.status_code, requests.codes.ok)
 558         res = response.json()
 559         self.assertIn('Path is calculated',
 560                       res['output']['configuration-response-common']['response-message'])
 561         nbElmPath = len(res['output']['response-parameters']['path-description']
 562                         ['aToZ-direction']['aToZ'])
 563         self.assertEqual(47, nbElmPath)
 564         link = {"link-id": "OpenROADM-1-3-DEG2-to-OpenROADM-1-2-DEG2"}
 565         find = False
 566         for i in range(0, nbElmPath):
 567             resource_i = res['output']['response-parameters']['path-description']['aToZ-direction']['aToZ'][i]['resource']
 568             if (resource_i == link):
 569                 find = True
 570         self.assertNotEqual(find, True)
 571         time.sleep(5)
 572
 573     # Delete complex topology
 574     def test_24_delete_complex_topology(self):
 575         response = test_utils.delete_request(test_utils.URL_CONFIG_ORDM_TOPO)
 576         self.assertEqual(response.status_code, requests.codes.ok)
 577         time.sleep(2)
 578
 579     # Test deleted complex topology
 580     def test_25_test_topology_complex_deleted(self):
 581         response = test_utils.get_ordm_topo_request("node/XPONDER-3-2")
 582         self.assertEqual(response.status_code, 404)
 583         time.sleep(1)
 584
 585
 586 if __name__ == "__main__":
 587     unittest.main(verbosity=2)