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