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