tests/transportpce_tests/1.2.1/test04_renderer_service_path_nominal.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 time
  17 import requests
  18 # pylint: disable=wrong-import-order
  19 import sys
  20 sys.path.append('transportpce_tests/common/')
  21 # pylint: disable=wrong-import-position
  22 # pylint: disable=import-error
  23 import test_utils_rfc8040  # nopep8
  24
  25
  26 class TransportPCERendererTesting(unittest.TestCase):
  27
  28     processes = None
  29     NODE_VERSION = '1.2.1'
  30
  31     @classmethod
  32     def setUpClass(cls):
  33         cls.processes = test_utils_rfc8040.start_tpce()
  34         cls.processes = test_utils_rfc8040.start_sims([('xpdra', cls.NODE_VERSION), ('roadma', cls.NODE_VERSION)])
  35
  36     @classmethod
  37     def tearDownClass(cls):
  38         # pylint: disable=not-an-iterable
  39         for process in cls.processes:
  40             test_utils_rfc8040.shutdown_process(process)
  41         print("all processes killed")
  42
  43     def setUp(self):
  44         # pylint: disable=consider-using-f-string
  45         print("execution of {}".format(self.id().split(".")[-1]))
  46         time.sleep(10)
  47
  48     def test_01_rdm_device_connected(self):
  49         response = test_utils_rfc8040.mount_device("ROADMA01", ('roadma', self.NODE_VERSION))
  50         self.assertEqual(response.status_code, requests.codes.created, test_utils_rfc8040.CODE_SHOULD_BE_201)
  51
  52     def test_02_xpdr_device_connected(self):
  53         response = test_utils_rfc8040.mount_device("XPDRA01", ('xpdra', self.NODE_VERSION))
  54         self.assertEqual(response.status_code, requests.codes.created, test_utils_rfc8040.CODE_SHOULD_BE_201)
  55
  56     def test_03_rdm_portmapping(self):
  57         response = test_utils_rfc8040.get_portmapping("ROADMA01")
  58         self.assertEqual(response['status_code'], requests.codes.ok)
  59         self.assertIn(
  60             {'supporting-port': 'L1', 'supporting-circuit-pack-name': '2/0',
  61              'logical-connection-point': 'DEG1-TTP-TXRX', 'port-direction': 'bidirectional',
  62              'port-admin-state': 'InService', 'port-oper-state': 'InService'},
  63             response['nodes'][0]['mapping'])
  64         self.assertIn(
  65             {'supporting-port': 'C7', 'supporting-circuit-pack-name': '4/0',
  66              'logical-connection-point': 'SRG1-PP7-TXRX', 'port-direction': 'bidirectional',
  67              'port-admin-state': 'InService', 'port-oper-state': 'InService'},
  68             response['nodes'][0]['mapping'])
  69
  70     def test_04_xpdr_portmapping(self):
  71         response = test_utils_rfc8040.get_portmapping("XPDRA01")
  72         self.assertEqual(response['status_code'], requests.codes.ok)
  73         self.assertIn(
  74             {'supporting-port': '1', 'supporting-circuit-pack-name': '1/0/1-PLUG-NET',
  75              'logical-connection-point': 'XPDR1-NETWORK1', 'port-direction': 'bidirectional',
  76              'connection-map-lcp': 'XPDR1-CLIENT1', 'port-qual': 'xpdr-network',
  77              'lcp-hash-val': 'OSvMgUyP+mE=', 'xponder-type': 'tpdr',
  78              'port-admin-state': 'InService', 'port-oper-state': 'InService'},
  79             response['nodes'][0]['mapping'])
  80         self.assertIn(
  81             {'supporting-port': 'C1',
  82              'supporting-circuit-pack-name': '1/0/C1-PLUG-CLIENT',
  83              'logical-connection-point': 'XPDR1-CLIENT1', 'port-direction': 'bidirectional',
  84              'connection-map-lcp': 'XPDR1-NETWORK1', 'port-qual': 'xpdr-client',
  85              'lcp-hash-val': 'AO9UFkY/TLYw', 'xponder-type': 'tpdr',
  86              'port-admin-state': 'InService', 'port-oper-state': 'InService'},
  87             response['nodes'][0]['mapping'])
  88
  89     def test_05_service_path_create(self):
  90         response = test_utils_rfc8040.device_renderer_service_path_request(
  91             {
  92                 'service-name': 'service_test',
  93                 'wave-number': '7',
  94                 'modulation-format': 'dp-qpsk',
  95                 'operation': 'create',
  96                 'nodes': [{'node-id': 'ROADMA01', 'src-tp': 'SRG1-PP7-TXRX', 'dest-tp': 'DEG1-TTP-TXRX'},
  97                           {'node-id': 'XPDRA01', 'src-tp': 'XPDR1-CLIENT1', 'dest-tp': 'XPDR1-NETWORK1'}],
  98                 'center-freq': 195.8,
  99                 'nmc-width': 40,
 100                 'min-freq': 195.775,
 101                 'max-freq': 195.825,
 102                 'lower-spectral-slot-number': 713,
 103                 'higher-spectral-slot-number': 720
 104             })
 105         self.assertEqual(response['status_code'], requests.codes.ok)
 106         self.assertIn('Interfaces created successfully for nodes: ROADMA01', response['output']['result'])
 107
 108     def test_06_service_path_create_rdm_check(self):
 109         response = test_utils_rfc8040.check_node_attribute_request("ROADMA01", "interface", "DEG1-TTP-TXRX-713:720")
 110         self.assertEqual(response['status_code'], requests.codes.ok)
 111         # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
 112         self.assertDictEqual(
 113             dict({
 114                 'name': 'DEG1-TTP-TXRX-713:720',
 115                 'administrative-state': 'inService',
 116                 'supporting-circuit-pack-name': '2/0',
 117                 'type': 'org-openroadm-interfaces:opticalChannel',
 118                 'supporting-port': 'L1'
 119             }, **response['interface'][0]), response['interface'][0])
 120         self.assertDictEqual(
 121             {'wavelength-number': 7},
 122             response['interface'][0]['org-openroadm-optical-channel-interfaces:och'])
 123
 124     def test_07_service_path_create_rdm_check(self):
 125         response = test_utils_rfc8040.check_node_attribute_request("ROADMA01", "interface", "SRG1-PP7-TXRX-713:720")
 126         self.assertEqual(response['status_code'], requests.codes.ok)
 127         # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
 128         self.assertDictEqual(
 129             dict({
 130                 'name': 'SRG1-PP7-TXRX-713:720',
 131                 'administrative-state': 'inService',
 132                 'supporting-circuit-pack-name': '4/0',
 133                 'type': 'org-openroadm-interfaces:opticalChannel',
 134                 'supporting-port': 'C7'
 135             }, **response['interface'][0]), response['interface'][0])
 136         self.assertDictEqual(
 137             {'wavelength-number': 7},
 138             response['interface'][0]['org-openroadm-optical-channel-interfaces:och'])
 139
 140     def test_08_service_path_create_rdm_check(self):
 141         response = test_utils_rfc8040.check_node_attribute_request(
 142             "ROADMA01", "roadm-connections", "SRG1-PP7-TXRX-DEG1-TTP-TXRX-713:720")
 143         self.assertEqual(response['status_code'], requests.codes.ok)
 144         # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
 145         self.assertDictEqual(
 146             dict({
 147                 'connection-number': 'SRG1-PP7-TXRX-DEG1-TTP-TXRX-713:720',
 148                 'wavelength-number': 7,
 149                 'opticalControlMode': 'off'
 150             }, **response['roadm-connections'][0]), response['roadm-connections'][0])
 151         self.assertDictEqual({'src-if': 'SRG1-PP7-TXRX-713:720'}, response['roadm-connections'][0]['source'])
 152         self.assertDictEqual({'dst-if': 'DEG1-TTP-TXRX-713:720'}, response['roadm-connections'][0]['destination'])
 153
 154     def test_09_service_path_create_xpdr_check(self):
 155         response = test_utils_rfc8040.check_node_attribute_request("XPDRA01", "interface", "XPDR1-NETWORK1-713:720")
 156         self.assertEqual(response['status_code'], requests.codes.ok)
 157         # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
 158         self.assertDictEqual(
 159             dict({
 160                 'name': 'XPDR1-NETWORK1-713:720',
 161                 'administrative-state': 'inService',
 162                 'supporting-circuit-pack-name': '1/0/1-PLUG-NET',
 163                 'type': 'org-openroadm-interfaces:opticalChannel',
 164                 'supporting-port': '1'
 165             }, **response['interface'][0]), response['interface'][0])
 166         self.assertIn(
 167             response['interface'][0]['org-openroadm-optical-channel-interfaces:och'],
 168             [{'rate': 'org-openroadm-optical-channel-interfaces:R100G',
 169               'transmit-power': '-5',
 170               'wavelength-number': 7,
 171               'modulation-format': 'dp-qpsk'},
 172              {'rate': 'org-openroadm-optical-channel-interfaces:R100G',
 173               'transmit-power': -5,
 174               'wavelength-number': 7,
 175               'modulation-format': 'dp-qpsk'}])
 176
 177     def test_10_service_path_create_xpdr_check(self):
 178         response = test_utils_rfc8040.check_node_attribute_request("XPDRA01", "interface", "XPDR1-NETWORK1-OTU")
 179         self.assertEqual(response['status_code'], requests.codes.ok)
 180         # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
 181         self.assertDictEqual(
 182             dict({
 183                 'name': 'XPDR1-NETWORK1-OTU',
 184                 'administrative-state': 'inService',
 185                 'supporting-circuit-pack-name': '1/0/1-PLUG-NET',
 186                 'type': 'org-openroadm-interfaces:otnOtu',
 187                 'supporting-port': '1',
 188                 'supporting-interface': 'XPDR1-NETWORK1-713:720'
 189             }, **response['interface'][0]), response['interface'][0])
 190         self.assertDictEqual(
 191             {'rate': 'org-openroadm-otn-otu-interfaces:OTU4', 'fec': 'scfec'},
 192             response['interface'][0]['org-openroadm-otn-otu-interfaces:otu'])
 193
 194     def test_11_service_path_create_xpdr_check(self):
 195         response = test_utils_rfc8040.check_node_attribute_request("XPDRA01", "interface", "XPDR1-NETWORK1-ODU")
 196         self.assertEqual(response['status_code'], requests.codes.ok)
 197         # the 2 following statements replace self.assertDictContainsSubset deprecated in python 3.2
 198         self.assertDictEqual(
 199             dict({
 200                 'name': 'XPDR1-NETWORK1-ODU',
 201                 'administrative-state': 'inService',
 202                 'supporting-circuit-pack-name': '1/0/1-PLUG-NET',
 203                 'type': 'org-openroadm-interfaces:otnOdu',
 204                 'supporting-port': '1',
 205                 'supporting-interface': 'XPDR1-NETWORK1-OTU'
 206             }, **response['interface'][0]), response['interface'][0])
 207         self.assertDictEqual(
 208             dict({
 209                 'rate': 'org-openroadm-otn-odu-interfaces:ODU4',
 210                 'monitoring-mode': 'terminated'
 211             }, **response['interface'][0]['org-openroadm-otn-odu-interfaces:odu']),
 212             response['interface'][0]['org-openroadm-otn-odu-interfaces:odu']
 213         )
 214         self.assertDictEqual({'exp-payload-type': '07', 'payload-type': '07'},
 215                              response['interface'][0]['org-openroadm-otn-odu-interfaces:odu']['opu'])
 216
 217     def test_12_service_path_create_xpdr_check(self):
 218         response = test_utils_rfc8040.check_node_attribute_request("XPDRA01", "interface", "XPDR1-CLIENT1-ETHERNET")
 219         self.assertEqual(response['status_code'], requests.codes.ok)
 220         # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
 221         self.assertDictEqual(
 222             dict({
 223                 'name': 'XPDR1-CLIENT1-ETHERNET',
 224                 'administrative-state': 'inService',
 225                 'supporting-circuit-pack-name': '1/0/C1-PLUG-CLIENT',
 226                 'type': 'org-openroadm-interfaces:ethernetCsmacd',
 227                 'supporting-port': 'C1'
 228             }, **response['interface'][0]), response['interface'][0])
 229         self.assertDictEqual(
 230             {'speed': 100000,
 231              'mtu': 9000,
 232              'auto-negotiation': 'enabled',
 233              'duplex': 'full',
 234              'fec': 'off'},
 235             response['interface'][0]['org-openroadm-ethernet-interfaces:ethernet'])
 236
 237     def test_13_service_path_create_xpdr_check(self):
 238         response = test_utils_rfc8040.check_node_attribute_request("XPDRA01", "circuit-packs", "1%2F0%2F1-PLUG-NET")
 239         # FIXME: https://jira.opendaylight.org/browse/TRNSPRTPCE-591
 240         self.assertEqual(response['status_code'], requests.codes.ok)
 241         self.assertIn('not-reserved-inuse', response['circuit-packs'][0]["equipment-state"])
 242
 243     def test_14_service_path_delete(self):
 244         response = test_utils_rfc8040.device_renderer_service_path_request(
 245             {
 246                 'service-name': 'service_test',
 247                 'wave-number': '7',
 248                 'modulation-format': 'dp-qpsk',
 249                 'operation': 'delete',
 250                 'nodes': [{'node-id': 'ROADMA01', 'src-tp': 'SRG1-PP7-TXRX', 'dest-tp': 'DEG1-TTP-TXRX'},
 251                           {'node-id': 'XPDRA01', 'src-tp': 'XPDR1-CLIENT1', 'dest-tp': 'XPDR1-NETWORK1'}],
 252                 'center-freq': 195.8,
 253                 'nmc-width': 40,
 254                 'min-freq': 195.775,
 255                 'max-freq': 195.825,
 256                 'lower-spectral-slot-number': 713,
 257                 'higher-spectral-slot-number': 720
 258             })
 259         self.assertEqual(response['status_code'], requests.codes.ok)
 260         self.assertDictEqual(response['output'], {'result': 'Request processed', 'success': True})
 261
 262     def test_15_service_path_delete_rdm_check(self):
 263         response = test_utils_rfc8040.check_node_attribute_request("ROADMA01", "interface", "DEG1-TTP-TXRX-713:720")
 264         self.assertEqual(response['status_code'], requests.codes.conflict)
 265
 266     def test_16_service_path_delete_rdm_check(self):
 267         response = test_utils_rfc8040.check_node_attribute_request("ROADMA01", "interface", "SRG1-PP7-TXRX-713:720")
 268         self.assertEqual(response['status_code'], requests.codes.conflict)
 269
 270     def test_17_service_path_delete_rdm_check(self):
 271         response = test_utils_rfc8040.check_node_attribute_request(
 272             "ROADMA01", "roadm-connections", "SRG1-PP7-TXRX-DEG1-TTP-TXRX-713:720")
 273         self.assertEqual(response['status_code'], requests.codes.conflict)
 274
 275     def test_18_service_path_delete_xpdr_check(self):
 276         response = test_utils_rfc8040.check_node_attribute_request("XPDRA01", "interface", "XPDR1-NETWORK1-713:720")
 277         self.assertEqual(response['status_code'], requests.codes.conflict)
 278
 279     def test_19_service_path_delete_xpdr_check(self):
 280         response = test_utils_rfc8040.check_node_attribute_request("XPDRA01", "interface", "XPDR1-NETWORK1-OTU")
 281         self.assertEqual(response['status_code'], requests.codes.conflict)
 282
 283     def test_20_service_path_delete_xpdr_check(self):
 284         response = test_utils_rfc8040.check_node_attribute_request("XPDRA01", "interface", "XPDR1-NETWORK1-ODU")
 285         self.assertEqual(response['status_code'], requests.codes.conflict)
 286
 287     def test_21_service_path_delete_xpdr_check(self):
 288         response = test_utils_rfc8040.check_node_attribute_request("XPDRA01", "interface", "XPDR1-CLIENT1-ETHERNET")
 289         self.assertEqual(response['status_code'], requests.codes.conflict)
 290
 291     def test_22_service_path_delete_xpdr_check(self):
 292         response = test_utils_rfc8040.check_node_attribute_request("XPDRA01", "circuit-packs", "1%2F0%2F1-PLUG-NET")
 293         self.assertEqual(response['status_code'], requests.codes.ok)
 294         self.assertEqual('not-reserved-available', response["circuit-packs"][0]['equipment-state'])
 295
 296     def test_23_rdm_device_disconnected(self):
 297         response = test_utils_rfc8040.unmount_device("ROADMA01")
 298         self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
 299
 300     def test_24_xpdr_device_disconnected(self):
 301         response = test_utils_rfc8040.unmount_device("XPDRA01")
 302         self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
 303
 304
 305 if __name__ == "__main__":
 306     unittest.main(verbosity=2, failfast=False)