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