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         response = test_utils.rawput_request(test_utils.URL_CONFIG_CLLI_NET, self.topo_cllinet_data)
  81         self.assertEqual(response.status_code, requests.codes.ok)
  82         time.sleep(3)
  83
  84     def test_02_connect_openroadmNetwork(self):
  85         url = "{}/config/ietf-network:networks/network/openroadm-network"
  86         response = test_utils.rawput_request(url, self.topo_ordnet_data)
  87         self.assertEqual(response.status_code, requests.codes.ok)
  88         time.sleep(3)
  89
  90     def test_03_connect_openroadmTopology(self):
  91         response = test_utils.rawput_request(test_utils.URL_CONFIG_ORDM_TOPO, self.topo_ordtopo_data)
  92         self.assertEqual(response.status_code, requests.codes.ok)
  93         time.sleep(3)
  94
  95     # Path computed by PCE is feasible according to Gnpy
  96     def test_04_path_computation_FeasibleWithPCE(self):
  97         url = "{}/operations/transportpce-pce:path-computation-request"
  98         data = {
  99             "input": {
 100                 "service-name": "service-1",
 101                 "resource-reserve": "true",
 102                 "pce-metric": "hop-count",
 103                 "service-handler-header": {
 104                     "request-id": "request-1"
 105                 },
 106                 "service-a-end": {
 107                     "node-id": "XPONDER-1",
 108                     "service-rate": "100",
 109                     "service-format": "Ethernet",
 110                     "clli": "Node1"
 111                 },
 112                 "service-z-end": {
 113                     "node-id": "XPONDER-5",
 114                     "service-rate": "100",
 115                     "service-format": "Ethernet",
 116                     "clli": "Node5"
 117                 }
 118             }
 119         }
 120         response = test_utils.post_request(url, data)
 121         self.assertEqual(response.status_code, requests.codes.ok)
 122         res = response.json()
 123         self.assertEqual(res['output']['configuration-response-common'][
 124             'response-code'], '200')
 125         self.assertEqual(res['output']['configuration-response-common'][
 126             'response-message'],
 127             'Path is calculated by PCE')
 128         self.assertEqual(res['output']['gnpy-response'][0]['path-dir'],
 129                          'A-to-Z')
 130         self.assertEqual(res['output']['gnpy-response'][0]['feasibility'], True)
 131         self.assertEqual(res['output']['gnpy-response'][1]['path-dir'],
 132                          'Z-to-A')
 133         self.assertEqual(res['output']['gnpy-response'][1]['feasibility'], True)
 134         time.sleep(5)
 135
 136     # Path computed by PCE is not feasible by GNPy and GNPy cannot find
 137     # another one (low SNR)
 138     def test_05_path_computation_FoundByPCE_NotFeasibleByGnpy(self):
 139         url = "{}/operations/transportpce-pce:path-computation-request"
 140         data = {
 141             "input": {
 142                 "service-name": "service-2",
 143                 "resource-reserve": "true",
 144                 "pce-metric": "hop-count",
 145                 "service-handler-header": {
 146                     "request-id": "request-2"
 147                 },
 148                 "service-a-end": {
 149                     "node-id": "XPONDER-1",
 150                     "service-rate": "100",
 151                     "service-format": "Ethernet",
 152                     "clli": "Node1"
 153                 },
 154                 "service-z-end": {
 155                     "node-id": "XPONDER-5",
 156                     "service-rate": "100",
 157                     "service-format": "Ethernet",
 158                     "clli": "Node5"
 159                 },
 160                 "hard-constraints": {
 161                     "include_": {
 162                         "ordered-hops": [
 163                             {
 164                                 "hop-number": "0",
 165                                 "hop-type": {
 166                                     "node-id": "OpenROADM-2"
 167                                 }
 168                             },
 169                             {
 170                                 "hop-number": "1",
 171                                 "hop-type": {
 172                                     "node-id": "OpenROADM-3"
 173                                 }
 174                             },
 175                             {
 176                                 "hop-number": "2",
 177                                 "hop-type": {
 178                                     "node-id": "OpenROADM-4"
 179                                 }
 180                             }
 181                         ]
 182                     }
 183                 }
 184             }
 185         }
 186         response = test_utils.post_request(url, data)
 187         self.assertEqual(response.status_code, requests.codes.ok)
 188         res = response.json()
 189         self.assertEqual(res['output']['configuration-response-common'][
 190             'response-code'], '500')
 191         self.assertEqual(res['output']['configuration-response-common'][
 192             'response-message'],
 193             'No path available by PCE and GNPy ')
 194         self.assertEqual(res['output']['gnpy-response'][0]['path-dir'],
 195                          'A-to-Z')
 196         self.assertEqual(res['output']['gnpy-response'][0]['feasibility'],
 197                          False)
 198         self.assertEqual(res['output']['gnpy-response'][1]['path-dir'],
 199                          'Z-to-A')
 200         self.assertEqual(res['output']['gnpy-response'][1]['feasibility'],
 201                          False)
 202         time.sleep(5)
 203
 204     # #PCE cannot find a path while GNPy finds a feasible one
 205     def test_06_path_computation_NotFoundByPCE_FoundByGNPy(self):
 206         url = "{}/operations/transportpce-pce:path-computation-request"
 207         data = {
 208             "input": {
 209                 "service-name": "service-3",
 210                 "resource-reserve": "true",
 211                 "pce-metric": "hop-count",
 212                 "service-handler-header": {
 213                     "request-id": "request-3"
 214                 },
 215                 "service-a-end": {
 216                     "node-id": "XPONDER-1",
 217                     "service-rate": "100",
 218                     "service-format": "Ethernet",
 219                     "clli": "Node1"
 220                 },
 221                 "service-z-end": {
 222                     "node-id": "XPONDER-4",
 223                     "service-rate": "100",
 224                     "service-format": "Ethernet",
 225                     "clli": "Node5"
 226                 },
 227                 "hard-constraints": {
 228                     "include_": {
 229                         "ordered-hops": [
 230                             {
 231                                 "hop-number": "0",
 232                                 "hop-type": {
 233                                     "node-id": "OpenROADM-2"
 234                                 }
 235                             },
 236                             {
 237                                 "hop-number": "1",
 238                                 "hop-type": {
 239                                     "node-id": "OpenROADM-3"
 240                                 }
 241                             }
 242                         ]
 243                     }
 244                 }
 245             }
 246         }
 247         response = test_utils.post_request(url, data)
 248         self.assertEqual(response.status_code, requests.codes.ok)
 249         res = response.json()
 250         self.assertEqual(res['output']['configuration-response-common'][
 251             'response-code'], '200')
 252         self.assertEqual(res['output']['configuration-response-common'][
 253             'response-message'],
 254             'Path is calculated by GNPy')
 255         self.assertEqual(res['output']['gnpy-response'][0]['path-dir'],
 256                          'A-to-Z')
 257         self.assertEqual(res['output']['gnpy-response'][0]['feasibility'], True)
 258         self.assertEqual(res['output']['gnpy-response'][1]['path-dir'],
 259                          'Z-to-A')
 260         self.assertEqual(res['output']['gnpy-response'][1]['feasibility'], True)
 261         time.sleep(5)
 262
 263     # Not found path by PCE and GNPy cannot find another one
 264     def test_07_path_computation_FoundByPCE_NotFeasibleByGnpy(self):
 265         url = "{}/operations/transportpce-pce:path-computation-request"
 266         data = {
 267             "input": {
 268                 "service-name": "service-4",
 269                 "resource-reserve": "true",
 270                 "pce-metric": "hop-count",
 271                 "service-handler-header": {
 272                     "request-id": "request-4"
 273                 },
 274                 "service-a-end": {
 275                     "node-id": "XPONDER-1",
 276                     "service-rate": "100",
 277                     "service-format": "Ethernet",
 278                     "clli": "Node1"
 279                 },
 280                 "service-z-end": {
 281                     "node-id": "XPONDER-4",
 282                     "service-rate": "100",
 283                     "service-format": "Ethernet",
 284                     "clli": "Node5"
 285                 },
 286                 "hard-constraints": {
 287                     "include_": {
 288                         "ordered-hops": [
 289                             {
 290                                 "hop-number": "0",
 291                                 "hop-type": {
 292                                     "node-id": "OpenROADM-2"
 293                                 }
 294                             },
 295                             {
 296                                 "hop-number": "1",
 297                                 "hop-type": {
 298                                     "node-id": "OpenROADM-3"
 299                                 }
 300                             },
 301                             {
 302                                 "hop-number": "2",
 303                                 "hop-type": {
 304                                     "node-id": "OpenROADM-4"
 305                                 }
 306                             },
 307                             {
 308                                 "hop-number": "3",
 309                                 "hop-type": {
 310                                     "node-id": "OpenROADM-3"
 311                                 }
 312                             }
 313                         ]
 314                     }
 315                 }
 316             }
 317         }
 318         response = test_utils.post_request(url, data)
 319         self.assertEqual(response.status_code, requests.codes.ok)
 320         res = response.json()
 321         self.assertEqual(res['output']['configuration-response-common'][
 322             'response-code'], '500')
 323         self.assertEqual(res['output']['configuration-response-common'][
 324             'response-message'],
 325             'No path available by PCE and GNPy ')
 326         time.sleep(5)
 327
 328     # Disconnect the different topologies
 329     def test_08_disconnect_openroadmTopology(self):
 330         response = test_utils.delete_request(test_utils.URL_CONFIG_ORDM_TOPO)
 331         self.assertEqual(response.status_code, requests.codes.ok)
 332         time.sleep(3)
 333
 334     def test_09_disconnect_openroadmNetwork(self):
 335         url = "{}/config/ietf-network:networks/network/openroadm-network"
 336         response = test_utils.delete_request(url)
 337         self.assertEqual(response.status_code, requests.codes.ok)
 338         time.sleep(3)
 339
 340     def test_10_disconnect_clliNetwork(self):
 341         response = test_utils.delete_request(test_utils.URL_CONFIG_CLLI_NET)
 342         self.assertEqual(response.status_code, requests.codes.ok)
 343         time.sleep(3)
 344
 345
 346 if __name__ == "__main__":
 347     # logging.basicConfig(filename='./transportpce_tests/log/response.log',filemode='w',level=logging.DEBUG)
 348     unittest.main(verbosity=2)