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 shutil
  18 import subprocess
  19 import time
  20 import unittest
  21 import logging
  22 import test_utils
  23
  24 class TransportGNPYtesting(unittest.TestCase):
  25
  26     gnpy_process = None
  27     odl_process = None
  28     restconf_baseurl = "http://localhost:8181/restconf"
  29
  30     @classmethod
  31     def __init_logfile(cls):
  32         if os.path.isfile("./transportpce_tests/gnpy.log"):
  33             os.remove("transportpce_tests/gnpy.log")
  34
  35     @classmethod
  36     def setUpClass(cls):
  37         print ("starting opendaylight...")
  38         cls.odl_process = test_utils.start_tpce()
  39         time.sleep(30)
  40         print ("opendaylight started")
  41
  42     @classmethod
  43     def tearDownClass(cls):
  44         for child in psutil.Process(cls.odl_process.pid).children():
  45             child.send_signal(signal.SIGINT)
  46             child.wait()
  47         cls.odl_process.send_signal(signal.SIGINT)
  48         cls.odl_process.wait()
  49
  50     def setUp(self):
  51         time.sleep(2)
  52
  53     #Mount the different topologies
  54     def test_01_connect_clliNetwork(self):
  55         url = ("{}/config/ietf-network:networks/network/clli-network"
  56                .format(self.restconf_baseurl))
  57         topo_clliNet_file = "sample_configs/gnpy/clliNetwork.json"
  58         if os.path.isfile(topo_clliNet_file):
  59             with open(topo_clliNet_file, 'r') as clli_net:
  60                 body = clli_net.read()
  61         headers = {'content-type': 'application/json'}
  62         response = requests.request(
  63              "PUT", url, data=body, headers=headers,
  64              auth=('admin', 'admin'))
  65         self.assertEqual(response.status_code, requests.codes.ok)
  66         time.sleep(3)
  67
  68     def test_02_connect_openroadmNetwork(self):
  69         url = ("{}/config/ietf-network:networks/network/openroadm-network"
  70                .format(self.restconf_baseurl))
  71         topo_ordNet_file = "sample_configs/gnpy/openroadmNetwork.json"
  72         if os.path.isfile(topo_ordNet_file):
  73             with open(topo_ordNet_file, 'r') as ord_net:
  74                 body = ord_net.read()
  75         headers = {'content-type': 'application/json'}
  76         response = requests.request(
  77              "PUT", url, data=body, headers=headers,
  78              auth=('admin', 'admin'))
  79         self.assertEqual(response.status_code, requests.codes.ok)
  80         time.sleep(3)
  81
  82     def test_03_connect_openroadmTopology(self):
  83         url = ("{}/config/ietf-network:networks/network/openroadm-topology"
  84                .format(self.restconf_baseurl))
  85         topo_ordTopo_file = "sample_configs/gnpy/openroadmTopology.json"
  86         if os.path.isfile(topo_ordTopo_file):
  87             with open(topo_ordTopo_file, 'r') as ord_topo:
  88                 body = ord_topo.read()
  89         headers = {'content-type': 'application/json'}
  90         response = requests.request(
  91              "PUT", url, data=body, headers=headers,
  92              auth=('admin', 'admin'))
  93         self.assertEqual(response.status_code, requests.codes.ok)
  94         time.sleep(3)
  95
  96     #Path computed by PCE is feasible according to Gnpy
  97     def test_04_path_computation_FeasibleWithPCE(self):
  98         url = ("{}/operations/transportpce-pce:path-computation-request"
  99               .format(self.restconf_baseurl))
 100         body = {"input": {
 101                 "service-name": "service-1",
 102                 "resource-reserve": "true",
 103                 "pce-metric": "hop-count",
 104                 "service-handler-header": {
 105                     "request-id": "request-1"
 106                 },
 107                 "service-a-end": {
 108                     "node-id": "XPONDER-1",
 109                     "service-rate": "100",
 110                     "service-format": "Ethernet",
 111                     "clli": "Node1"
 112                 },
 113                 "service-z-end": {
 114                     "node-id": "XPONDER-5",
 115                     "service-rate": "100",
 116                     "service-format": "Ethernet",
 117                     "clli": "Node5"
 118                 }
 119             }
 120         }
 121         headers = {'content-type': 'application/json',
 122         "Accept": "application/json"}
 123         response = requests.request(
 124             "POST", url, data=json.dumps(body), headers=headers,
 125             auth=('admin', 'admin'))
 126         self.assertEqual(response.status_code, requests.codes.ok)
 127         res = response.json()
 128         self.assertEqual(res['output']['configuration-response-common']['response-code'], '200')
 129         self.assertEqual(res['output']['configuration-response-common']['response-message'],
 130                          'Path is calculated by PCE')
 131         self.assertEqual(res['output']['gnpy-response'][0]['path-dir'], 'A-to-Z')
 132         self.assertEqual(res['output']['gnpy-response'][0]['feasibility'],True)
 133         self.assertEqual(res['output']['gnpy-response'][1]['path-dir'], 'Z-to-A')
 134         self.assertEqual(res['output']['gnpy-response'][1]['feasibility'],True)
 135         time.sleep(5)
 136
 137     #Path computed by PCE is not feasible by GNPy and GNPy cannot find another one (low SNR)
 138     def test_05_path_computation_FoundByPCE_NotFeasibleByGnpy(self):
 139         url = ("{}/operations/transportpce-pce:path-computation-request"
 140               .format(self.restconf_baseurl))
 141         body = {"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         headers = {'content-type': 'application/json',
 187         "Accept": "application/json"}
 188         response = requests.request(
 189             "POST", url, data=json.dumps(body), headers=headers,
 190             auth=('admin', 'admin'))
 191         self.assertEqual(response.status_code, requests.codes.ok)
 192         res = response.json()
 193         self.assertEqual(res['output']['configuration-response-common']['response-code'], '500')
 194         self.assertEqual(res['output']['configuration-response-common']['response-message'],
 195                          'No path available by PCE and GNPy ')
 196         self.assertEqual(res['output']['gnpy-response'][0]['path-dir'], 'A-to-Z')
 197         self.assertEqual(res['output']['gnpy-response'][0]['feasibility'],False)
 198         self.assertEqual(res['output']['gnpy-response'][1]['path-dir'], 'Z-to-A')
 199         self.assertEqual(res['output']['gnpy-response'][1]['feasibility'],False)
 200         time.sleep(5)
 201
 202     # #PCE cannot find a path while GNPy finds a feasible one
 203     def test_06_path_computation_NotFoundByPCE_FoundByGNPy(self):
 204         url = ("{}/operations/transportpce-pce:path-computation-request"
 205               .format(self.restconf_baseurl))
 206         body = {"input": {
 207                 "service-name": "service-3",
 208                 "resource-reserve": "true",
 209                 "pce-metric": "hop-count",
 210                 "service-handler-header": {
 211                     "request-id": "request-3"
 212                 },
 213                 "service-a-end": {
 214                     "node-id": "XPONDER-1",
 215                     "service-rate": "100",
 216                     "service-format": "Ethernet",
 217                     "clli": "Node1"
 218                 },
 219                 "service-z-end": {
 220                     "node-id": "XPONDER-4",
 221                     "service-rate": "100",
 222                     "service-format": "Ethernet",
 223                     "clli": "Node5"
 224                 },
 225                 "hard-constraints": {
 226                     "include_": {
 227                         "ordered-hops": [
 228                             {
 229                                 "hop-number": "0",
 230                                 "hop-type": {
 231                                     "node-id": "OpenROADM-2"
 232                                 }
 233                             },
 234                             {
 235                                 "hop-number": "1",
 236                                 "hop-type": {
 237                                     "node-id": "OpenROADM-3"
 238                                 }
 239                             }
 240                         ]
 241                     }
 242                 }
 243             }
 244         }
 245         headers = {'content-type': 'application/json',
 246         "Accept": "application/json"}
 247         response = requests.request(
 248             "POST", url, data=json.dumps(body), headers=headers,
 249             auth=('admin', 'admin'))
 250         self.assertEqual(response.status_code, requests.codes.ok)
 251         res = response.json()
 252         self.assertEqual(res['output']['configuration-response-common']['response-code'], '200')
 253         self.assertEqual(res['output']['configuration-response-common']['response-message'],
 254                          'Path is calculated by GNPy')
 255         self.assertEqual(res['output']['gnpy-response'][0]['path-dir'], 'A-to-Z')
 256         self.assertEqual(res['output']['gnpy-response'][0]['feasibility'],True)
 257         self.assertEqual(res['output']['gnpy-response'][1]['path-dir'], '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(self.restconf_baseurl))
 265         body = {"input": {
 266                 "service-name": "service-4",
 267                 "resource-reserve": "true",
 268                 "pce-metric": "hop-count",
 269                 "service-handler-header": {
 270                     "request-id": "request-4"
 271                 },
 272                 "service-a-end": {
 273                     "node-id": "XPONDER-1",
 274                     "service-rate": "100",
 275                     "service-format": "Ethernet",
 276                     "clli": "Node1"
 277                 },
 278                 "service-z-end": {
 279                     "node-id": "XPONDER-4",
 280                     "service-rate": "100",
 281                     "service-format": "Ethernet",
 282                     "clli": "Node5"
 283                 },
 284                 "hard-constraints": {
 285                     "include_": {
 286                         "ordered-hops": [
 287                             {
 288                                 "hop-number": "0",
 289                                 "hop-type": {
 290                                     "node-id": "OpenROADM-2"
 291                                 }
 292                             },
 293                             {
 294                                 "hop-number": "1",
 295                                 "hop-type": {
 296                                     "node-id": "OpenROADM-3"
 297                                 }
 298                             },
 299                             {
 300                                 "hop-number": "2",
 301                                 "hop-type": {
 302                                     "node-id": "OpenROADM-4"
 303                                 }
 304                             },
 305                             {
 306                                 "hop-number": "3",
 307                                 "hop-type": {
 308                                     "node-id": "OpenROADM-3"
 309                                 }
 310                             }
 311                         ]
 312                     }
 313                 }
 314             }
 315         }
 316         headers = {'content-type': 'application/json',
 317         "Accept": "application/json"}
 318         response = requests.request(
 319             "POST", url, data=json.dumps(body), headers=headers,
 320             auth=('admin', 'admin'))
 321         self.assertEqual(response.status_code, requests.codes.ok)
 322         res = response.json()
 323         self.assertEqual(res['output']['configuration-response-common']['response-code'], '500')
 324         self.assertEqual(res['output']['configuration-response-common']['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         url = ("{}/config/ietf-network:networks/network/openroadm-topology"
 331                .format(self.restconf_baseurl))
 332         data = {}
 333         headers = {'content-type': 'application/json'}
 334         response = requests.request(
 335              "DELETE", url, data=json.dumps(data), headers=headers,
 336              auth=('admin', 'admin'))
 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(self.restconf_baseurl))
 343         data = {}
 344         headers = {'content-type': 'application/json'}
 345         response = requests.request(
 346              "DELETE", url, data=json.dumps(data), headers=headers,
 347              auth=('admin', 'admin'))
 348         self.assertEqual(response.status_code, requests.codes.ok)
 349         time.sleep(3)
 350
 351     def test_10_disconnect_clliNetwork(self):
 352         url = ("{}/config/ietf-network:networks/network/clli-network"
 353                .format(self.restconf_baseurl))
 354         data = {}
 355         headers = {'content-type': 'application/json'}
 356         response = requests.request(
 357              "DELETE", url, data=json.dumps(data), headers=headers,
 358              auth=('admin', 'admin'))
 359         self.assertEqual(response.status_code, requests.codes.ok)
 360         time.sleep(3)
 361
 362 if __name__ == "__main__":
 363     #logging.basicConfig(filename='./transportpce_tests/log/response.log',filemode='w',level=logging.DEBUG)
 364     unittest.main(verbosity=2)