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