tests/transportpce_tests/pce/test02_pce_400G.py

   1 #!/usr/bin/env python
   2 ##############################################################################
   3 # Copyright (c) 2021 Orange, Inc. and others.  All rights reserved.
   4 #
   5 # All rights reserved. This program and the accompanying materials
   6 # are made available under the terms of the Apache License, Version 2.0
   7 # which accompanies this distribution, and is available at
   8 # http://www.apache.org/licenses/LICENSE-2.0
   9 ##############################################################################
  10
  11 # pylint: disable=invalid-name
  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 time
  20 import requests
  21 sys.path.append('transportpce_tests/common/')
  22 # pylint: disable=wrong-import-position
  23 # pylint: disable=import-error
  24 import test_utils  # nopep8
  25
  26
  27 class TransportPCE400Gtesting(unittest.TestCase):
  28     path_computation_input_data = {
  29         "service-name": "service-1",
  30         "resource-reserve": "true",
  31         "service-handler-header": {
  32             "request-id": "request1"
  33         },
  34         "service-a-end": {
  35             "service-rate": "400",
  36             "clli": "nodeA",
  37             "service-format": "Ethernet",
  38             "node-id": "XPDR-A2"
  39         },
  40         "service-z-end": {
  41             "service-rate": "400",
  42             "clli": "nodeC",
  43             "service-format": "Ethernet",
  44             "node-id": "XPDR-C2"
  45         },
  46         "pce-routing-metric": "hop-count"
  47     }
  48
  49     simple_topo_bi_dir_data = None
  50     port_mapping_data = None
  51     processes = None
  52
  53     @classmethod
  54     def setUpClass(cls):
  55         # pylint: disable=bare-except
  56         sample_files_parsed = False
  57         try:
  58             TOPO_BI_DIR_FILE = os.path.join(os.path.dirname(os.path.realpath(__file__)),
  59                                             "..", "..", "sample_configs",
  60                                             "honeynode-topo400G.json")
  61             with open(TOPO_BI_DIR_FILE, 'r', encoding='utf-8') as topo_bi_dir:
  62                 cls.topo_bi_dir_data = topo_bi_dir.read()
  63
  64             OTN_TOPO_BI_DIR_FILE = os.path.join(os.path.dirname(os.path.realpath(__file__)),
  65                                                 "..", "..", "sample_configs",
  66                                                 "honeynode-otntopo400G.json")
  67             with open(OTN_TOPO_BI_DIR_FILE, 'r', encoding='utf-8') as otn_topo_bi_dir:
  68                 cls.otn_topo_bi_dir_data = otn_topo_bi_dir.read()
  69
  70             OTUC4_OTN_TOPO_BI_DIR_FILE = os.path.join(os.path.dirname(os.path.realpath(__file__)),
  71                                                       "..", "..", "sample_configs",
  72                                                       "honeynode-otntopo400GwithOTUC4.json")
  73             with open(OTUC4_OTN_TOPO_BI_DIR_FILE, 'r', encoding='utf-8') as otuc4_otn_topo_bi_dir:
  74                 cls.otuc4_otn_topo_bi_dir_data = otuc4_otn_topo_bi_dir.read()
  75
  76             ODUC4_OTN_TOPO_BI_DIR_FILE = os.path.join(os.path.dirname(os.path.realpath(__file__)),
  77                                                       "..", "..", "sample_configs",
  78                                                       "honeynode-otntopo400GwithODUC4.json")
  79             with open(ODUC4_OTN_TOPO_BI_DIR_FILE, 'r', encoding='utf-8') as oduc4_otn_topo_bi_dir:
  80                 cls.oduc4_otn_topo_bi_dir_data = oduc4_otn_topo_bi_dir.read()
  81
  82             PORT_MAPPING_FILE = os.path.join(os.path.dirname(os.path.realpath(__file__)),
  83                                              "..", "..", "sample_configs",
  84                                              "pce_portmapping_71.json")
  85             with open(PORT_MAPPING_FILE, 'r', encoding='utf-8') as port_mapping:
  86                 cls.port_mapping_data = port_mapping.read()
  87             sample_files_parsed = True
  88         except PermissionError as err:
  89             print("Permission Error when trying to read sample files\n", err)
  90             sys.exit(2)
  91         except FileNotFoundError as err:
  92             print("File Not found Error when trying to read sample files\n", err)
  93             sys.exit(2)
  94         except:
  95             print("Unexpected error when trying to read sample files\n", sys.exc_info()[0])
  96             sys.exit(2)
  97         finally:
  98             if sample_files_parsed:
  99                 print("sample files content loaded")
 100         cls.processes = test_utils.start_tpce()
 101
 102     @classmethod
 103     def tearDownClass(cls):
 104         # clean datastores
 105         test_utils.del_portmapping()
 106         test_utils.del_ietf_network('openroadm-topology')
 107         test_utils.del_ietf_network('otn-topology')
 108         # pylint: disable=not-an-iterable
 109         for process in cls.processes:
 110             test_utils.shutdown_process(process)
 111         print("all processes killed")
 112
 113     def setUp(self):  # instruction executed before each test method
 114         # pylint: disable=consider-using-f-string
 115         print("execution of {}".format(self.id().split(".")[-1]))
 116         time.sleep(1)
 117
 118     # Load port mapping
 119     def test_01_load_port_mapping(self):
 120         response = test_utils.post_portmapping(self.port_mapping_data)
 121         self.assertIn(response['status_code'], (requests.codes.created, requests.codes.no_content))
 122         time.sleep(1)
 123
 124     # Load openroadm topology
 125     def test_02_load_openroadm_topology_bi(self):
 126         response = test_utils.put_ietf_network('openroadm-topology', self.topo_bi_dir_data)
 127         self.assertIn(response['status_code'], (requests.codes.ok, requests.codes.no_content))
 128         time.sleep(1)
 129
 130     # Path Computation success
 131     def test_03_path_computation_400G_xpdr_bi(self):
 132         response = test_utils.transportpce_api_rpc_request('transportpce-pce',
 133                                                            'path-computation-request',
 134                                                            self.path_computation_input_data)
 135         self.assertEqual(response['status_code'], requests.codes.ok)
 136         self.assertIn('Path is calculated',
 137                       response['output']['configuration-response-common']['response-message'])
 138
 139         self.assertEqual(1, response['output']['response-parameters']['path-description']
 140                          ['aToZ-direction']['aToZ-wavelength-number'])
 141         self.assertEqual(400, response['output']['response-parameters']['path-description']
 142                          ['aToZ-direction']['rate'])
 143         self.assertEqual(196.0375, float(response['output']['response-parameters']['path-description']
 144                          ['aToZ-direction']['aToZ-min-frequency']))
 145         self.assertEqual(196.12500, float(response['output']['response-parameters']['path-description']
 146                          ['aToZ-direction']['aToZ-max-frequency']))
 147         self.assertEqual('dp-qam16', response['output']['response-parameters']['path-description']
 148                          ['aToZ-direction']['modulation-format'])
 149
 150         self.assertEqual(1, response['output']['response-parameters']['path-description']
 151                          ['zToA-direction']['zToA-wavelength-number'])
 152         self.assertEqual(400, response['output']['response-parameters']['path-description']
 153                          ['zToA-direction']['rate'])
 154         self.assertEqual(196.0375, float(response['output']['response-parameters']['path-description']
 155                          ['zToA-direction']['zToA-min-frequency']))
 156         self.assertEqual(196.12500, float(response['output']['response-parameters']['path-description']
 157                          ['zToA-direction']['zToA-max-frequency']))
 158         self.assertEqual('dp-qam16', response['output']['response-parameters']['path-description']
 159                          ['zToA-direction']['modulation-format'])
 160         time.sleep(2)
 161
 162     # Load otn topology
 163     def test_04_load_otn_topology_bi(self):
 164         response = test_utils.put_ietf_network('otn-topology', self.otn_topo_bi_dir_data)
 165         self.assertIn(response['status_code'], (requests.codes.ok, requests.codes.no_content))
 166         time.sleep(1)
 167
 168     # Path Computation success
 169     def test_05_path_computation_OTUC4_xpdr_bi(self):
 170         self.path_computation_input_data["service-name"] = "service-OTUC4"
 171         self.path_computation_input_data["service-a-end"]["service-format"] = "OTU"
 172         self.path_computation_input_data["service-a-end"]["tx-direction"] =\
 173             {"port": {"port-device-name": "XPDR-A2-XPDR2"}}
 174         self.path_computation_input_data["service-z-end"]["service-format"] = "OTU"
 175         self.path_computation_input_data["service-z-end"]["tx-direction"] =\
 176             {"port": {"port-device-name": "XPDR-C2-XPDR2"}}
 177         response = test_utils.transportpce_api_rpc_request('transportpce-pce',
 178                                                            'path-computation-request',
 179                                                            self.path_computation_input_data)
 180         self.assertEqual(response['status_code'], requests.codes.ok)
 181         self.assertIn('Path is calculated',
 182                       response['output']['configuration-response-common']['response-message'])
 183
 184         self.assertEqual(1, response['output']['response-parameters']['path-description']
 185                          ['aToZ-direction']['aToZ-wavelength-number'])
 186         self.assertEqual(400, response['output']['response-parameters']['path-description']
 187                          ['aToZ-direction']['rate'])
 188         self.assertEqual(196.0375, float(response['output']['response-parameters']['path-description']
 189                          ['aToZ-direction']['aToZ-min-frequency']))
 190         self.assertEqual(196.12500, float(response['output']['response-parameters']['path-description']
 191                          ['aToZ-direction']['aToZ-max-frequency']))
 192         self.assertEqual('dp-qam16', response['output']['response-parameters']['path-description']
 193                          ['aToZ-direction']['modulation-format'])
 194
 195         self.assertEqual(1, response['output']['response-parameters']['path-description']
 196                          ['zToA-direction']['zToA-wavelength-number'])
 197         self.assertEqual(400, response['output']['response-parameters']['path-description']
 198                          ['zToA-direction']['rate'])
 199         self.assertEqual(196.0375, float(response['output']['response-parameters']['path-description']
 200                          ['zToA-direction']['zToA-min-frequency']))
 201         self.assertEqual(196.12500, float(response['output']['response-parameters']['path-description']
 202                          ['zToA-direction']['zToA-max-frequency']))
 203         self.assertEqual('dp-qam16', response['output']['response-parameters']['path-description']
 204                          ['zToA-direction']['modulation-format'])
 205         time.sleep(2)
 206
 207     # Load otn topology with OTUC4 links
 208     def test_06_load_otuc4_otn_topology_bi(self):
 209         response = test_utils.put_ietf_network('otn-topology', self.otuc4_otn_topo_bi_dir_data)
 210         self.assertIn(response['status_code'], (requests.codes.ok, requests.codes.no_content))
 211         time.sleep(1)
 212
 213     # Path Computation success
 214     def test_07_path_computation_ODUC4_xpdr_bi(self):
 215         self.path_computation_input_data["service-name"] = "service-ODUC4"
 216         self.path_computation_input_data["service-a-end"]["service-format"] = "ODU"
 217         self.path_computation_input_data["service-a-end"]["tx-direction"] = \
 218             {"port": {"port-device-name": "XPDR-A2-XPDR2"}}
 219         self.path_computation_input_data["service-z-end"]["service-format"] = "ODU"
 220         self.path_computation_input_data["service-z-end"]["tx-direction"] = \
 221             {"port": {"port-device-name": "XPDR-C2-XPDR2"}}
 222         response = test_utils.transportpce_api_rpc_request('transportpce-pce',
 223                                                            'path-computation-request',
 224                                                            self.path_computation_input_data)
 225         self.assertEqual(response['status_code'], requests.codes.ok)
 226         self.assertIn('Path is calculated',
 227                       response['output']['configuration-response-common']['response-message'])
 228
 229         self.assertEqual(400, response['output']['response-parameters']['path-description']
 230                          ['aToZ-direction']['rate'])
 231         self.assertEqual('dp-qam16', response['output']['response-parameters']['path-description']
 232                          ['aToZ-direction']['modulation-format'])
 233
 234         self.assertEqual(400, response['output']['response-parameters']['path-description']
 235                          ['zToA-direction']['rate'])
 236         self.assertEqual('dp-qam16', response['output']['response-parameters']['path-description']
 237                          ['zToA-direction']['modulation-format'])
 238         time.sleep(2)
 239
 240     # Load otn topology with OTUC4 links
 241     def test_08_load_oduc4_otn_topology_bi(self):
 242         response = test_utils.put_ietf_network('otn-topology', self.oduc4_otn_topo_bi_dir_data)
 243         self.assertIn(response['status_code'], (requests.codes.ok, requests.codes.no_content))
 244         time.sleep(1)
 245
 246     # Path Computation success
 247     def test_09_path_computation_100G_xpdr_bi(self):
 248         self.path_computation_input_data["service-name"] = "service-100GE"
 249         self.path_computation_input_data["service-a-end"]["service-rate"] = "100"
 250         self.path_computation_input_data["service-a-end"]["service-format"] = "Ethernet"
 251         self.path_computation_input_data["service-a-end"]["tx-direction"] = \
 252             {"port": {"port-device-name": "XPDR-A2-XPDR2",
 253                       "port-name": "XPDR2-CLIENT1"}}
 254         self.path_computation_input_data["service-z-end"]["service-rate"] = "100"
 255         self.path_computation_input_data["service-z-end"]["service-format"] = "Ethernet"
 256         self.path_computation_input_data["service-z-end"]["tx-direction"] = \
 257             {"port": {"port-device-name": "XPDR-C2-XPDR2",
 258                       "port-name": "XPDR2-CLIENT1"}}
 259         self.path_computation_input_data["service-z-end"]["service-format"] = "ODU"
 260         response = test_utils.transportpce_api_rpc_request('transportpce-pce',
 261                                                            'path-computation-request',
 262                                                            self.path_computation_input_data)
 263         self.assertEqual(response['status_code'], requests.codes.ok)
 264         self.assertIn('Path is calculated',
 265                       response['output']['configuration-response-common']['response-message'])
 266
 267         self.assertEqual(100, response['output']['response-parameters']['path-description']
 268                          ['aToZ-direction']['rate'])
 269         self.assertEqual('1.1', response['output']['response-parameters']['path-description']
 270                          ['aToZ-direction']['min-trib-slot'])
 271         self.assertEqual('1.20', response['output']['response-parameters']['path-description']
 272                          ['aToZ-direction']['max-trib-slot'])
 273         self.assertEqual('dp-qpsk', response['output']['response-parameters']['path-description']
 274                          ['aToZ-direction']['modulation-format'])
 275
 276         self.assertEqual(100, response['output']['response-parameters']['path-description']
 277                          ['zToA-direction']['rate'])
 278         self.assertEqual('1.1', response['output']['response-parameters']['path-description']
 279                          ['zToA-direction']['min-trib-slot'])
 280         self.assertEqual('1.20', response['output']['response-parameters']['path-description']
 281                          ['zToA-direction']['max-trib-slot'])
 282         self.assertEqual('dp-qpsk', response['output']['response-parameters']['path-description']
 283                          ['zToA-direction']['modulation-format'])
 284         time.sleep(2)
 285
 286
 287 if __name__ == "__main__":
 288     unittest.main(verbosity=2)