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