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 not feasible and Gnpy computes a new one feasible
  97     def test_04_path_computation_NotFeasibleWithPCE_FeasibleWithGnpy(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                     "clli": "Node1"
 111                 },
 112                 "service-z-end": {
 113                     "node-id": "XPONDER-5",
 114                     "service-rate": "100",
 115                     "clli": "Node5"
 116                 }
 117             }
 118         }
 119         headers = {'content-type': 'application/json',
 120         "Accept": "application/json"}
 121         response = requests.request(
 122             "POST", url, data=json.dumps(body), headers=headers,
 123             auth=('admin', 'admin'))
 124         self.assertEqual(response.status_code, requests.codes.ok)
 125         res = response.json()
 126         self.assertEqual(res['output']['configuration-response-common']['response-code'], '200')
 127         self.assertEqual(res['output']['gnpy-response'][0]['path-dir'], 'A-to-Z')
 128         self.assertEqual(res['output']['gnpy-response'][0]['feasibility'],True)
 129         self.assertEqual(res['output']['gnpy-response'][1]['path-dir'], 'Z-to-A')
 130         self.assertEqual(res['output']['gnpy-response'][1]['feasibility'],True)
 131         time.sleep(5)
 132
 133     #Path computed by PCE is not feasible and Gnpy cannot find a new one because the constraints
 134     def test_05_path_computation_NotFeasibleWithPCE_NotFeasibleWithGnpy_withConstraints(self):
 135         url = ("{}/operations/transportpce-pce:path-computation-request"
 136               .format(self.restconf_baseurl))
 137         body = {"input": {
 138                 "service-name": "service-2",
 139                 "resource-reserve": "true",
 140                 "pce-metric": "hop-count",
 141                 "service-handler-header": {
 142                     "request-id": "request-2"
 143                 },
 144                 "service-a-end": {
 145                     "node-id": "XPONDER-1",
 146                     "service-rate": "100",
 147                     "clli": "Node1"
 148                 },
 149                 "service-z-end": {
 150                     "node-id": "XPONDER-5",
 151                     "service-rate": "100",
 152                     "clli": "Node5"
 153                 },
 154                 "hard-constraints": {
 155                     "include_": {
 156                         "ordered-hops": [
 157                             {
 158                                 "hop-number": "0",
 159                                 "hop-type": {
 160                                     "node-id": "XPONDER-1"
 161                                 }
 162                             },
 163                             {
 164                                 "hop-number": "2",
 165                                 "hop-type": {
 166                                     "node-id": "OpenROADM-2"
 167                                 }
 168                             }
 169                         ]
 170                     }
 171                 }
 172             }
 173         }
 174         headers = {'content-type': 'application/json',
 175         "Accept": "application/json"}
 176         response = requests.request(
 177             "POST", url, data=json.dumps(body), headers=headers,
 178             auth=('admin', 'admin'))
 179         self.assertEqual(response.status_code, requests.codes.ok)
 180         res = response.json()
 181         self.assertEqual(res['output']['configuration-response-common']['response-code'], '500')
 182         self.assertEqual(res['output']['gnpy-response'][0]['path-dir'], 'A-to-Z')
 183         self.assertEqual(res['output']['gnpy-response'][0]['feasibility'],False)
 184         self.assertEqual(res['output']['gnpy-response'][1]['path-dir'], 'Z-to-A')
 185         self.assertEqual(res['output']['gnpy-response'][1]['feasibility'],False)
 186         time.sleep(5)
 187
 188     #Path computed by PCE is feasible and Gnpy confirms the feasibility
 189     def test_06_path_computation_FeasibleWithPCE_withConstraints(self):
 190         url = ("{}/operations/transportpce-pce:path-computation-request"
 191               .format(self.restconf_baseurl))
 192         body = {"input": {
 193                 "service-name": "service-3",
 194                 "resource-reserve": "true",
 195                 "pce-metric": "hop-count",
 196                 "service-handler-header": {
 197                     "request-id": "request-3"
 198                 },
 199                 "service-a-end": {
 200                     "node-id": "XPONDER-1",
 201                     "service-rate": "100",
 202                     "clli": "Node1"
 203                 },
 204                 "service-z-end": {
 205                     "node-id": "XPONDER-5",
 206                     "service-rate": "100",
 207                     "clli": "Node5"
 208                 },
 209                 "hard-constraints": {
 210                     "include_": {
 211                         "ordered-hops": [
 212                             {
 213                                 "hop-number": "0",
 214                                 "hop-type": {
 215                                     "node-id": "XPONDER-1"
 216                                 }
 217                             },
 218                             {
 219                                 "hop-number": "2",
 220                                 "hop-type": {
 221                                     "node-id": "OpenROADM-3"
 222                                 }
 223                             },
 224                             {
 225                                 "hop-number": "4",
 226                                 "hop-type": {
 227                                     "node-id": "OpenROADM-4"
 228                                 }
 229                             }
 230                         ]
 231                     }
 232                 }
 233             }
 234         }
 235         headers = {'content-type': 'application/json',
 236         "Accept": "application/json"}
 237         response = requests.request(
 238             "POST", url, data=json.dumps(body), headers=headers,
 239             auth=('admin', 'admin'))
 240         self.assertEqual(response.status_code, requests.codes.ok)
 241         res = response.json()
 242         self.assertEqual(res['output']['configuration-response-common']['response-code'], '200')
 243         self.assertEqual(res['output']['gnpy-response'][0]['path-dir'], 'A-to-Z')
 244         self.assertEqual(res['output']['gnpy-response'][0]['feasibility'],True)
 245         self.assertEqual(res['output']['gnpy-response'][1]['path-dir'], 'Z-to-A')
 246         self.assertEqual(res['output']['gnpy-response'][1]['feasibility'],True)
 247         time.sleep(5)
 248
 249     #Disconnect the different topologies
 250     def test_07_disconnect_openroadmTopology(self):
 251         url = ("{}/config/ietf-network:networks/network/openroadm-topology"
 252                .format(self.restconf_baseurl))
 253         data = {}
 254         headers = {'content-type': 'application/json'}
 255         response = requests.request(
 256              "DELETE", url, data=json.dumps(data), headers=headers,
 257              auth=('admin', 'admin'))
 258         self.assertEqual(response.status_code, requests.codes.ok)
 259         time.sleep(3)
 260
 261     def test_08_disconnect_openroadmNetwork(self):
 262         url = ("{}/config/ietf-network:networks/network/openroadm-network"
 263                .format(self.restconf_baseurl))
 264         data = {}
 265         headers = {'content-type': 'application/json'}
 266         response = requests.request(
 267              "DELETE", url, data=json.dumps(data), headers=headers,
 268              auth=('admin', 'admin'))
 269         self.assertEqual(response.status_code, requests.codes.ok)
 270         time.sleep(3)
 271
 272     def test_09_disconnect_clliNetwork(self):
 273         url = ("{}/config/ietf-network:networks/network/clli-network"
 274                .format(self.restconf_baseurl))
 275         data = {}
 276         headers = {'content-type': 'application/json'}
 277         response = requests.request(
 278              "DELETE", url, data=json.dumps(data), headers=headers,
 279              auth=('admin', 'admin'))
 280         self.assertEqual(response.status_code, requests.codes.ok)
 281         time.sleep(3)
 282
 283 if __name__ == "__main__":
 284     #logging.basicConfig(filename='./transportpce_tests/log/response.log',filemode='w',level=logging.DEBUG)
 285     unittest.main(verbosity=2)