tests/transportpce_tests/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
  21
  22 class TransportPCEtesting(unittest.TestCase):
  23
  24     odl_process = None
  25     simple_data = None
  26     complex_data = None
  27     restconf_baseurl = "http://localhost:8181/restconf"
  28
  29     @classmethod
  30     def _get_file(cls):
  31         simple_topology_file = "sample_configs/NW-simple-topology.xml"
  32         if os.path.isfile(simple_topology_file):
  33             with open(simple_topology_file, 'r') as simple_file:
  34                 cls.simple_data = simple_file.read();
  35         complex_topology_file = "sample_configs/NW-for-test-5-4.xml"
  36         if os.path.isfile(complex_topology_file):
  37             with open(complex_topology_file, 'r') as complex_file:
  38                 cls.complex_data = complex_file.read();
  39
  40     @classmethod
  41     def __start_odl(cls):
  42         executable = "../karaf/target/assembly/bin/karaf"
  43         with open('odl.log', 'w') as outfile:
  44             cls.odl_process = subprocess.Popen(
  45                 ["bash", executable, "server"], stdout=outfile,
  46                 stdin=open(os.devnull))
  47
  48     @classmethod
  49     def setUpClass(cls):  # a class method called before tests in an individual class run.
  50         cls._get_file()
  51         cls.__start_odl()
  52         time.sleep(90)
  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 topology
  66     def test_01_load_simple_topology(self):
  67         url = ("{}/config/ietf-network:network/openroadm-topology"
  68               .format(self.restconf_baseurl))
  69         body = self.simple_data
  70         headers = {'content-type': 'application/xml',
  71         "Accept": "application/json"}
  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:network/openroadm-topology/node/XPONDER-1-2"
  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'],
  90             'XPONDER-1-2')
  91         time.sleep(1)
  92
  93     # Get existing linkId
  94     def test_03_get_linkId(self):
  95         url = ("{}/config/ietf-network:network/openroadm-topology/link/XPONDER-1-2XPDR-NW1-TX-toOpenROADM-1-2-SRG1-SRG1-PP1-RX"
  96               .format(self.restconf_baseurl))
  97         headers = {'content-type': 'application/json',
  98         "Accept": "application/json"}
  99         response = requests.request(
 100             "GET", url, headers=headers, auth=('admin', 'admin'))
 101         self.assertEqual(response.status_code, requests.codes.ok)
 102         res = response.json()
 103         self.assertEqual(
 104             res['ietf-network-topology:link'][0]['link-id'],
 105             'XPONDER-1-2XPDR-NW1-TX-toOpenROADM-1-2-SRG1-SRG1-PP1-RX')
 106         time.sleep(1)
 107
 108     # Path Computation success
 109     def test_04_path_computation(self):
 110         url = ("{}/operations/pce:path-computation-request"
 111               .format(self.restconf_baseurl))
 112         body = {"input": {
 113                 "service-name": "service-1",
 114                 "resource-reserve": "true",
 115                 "pce-metric": "hop-count",
 116                 "service-handler-header": {
 117                     "request-id": "request-1"
 118                 },
 119                 "service-a-end": {
 120                     "node-id": "XPONDER-1-2",
 121                     "service-rate": "0",
 122                     "clli": "ORANGE1"
 123                 },
 124                 "service-z-end": {
 125                     "node-id": "XPONDER-3-2",
 126                     "service-rate": "0",
 127                     "clli": "ORANGE1"
 128                 }
 129             }
 130         }
 131         headers = {'content-type': 'application/json',
 132         "Accept": "application/json"}
 133         response = requests.request(
 134             "POST", url, data=json.dumps(body), headers=headers,
 135             auth=('admin', 'admin'))
 136         self.assertEqual(response.status_code, requests.codes.ok)
 137         res = response.json()
 138         self.assertIn('Path is calculated',
 139             res['output']['configuration-response-common']['response-message'])
 140         time.sleep(5)
 141
 142     # Delete topology
 143     def test_05_delete_simple_topology(self):
 144         url = ("{}/config/ietf-network:network/openroadm-topology"
 145               .format(self.restconf_baseurl))
 146         headers = {'content-type': 'application/xml',
 147         "Accept": "application/json"}
 148         response = requests.request(
 149             "DELETE", url, headers=headers, auth=('admin', 'admin'))
 150         self.assertEqual(response.status_code, requests.codes.ok)
 151         time.sleep(2)
 152
 153     # Test deleted topology
 154     def test_06_test_topology_simple_deleted(self):
 155         url = ("{}/config/ietf-network:network/openroadm-topology/node/XPONDER-1-2"
 156                 .format(self.restconf_baseurl))
 157         headers = {'content-type': 'application/json',
 158         "Accept": "application/json"}
 159         response = requests.request(
 160             "GET", url, headers=headers, auth=('admin', 'admin'))
 161         self.assertEqual(response.status_code, 404)
 162         time.sleep(1)
 163
 164     # Load simple topology
 165     def test_07_load_complex_topology(self):
 166         url = ("{}/config/ietf-network:network/openroadm-topology"
 167               .format(self.restconf_baseurl))
 168         body = self.complex_data
 169         headers = {'content-type': 'application/xml',
 170         "Accept": "application/json"}
 171         response = requests.request(
 172             "PUT", url, data=body, headers=headers,
 173             auth=('admin', 'admin'))
 174         self.assertEqual(response.status_code, 201)
 175         time.sleep(2)
 176
 177     # Get existing nodeId
 178     def test_08_get_nodeId(self):
 179         url = ("{}/config/ietf-network:network/openroadm-topology/node/XPONDER-3-2"
 180                 .format(self.restconf_baseurl))
 181         headers = {'content-type': 'application/json',
 182         "Accept": "application/json"}
 183         response = requests.request(
 184             "GET", url, headers=headers, auth=('admin', 'admin'))
 185         self.assertEqual(response.status_code, requests.codes.ok)
 186         res = response.json()
 187         self.assertEqual(
 188             res['node'][0]['node-id'],
 189             'XPONDER-3-2')
 190         time.sleep(1)
 191
 192     # Test failed path computation
 193     def test_09_fail_path_computation(self):
 194         url = ("{}/operations/pce:path-computation-request"
 195               .format(self.restconf_baseurl))
 196         body = {"input": {
 197                 "service-handler-header": {
 198                     "request-id": "request-1"
 199                 }
 200             }
 201         }
 202         headers = {'content-type': 'application/json',
 203         "Accept": "application/json"}
 204         response = requests.request(
 205             "POST", url, data=json.dumps(body), headers=headers,
 206             auth=('admin', 'admin'))
 207         self.assertEqual(response.status_code, requests.codes.ok)
 208         res = response.json()
 209         self.assertIn('Service Name is not set',
 210             res['output']['configuration-response-common']['response-message'])
 211         time.sleep(2)
 212
 213     # Test1 success path computation
 214     def test_10_success1_path_computation(self):
 215         url = ("{}/operations/pce:path-computation-request"
 216               .format(self.restconf_baseurl))
 217         body = {"input": {
 218                 "service-name": "service1",
 219                 "resource-reserve": "true",
 220                 "service-handler-header": {
 221                   "request-id": "request1"
 222                 },
 223                 "service-a-end": {
 224                   "service-format": "Ethernet",
 225                   "service-rate": "0",
 226                   "clli": "clli11",
 227                   "node-id": "XPONDER-2-2",
 228                   "tx-direction": {
 229                     "port": {
 230                       "port-device-name": "Some port-device-name",
 231                       "port-type": "Some port-type",
 232                       "port-name": "Some port-name",
 233                       "port-rack": "Some port-rack",
 234                       "port-shelf": "Some port-shelf",
 235                       "port-slot": "Some port-slot",
 236                       "port-sub-slot": "Some port-sub-slot"
 237                     }
 238                   },
 239                   "rx-direction": {
 240                     "port": {
 241                       "port-device-name": "Some port-device-name",
 242                       "port-type": "Some port-type",
 243                       "port-name": "Some port-name",
 244                       "port-rack": "Some port-rack",
 245                       "port-shelf": "Some port-shelf",
 246                       "port-slot": "Some port-slot",
 247                       "port-sub-slot": "Some port-sub-slot"
 248                     }
 249                   }
 250                 },
 251                 "service-z-end": {
 252                   "service-format": "Ethernet",
 253                   "service-rate": "0",
 254                   "clli": "clli11",
 255                   "node-id": "XPONDER-1-2",
 256                   "tx-direction": {
 257                     "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                   },
 267                   "rx-direction": {
 268                     "port": {
 269                       "port-device-name": "Some port-device-name",
 270                       "port-type": "Some port-type",
 271                       "port-name": "Some port-name",
 272                       "port-rack": "Some port-rack",
 273                       "port-shelf": "Some port-shelf",
 274                       "port-slot": "Some port-slot",
 275                       "port-sub-slot": "Some port-sub-slot"
 276                     }
 277                   }
 278                 },
 279                 "hard-constraints": {
 280                   "customer-code": [
 281                     "Some customer-code"
 282                   ],
 283                   "co-routing": {
 284                     "existing-service": [
 285                       "Some existing-service"
 286                     ]
 287                   }
 288                 },
 289                 "soft-constraints": {
 290                   "customer-code": [
 291                     "Some customer-code"
 292                   ],
 293                   "co-routing": {
 294                     "existing-service": [
 295                       "Some existing-service"
 296                     ]
 297                   }
 298                 },
 299                 "pce-metric": "hop-count",
 300                 "locally-protected-links": "true"
 301             }
 302         }
 303         headers = {'content-type': 'application/json',
 304         "Accept": "application/json"}
 305         response = requests.request(
 306             "POST", url, data=json.dumps(body), headers=headers,
 307             auth=('admin', 'admin'))
 308         self.assertEqual(response.status_code, requests.codes.ok)
 309         res = response.json()
 310         self.assertIn('Path is calculated',
 311             res['output']['configuration-response-common']['response-message'])
 312         time.sleep(5)
 313
 314     # Test2 success path computation with path description
 315     def test_11_success2_path_computation(self):
 316         url = ("{}/operations/pce:path-computation-request"
 317               .format(self.restconf_baseurl))
 318         body = {"input": {
 319                 "service-name": "service 1",
 320                 "resource-reserve": "true",
 321                 "service-handler-header": {
 322                     "request-id": "request 1"
 323                 },
 324                 "service-a-end": {
 325                     "service-rate": "0",
 326                     "node-id": "XPONDER-1-2"
 327                 },
 328                 "service-z-end": {
 329                     "service-rate": "0",
 330                     "node-id": "XPONDER-3-2"
 331              },
 332              "pce-metric": "hop-count"
 333            }
 334         }
 335         headers = {'content-type': 'application/json',
 336         "Accept": "application/json"}
 337         response = requests.request(
 338             "POST", url, data=json.dumps(body), headers=headers,
 339             auth=('admin', 'admin'))
 340         self.assertEqual(response.status_code, requests.codes.ok)
 341         res = response.json()
 342         self.assertIn('Path is calculated',
 343             res['output']['configuration-response-common']['response-message'])
 344         self.assertEqual(5 , res['output']['response-parameters']['path-description']
 345             ['aToZ-direction']['aToZ-wavelength-number'])
 346         self.assertEqual(5 , res['output']['response-parameters']['path-description']
 347             ['zToA-direction']['zToA-wavelength-number'])
 348         time.sleep(5)
 349
 350     # Test3 success path computation with hard-constraints exclude
 351     def test_12_success3_path_computation(self):
 352         url = ("{}/operations/pce:path-computation-request"
 353               .format(self.restconf_baseurl))
 354         body = {"input": {
 355                 "service-name": "service 1",
 356                 "resource-reserve": "true",
 357                 "service-handler-header": {
 358                     "request-id": "request 1"
 359                 },
 360                 "service-a-end": {
 361                     "service-rate": "0",
 362                     "node-id": "XPONDER-1-2"
 363                 },
 364                 "service-z-end": {
 365                     "service-rate": "0",
 366                     "node-id": "XPONDER-3-2"
 367                 },
 368                 "hard-constraints": {
 369                     "exclude_": {
 370                         "node-id": ["OpenROADM-2-1", "OpenROADM-2-2"]
 371                     }
 372                 },
 373                 "pce-metric": "hop-count"
 374             }
 375         }
 376         headers = {'content-type': 'application/json',
 377         "Accept": "application/json"}
 378         response = requests.request(
 379             "POST", url, data=json.dumps(body), headers=headers,
 380             auth=('admin', 'admin'))
 381         self.assertEqual(response.status_code, requests.codes.ok)
 382         res = response.json()
 383         self.assertIn('Path is calculated',
 384             res['output']['configuration-response-common']['response-message'])
 385         self.assertEqual(9 , res['output']['response-parameters']['path-description']
 386             ['aToZ-direction']['aToZ-wavelength-number'])
 387         self.assertEqual(9 , res['output']['response-parameters']['path-description']
 388             ['zToA-direction']['zToA-wavelength-number'])
 389         time.sleep(5)
 390
 391     # Delete complex topology
 392     def test_13_delete_complex_topology(self):
 393         url = ("{}/config/ietf-network:network/openroadm-topology"
 394               .format(self.restconf_baseurl))
 395         headers = {'content-type': 'application/xml',
 396         "Accept": "application/json"}
 397         response = requests.request(
 398             "DELETE", url, headers=headers, auth=('admin', 'admin'))
 399         self.assertEqual(response.status_code, requests.codes.ok)
 400         time.sleep(2)
 401
 402     # Test deleted complex topology
 403     def test_14_test_topology_complex_deleted(self):
 404         url = ("{}/config/ietf-network:network/openroadm-topology/node/XPONDER-3-2"
 405                 .format(self.restconf_baseurl))
 406         headers = {'content-type': 'application/json',
 407         "Accept": "application/json"}
 408         response = requests.request(
 409             "GET", url, headers=headers, auth=('admin', 'admin'))
 410         self.assertEqual(response.status_code, 404)
 411         time.sleep(1)
 412
 413
 414 if __name__ == "__main__":
 415     unittest.main(verbosity=2)