tests/transportpce_tests/pce/test03_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 # pylint: disable=no-member
  13 # pylint: disable=too-many-public-methods
  14
  15 import unittest
  16 import os
  17 # pylint: disable=wrong-import-order
  18 import sys
  19 import subprocess
  20 import time
  21 import requests
  22 sys.path.append('transportpce_tests/common/')
  23 # pylint: disable=wrong-import-position
  24 # pylint: disable=import-error
  25 import test_utils  # nopep8
  26
  27
  28 class TransportGNPYtesting(unittest.TestCase):
  29     path_computation_input_data = {
  30         "service-name": "service-1",
  31         "resource-reserve": "true",
  32         "service-handler-header": {
  33             "request-id": "request-1"
  34         },
  35         "service-a-end": {
  36             "service-rate": "100",
  37             "clli": "Node1",
  38             "service-format": "Ethernet",
  39             "node-id": "XPONDER-1"
  40         },
  41         "service-z-end": {
  42             "service-rate": "100",
  43             "clli": "Node5",
  44             "service-format": "Ethernet",
  45             "node-id": "XPONDER-5"
  46         },
  47         "pce-routing-metric": "hop-count"
  48     }
  49
  50     topo_cllinet_data = None
  51     topo_ordnet_data = None
  52     topo_ordtopo_data = None
  53     port_mapping_data = None
  54     processes = []
  55
  56     @classmethod
  57     def setUpClass(cls):
  58         # pylint: disable=bare-except
  59         sample_files_parsed = False
  60         try:
  61             TOPO_CLLINET_FILE = os.path.join(os.path.dirname(os.path.realpath(__file__)),
  62                                              "..", "..", "sample_configs", "gnpy", "clliNetwork.json")
  63             with open(TOPO_CLLINET_FILE, 'r', encoding='utf-8') as topo_cllinet:
  64                 cls.topo_cllinet_data = topo_cllinet.read()
  65
  66             TOPO_ORDNET_FILE = os.path.join(os.path.dirname(os.path.realpath(__file__)),
  67                                             "..", "..", "sample_configs", "gnpy", "openroadmNetwork.json")
  68             with open(TOPO_ORDNET_FILE, 'r', encoding='utf-8') as topo_ordnet:
  69                 cls.topo_ordnet_data = topo_ordnet.read()
  70
  71             TOPO_ORDTOPO_FILE = os.path.join(os.path.dirname(os.path.realpath(__file__)),
  72                                              "..", "..", "sample_configs", "gnpy", "openroadmTopology.json")
  73             with open(TOPO_ORDTOPO_FILE, 'r', encoding='utf-8') as topo_ordtopo:
  74                 cls.topo_ordtopo_data = topo_ordtopo.read()
  75             PORT_MAPPING_FILE = os.path.join(os.path.dirname(os.path.realpath(__file__)),
  76                                              "..", "..", "sample_configs", "gnpy", "gnpy_portmapping_121.json")
  77             with open(PORT_MAPPING_FILE, 'r', encoding='utf-8') as port_mapping:
  78                 cls.port_mapping_data = port_mapping.read()
  79             sample_files_parsed = True
  80         except PermissionError as err:
  81             print("Permission Error when trying to read sample files\n", err)
  82             sys.exit(2)
  83         except FileNotFoundError as err:
  84             print("File Not found Error when trying to read sample files\n", err)
  85             sys.exit(2)
  86         except:
  87             print("Unexpected error when trying to read sample files\n", sys.exc_info()[0])
  88             sys.exit(2)
  89         finally:
  90             if sample_files_parsed:
  91                 print("sample files content loaded")
  92
  93         with open('gnpy.log', 'w', encoding='utf-8') as outfile:
  94             print('starting GNPy REST server...')
  95             # pylint: disable=consider-using-with
  96             test_utils.process_list.append(subprocess.Popen(
  97                 ['gnpy-rest'], stdout=outfile, stderr=outfile, stdin=None))
  98         cls.processes = test_utils.start_tpce()
  99
 100     @classmethod
 101     def tearDownClass(cls):
 102         # clean datastores
 103         test_utils.del_portmapping()
 104         test_utils.del_ietf_network('openroadm-topology')
 105         test_utils.del_ietf_network('openroadm-network')
 106         test_utils.del_ietf_network('clli-network')
 107         # pylint: disable=not-an-iterable
 108         for process in cls.processes:
 109             test_utils.shutdown_process(process)
 110         print("all processes killed")
 111
 112     def setUp(self):
 113         time.sleep(1)
 114
 115      # Load port mapping
 116     def test_00_load_port_mapping(self):
 117         response = test_utils.post_portmapping(self.port_mapping_data)
 118         self.assertIn(response['status_code'], (requests.codes.created, requests.codes.no_content))
 119         time.sleep(1)
 120
 121     # Mount the different topologies
 122     def test_01_connect_clliNetwork(self):
 123         response = test_utils.put_ietf_network('clli-network', self.topo_cllinet_data)
 124         self.assertIn(response['status_code'], (requests.codes.ok, requests.codes.no_content))
 125         time.sleep(1)
 126
 127     def test_02_connect_openroadmNetwork(self):
 128         response = test_utils.put_ietf_network('openroadm-network', self.topo_ordnet_data)
 129         self.assertIn(response['status_code'], (requests.codes.ok, requests.codes.no_content))
 130         time.sleep(1)
 131
 132     def test_03_connect_openroadmTopology(self):
 133         response = test_utils.put_ietf_network('openroadm-topology', self.topo_ordtopo_data)
 134         self.assertIn(response['status_code'], (requests.codes.ok, requests.codes.no_content))
 135         time.sleep(1)
 136
 137     # Path computed by PCE is feasible according to Gnpy
 138     def test_04_path_computation_FeasibleWithPCE(self):
 139         response = test_utils.transportpce_api_rpc_request('transportpce-pce',
 140                                                            'path-computation-request',
 141                                                            self.path_computation_input_data)
 142         self.assertEqual(response['status_code'], requests.codes.ok)
 143         self.assertEqual(response['output']['configuration-response-common']['response-code'], '200')
 144         self.assertEqual(response['output']['configuration-response-common']['response-message'],
 145                          'Path is calculated by PCE')
 146         self.assertIn('A-to-Z',
 147                       [response['output']['gnpy-response'][0]['path-dir'],
 148                        response['output']['gnpy-response'][1]['path-dir']])
 149         self.assertIn('Z-to-A',
 150                       [response['output']['gnpy-response'][0]['path-dir'],
 151                        response['output']['gnpy-response'][1]['path-dir']])
 152         self.assertEqual(response['output']['gnpy-response'][0]['feasibility'], True)
 153         self.assertEqual(response['output']['gnpy-response'][1]['feasibility'], True)
 154         time.sleep(2)
 155
 156     # Path computed by PCE is not feasible by GNPy and GNPy cannot find
 157     # another one (low SNR)
 158     def test_05_path_computation_FoundByPCE_NotFeasibleByGnpy(self):
 159         self.path_computation_input_data["service-name"] = "service-2"
 160         self.path_computation_input_data["service-handler-header"]["request-id"] = "request-2"
 161         self.path_computation_input_data["hard-constraints"] =\
 162             {"include": {"node-id": ["OpenROADM-2", "OpenROADM-3", "OpenROADM-4"]}}
 163         response = test_utils.transportpce_api_rpc_request('transportpce-pce',
 164                                                            'path-computation-request',
 165                                                            self.path_computation_input_data)
 166         self.assertEqual(response['status_code'], requests.codes.ok)
 167         self.assertEqual(response['output']['configuration-response-common'][
 168             'response-code'], '500')
 169         self.assertEqual(response['output']['configuration-response-common'][
 170             'response-message'],
 171             'No path available by PCE and GNPy ')
 172         self.assertIn('A-to-Z',
 173                       [response['output']['gnpy-response'][0]['path-dir'],
 174                        response['output']['gnpy-response'][1]['path-dir']])
 175         self.assertIn('Z-to-A',
 176                       [response['output']['gnpy-response'][0]['path-dir'],
 177                        response['output']['gnpy-response'][1]['path-dir']])
 178         self.assertEqual(response['output']['gnpy-response'][0]['feasibility'],
 179                          False)
 180         self.assertEqual(response['output']['gnpy-response'][1]['feasibility'],
 181                          False)
 182         time.sleep(2)
 183
 184     # #PCE cannot find a path while GNPy finds a feasible one
 185     def test_06_path_computation_NotFoundByPCE_FoundByGNPy(self):
 186         self.path_computation_input_data["service-name"] = "service-3"
 187         self.path_computation_input_data["service-handler-header"]["request-id"] = "request-3"
 188         self.path_computation_input_data["service-z-end"]["node-id"] = "XPONDER-4"
 189         self.path_computation_input_data["hard-constraints"] =\
 190             {"include": {"node-id": ["OpenROADM-2", "OpenROADM-3"]}}
 191         response = test_utils.transportpce_api_rpc_request('transportpce-pce',
 192                                                            'path-computation-request',
 193                                                            self.path_computation_input_data)
 194         self.assertEqual(response['status_code'], requests.codes.ok)
 195         self.assertEqual(response['output']['configuration-response-common'][
 196             'response-code'], '200')
 197         self.assertEqual(response['output']['configuration-response-common'][
 198             'response-message'],
 199             'Path is calculated by GNPy')
 200         self.assertIn('A-to-Z',
 201                       [response['output']['gnpy-response'][0]['path-dir'],
 202                        response['output']['gnpy-response'][1]['path-dir']])
 203         self.assertIn('Z-to-A',
 204                       [response['output']['gnpy-response'][0]['path-dir'],
 205                        response['output']['gnpy-response'][1]['path-dir']])
 206         self.assertEqual(response['output']['gnpy-response'][1]['feasibility'], True)
 207         self.assertEqual(response['output']['gnpy-response'][0]['feasibility'], True)
 208         time.sleep(2)
 209
 210     # Not found path by PCE and GNPy cannot find another one
 211     def test_07_path_computation_FoundByPCE_NotFeasibleByGnpy(self):
 212         self.path_computation_input_data["service-name"] = "service-4"
 213         self.path_computation_input_data["service-handler-header"]["request-id"] = "request-4"
 214         self.path_computation_input_data["service-a-end"]["service-rate"] = "400"
 215         self.path_computation_input_data["service-z-end"]["service-rate"] = "400"
 216         self.path_computation_input_data["service-z-end"]["clli"] = "Node4"
 217         self.path_computation_input_data["hard-constraints"] =\
 218             {"include": {"node-id": ["OpenROADM-3", "OpenROADM-2", "OpenROADM-5"]}}
 219         response = test_utils.transportpce_api_rpc_request('transportpce-pce',
 220                                                            'path-computation-request',
 221                                                            self.path_computation_input_data)
 222         self.assertEqual(response['status_code'], requests.codes.ok)
 223         self.assertEqual(response['output']['configuration-response-common'][
 224             'response-code'], '500')
 225         self.assertEqual(response['output']['configuration-response-common'][
 226             'response-message'],
 227             'No path available by PCE and GNPy ')
 228         time.sleep(2)
 229
 230
 231 if __name__ == "__main__":
 232     # logging.basicConfig(filename='./transportpce_tests/log/response.log',filemode='w',level=logging.DEBUG)
 233     unittest.main(verbosity=2)