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