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