tests/transportpce_tests/1.2.1/test_gnpy.py

   1 #!/usr/bin/env python
   2
   3 ##############################################################################
   4 # Copyright (c) 2017 Orange, Inc. and others.  All rights reserved.
   5 #
   6 # All rights reserved. This program and the accompanying materials
   7 # are made available under the terms of the Apache License, Version 2.0
   8 # which accompanies this distribution, and is available at
   9 # http://www.apache.org/licenses/LICENSE-2.0
  10 ##############################################################################
  11
  12 import json
  13 import os
  14 import psutil
  15 import requests
  16 import signal
  17 import time
  18 import unittest
  19 import test_utils
  20
  21
  22 class TransportGNPYtesting(unittest.TestCase):
  23
  24     gnpy_process = None
  25     odl_process = None
  26     restconf_baseurl = "http://localhost:8181/restconf"
  27
  28     @classmethod
  29     def __init_logfile(cls):
  30         if os.path.isfile("./transportpce_tests/gnpy.log"):
  31             os.remove("transportpce_tests/gnpy.log")
  32
  33     @classmethod
  34     def setUpClass(cls):
  35         cls.odl_process = test_utils.start_tpce()
  36         time.sleep(30)
  37         print("opendaylight started")
  38
  39     @classmethod
  40     def tearDownClass(cls):
  41         for child in psutil.Process(cls.odl_process.pid).children():
  42             child.send_signal(signal.SIGINT)
  43             child.wait()
  44         cls.odl_process.send_signal(signal.SIGINT)
  45         cls.odl_process.wait()
  46
  47     def setUp(self):
  48         time.sleep(2)
  49
  50     # Mount the different topologies
  51     def test_01_connect_clliNetwork(self):
  52         url = ("{}/config/ietf-network:networks/network/clli-network"
  53                .format(self.restconf_baseurl))
  54         topo_cllinet_file = "sample_configs/gnpy/clliNetwork.json"
  55         if os.path.isfile(topo_cllinet_file):
  56             with open(topo_cllinet_file, 'r') as clli_net:
  57                 body = clli_net.read()
  58         headers = {'content-type': 'application/json'}
  59         response = requests.request(
  60             "PUT", url, data=body, headers=headers,
  61             auth=('admin', 'admin'))
  62         self.assertEqual(response.status_code, requests.codes.ok)
  63         time.sleep(3)
  64
  65     def test_02_connect_openroadmNetwork(self):
  66         url = ("{}/config/ietf-network:networks/network/openroadm-network"
  67                .format(self.restconf_baseurl))
  68         topo_ordnet_file = "sample_configs/gnpy/openroadmNetwork.json"
  69         if os.path.isfile(topo_ordnet_file):
  70             with open(topo_ordnet_file, 'r') as ord_net:
  71                 body = ord_net.read()
  72         headers = {'content-type': 'application/json'}
  73         response = requests.request(
  74             "PUT", url, data=body, headers=headers,
  75             auth=('admin', 'admin'))
  76         self.assertEqual(response.status_code, requests.codes.ok)
  77         time.sleep(3)
  78
  79     def test_03_connect_openroadmTopology(self):
  80         url = ("{}/config/ietf-network:networks/network/openroadm-topology"
  81                .format(self.restconf_baseurl))
  82         topo_ordtopo_file = "sample_configs/gnpy/openroadmTopology.json"
  83         if os.path.isfile(topo_ordtopo_file):
  84             with open(topo_ordtopo_file, 'r') as ord_topo:
  85                 body = ord_topo.read()
  86         headers = {'content-type': 'application/json'}
  87         response = requests.request(
  88             "PUT", url, data=body, headers=headers,
  89             auth=('admin', 'admin'))
  90         self.assertEqual(response.status_code, requests.codes.ok)
  91         time.sleep(3)
  92
  93     # Path computed by PCE is feasible according to Gnpy
  94     def test_04_path_computation_FeasibleWithPCE(self):
  95         url = ("{}/operations/transportpce-pce:path-computation-request"
  96                .format(self.restconf_baseurl))
  97         body = {
  98             "input": {
  99                 "service-name": "service-1",
 100                 "resource-reserve": "true",
 101                 "pce-metric": "hop-count",
 102                 "service-handler-header": {
 103                     "request-id": "request-1"
 104                 },
 105                 "service-a-end": {
 106                     "node-id": "XPONDER-1",
 107                     "service-rate": "100",
 108                     "service-format": "Ethernet",
 109                     "clli": "Node1"
 110                 },
 111                 "service-z-end": {
 112                     "node-id": "XPONDER-5",
 113                     "service-rate": "100",
 114                     "service-format": "Ethernet",
 115                     "clli": "Node5"
 116                 }
 117             }
 118         }
 119         headers = {'content-type': 'application/json',
 120                    "Accept": "application/json"}
 121         response = requests.request(
 122             "POST", url, data=json.dumps(body), headers=headers,
 123             auth=('admin', 'admin'))
 124         self.assertEqual(response.status_code, requests.codes.ok)
 125         res = response.json()
 126         self.assertEqual(res['output']['configuration-response-common'][
 127             'response-code'], '200')
 128         self.assertEqual(res['output']['configuration-response-common'][
 129             'response-message'],
 130             'Path is calculated by PCE')
 131         self.assertEqual(res['output']['gnpy-response'][0]['path-dir'],
 132                          'A-to-Z')
 133         self.assertEqual(res['output']['gnpy-response'][0]['feasibility'], True)
 134         self.assertEqual(res['output']['gnpy-response'][1]['path-dir'],
 135                          'Z-to-A')
 136         self.assertEqual(res['output']['gnpy-response'][1]['feasibility'], True)
 137         time.sleep(5)
 138
 139     # Path computed by PCE is not feasible by GNPy and GNPy cannot find
 140     # another one (low SNR)
 141     def test_05_path_computation_FoundByPCE_NotFeasibleByGnpy(self):
 142         url = ("{}/operations/transportpce-pce:path-computation-request"
 143                .format(self.restconf_baseurl))
 144         body = {
 145             "input": {
 146                 "service-name": "service-2",
 147                 "resource-reserve": "true",
 148                 "pce-metric": "hop-count",
 149                 "service-handler-header": {
 150                     "request-id": "request-2"
 151                 },
 152                 "service-a-end": {
 153                     "node-id": "XPONDER-1",
 154                     "service-rate": "100",
 155                     "service-format": "Ethernet",
 156                     "clli": "Node1"
 157                 },
 158                 "service-z-end": {
 159                     "node-id": "XPONDER-5",
 160                     "service-rate": "100",
 161                     "service-format": "Ethernet",
 162                     "clli": "Node5"
 163                 },
 164                 "hard-constraints": {
 165                     "include_": {
 166                         "ordered-hops": [
 167                             {
 168                                 "hop-number": "0",
 169                                 "hop-type": {
 170                                     "node-id": "OpenROADM-2"
 171                                 }
 172                             },
 173                             {
 174                                 "hop-number": "1",
 175                                 "hop-type": {
 176                                     "node-id": "OpenROADM-3"
 177                                 }
 178                             },
 179                             {
 180                                 "hop-number": "2",
 181                                 "hop-type": {
 182                                     "node-id": "OpenROADM-4"
 183                                 }
 184                             }
 185                         ]
 186                     }
 187                 }
 188             }
 189         }
 190         headers = {'content-type': 'application/json',
 191                    "Accept": "application/json"}
 192         response = requests.request(
 193             "POST", url, data=json.dumps(body), headers=headers,
 194             auth=('admin', 'admin'))
 195         self.assertEqual(response.status_code, requests.codes.ok)
 196         res = response.json()
 197         self.assertEqual(res['output']['configuration-response-common'][
 198             'response-code'], '500')
 199         self.assertEqual(res['output']['configuration-response-common'][
 200             'response-message'],
 201             'No path available by PCE and GNPy ')
 202         self.assertEqual(res['output']['gnpy-response'][0]['path-dir'],
 203                          'A-to-Z')
 204         self.assertEqual(res['output']['gnpy-response'][0]['feasibility'],
 205                          False)
 206         self.assertEqual(res['output']['gnpy-response'][1]['path-dir'],
 207                          'Z-to-A')
 208         self.assertEqual(res['output']['gnpy-response'][1]['feasibility'],
 209                          False)
 210         time.sleep(5)
 211
 212     # #PCE cannot find a path while GNPy finds a feasible one
 213     def test_06_path_computation_NotFoundByPCE_FoundByGNPy(self):
 214         url = ("{}/operations/transportpce-pce:path-computation-request"
 215                .format(self.restconf_baseurl))
 216         body = {
 217             "input": {
 218                 "service-name": "service-3",
 219                 "resource-reserve": "true",
 220                 "pce-metric": "hop-count",
 221                 "service-handler-header": {
 222                     "request-id": "request-3"
 223                 },
 224                 "service-a-end": {
 225                     "node-id": "XPONDER-1",
 226                     "service-rate": "100",
 227                     "service-format": "Ethernet",
 228                     "clli": "Node1"
 229                 },
 230                 "service-z-end": {
 231                     "node-id": "XPONDER-4",
 232                     "service-rate": "100",
 233                     "service-format": "Ethernet",
 234                     "clli": "Node5"
 235                 },
 236                 "hard-constraints": {
 237                     "include_": {
 238                         "ordered-hops": [
 239                             {
 240                                 "hop-number": "0",
 241                                 "hop-type": {
 242                                     "node-id": "OpenROADM-2"
 243                                 }
 244                             },
 245                             {
 246                                 "hop-number": "1",
 247                                 "hop-type": {
 248                                     "node-id": "OpenROADM-3"
 249                                 }
 250                             }
 251                         ]
 252                     }
 253                 }
 254             }
 255         }
 256         headers = {'content-type': 'application/json',
 257                    "Accept": "application/json"}
 258         response = requests.request(
 259             "POST", url, data=json.dumps(body), headers=headers,
 260             auth=('admin', 'admin'))
 261         self.assertEqual(response.status_code, requests.codes.ok)
 262         res = response.json()
 263         self.assertEqual(res['output']['configuration-response-common'][
 264             'response-code'], '200')
 265         self.assertEqual(res['output']['configuration-response-common'][
 266             'response-message'],
 267             'Path is calculated by GNPy')
 268         self.assertEqual(res['output']['gnpy-response'][0]['path-dir'],
 269                          'A-to-Z')
 270         self.assertEqual(res['output']['gnpy-response'][0]['feasibility'], True)
 271         self.assertEqual(res['output']['gnpy-response'][1]['path-dir'],
 272                          'Z-to-A')
 273         self.assertEqual(res['output']['gnpy-response'][1]['feasibility'], True)
 274         time.sleep(5)
 275
 276     # Not found path by PCE and GNPy cannot find another one
 277     def test_07_path_computation_FoundByPCE_NotFeasibleByGnpy(self):
 278         url = ("{}/operations/transportpce-pce:path-computation-request"
 279                .format(self.restconf_baseurl))
 280         body = {
 281             "input": {
 282                 "service-name": "service-4",
 283                 "resource-reserve": "true",
 284                 "pce-metric": "hop-count",
 285                 "service-handler-header": {
 286                     "request-id": "request-4"
 287                 },
 288                 "service-a-end": {
 289                     "node-id": "XPONDER-1",
 290                     "service-rate": "100",
 291                     "service-format": "Ethernet",
 292                     "clli": "Node1"
 293                 },
 294                 "service-z-end": {
 295                     "node-id": "XPONDER-4",
 296                     "service-rate": "100",
 297                     "service-format": "Ethernet",
 298                     "clli": "Node5"
 299                 },
 300                 "hard-constraints": {
 301                     "include_": {
 302                         "ordered-hops": [
 303                             {
 304                                 "hop-number": "0",
 305                                 "hop-type": {
 306                                     "node-id": "OpenROADM-2"
 307                                 }
 308                             },
 309                             {
 310                                 "hop-number": "1",
 311                                 "hop-type": {
 312                                     "node-id": "OpenROADM-3"
 313                                 }
 314                             },
 315                             {
 316                                 "hop-number": "2",
 317                                 "hop-type": {
 318                                     "node-id": "OpenROADM-4"
 319                                 }
 320                             },
 321                             {
 322                                 "hop-number": "3",
 323                                 "hop-type": {
 324                                     "node-id": "OpenROADM-3"
 325                                 }
 326                             }
 327                         ]
 328                     }
 329                 }
 330             }
 331         }
 332         headers = {'content-type': 'application/json',
 333                    "Accept": "application/json"}
 334         response = requests.request(
 335             "POST", url, data=json.dumps(body), headers=headers,
 336             auth=('admin', 'admin'))
 337         self.assertEqual(response.status_code, requests.codes.ok)
 338         res = response.json()
 339         self.assertEqual(res['output']['configuration-response-common'][
 340             'response-code'], '500')
 341         self.assertEqual(res['output']['configuration-response-common'][
 342             'response-message'],
 343             'No path available by PCE and GNPy ')
 344         time.sleep(5)
 345
 346     # Disconnect the different topologies
 347     def test_08_disconnect_openroadmTopology(self):
 348         url = ("{}/config/ietf-network:networks/network/openroadm-topology"
 349                .format(self.restconf_baseurl))
 350         data = {}
 351         headers = {'content-type': 'application/json'}
 352         response = requests.request(
 353             "DELETE", url, data=json.dumps(data), headers=headers,
 354             auth=('admin', 'admin'))
 355         self.assertEqual(response.status_code, requests.codes.ok)
 356         time.sleep(3)
 357
 358     def test_09_disconnect_openroadmNetwork(self):
 359         url = ("{}/config/ietf-network:networks/network/openroadm-network"
 360                .format(self.restconf_baseurl))
 361         data = {}
 362         headers = {'content-type': 'application/json'}
 363         response = requests.request(
 364             "DELETE", url, data=json.dumps(data), headers=headers,
 365             auth=('admin', 'admin'))
 366         self.assertEqual(response.status_code, requests.codes.ok)
 367         time.sleep(3)
 368
 369     def test_10_disconnect_clliNetwork(self):
 370         url = ("{}/config/ietf-network:networks/network/clli-network"
 371                .format(self.restconf_baseurl))
 372         data = {}
 373         headers = {'content-type': 'application/json'}
 374         response = requests.request(
 375             "DELETE", url, data=json.dumps(data), headers=headers,
 376             auth=('admin', 'admin'))
 377         self.assertEqual(response.status_code, requests.codes.ok)
 378         time.sleep(3)
 379
 380
 381 if __name__ == "__main__":
 382     # logging.basicConfig(filename='./transportpce_tests/log/response.log',filemode='w',level=logging.DEBUG)
 383     unittest.main(verbosity=2)