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  # 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.start_tpce()
  34         cls.processes = test_utils.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.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(2)
  47
  48     def test_01_rdm_device_connected(self):
  49         response = test_utils.mount_device("ROADMA01", ('roadma', self.NODE_VERSION))
  50         self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
  51
  52     def test_02_xpdr_device_connected(self):
  53         response = test_utils.mount_device("XPDRA01", ('xpdra', self.NODE_VERSION))
  54         self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
  55
  56     def test_03_rdm_portmapping(self):
  57         response = test_utils.get_portmapping_node_attr("ROADMA01", None, None)
  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.get_portmapping_node_attr("XPDRA01", None, None)
  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=', 'xpdr-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', 'xpdr-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.transportpce_api_rpc_request(
  91             'transportpce-device-renderer', 'service-path',
  92             {
  93                 'service-name': 'service_test',
  94                 'wave-number': '7',
  95                 'modulation-format': 'dp-qpsk',
  96                 'operation': 'create',
  97                 'nodes': [{'node-id': 'ROADMA01', 'src-tp': 'SRG1-PP7-TXRX', 'dest-tp': 'DEG1-TTP-TXRX'},
  98                           {'node-id': 'XPDRA01', 'src-tp': 'XPDR1-CLIENT1', 'dest-tp': 'XPDR1-NETWORK1'}],
  99                 'center-freq': 195.8,
 100                 'nmc-width': 40,
 101                 'min-freq': 195.775,
 102                 'max-freq': 195.825,
 103                 'lower-spectral-slot-number': 713,
 104                 'higher-spectral-slot-number': 720
 105             })
 106         self.assertEqual(response['status_code'], requests.codes.ok)
 107         self.assertIn('Interfaces created successfully for nodes: ROADMA01', response['output']['result'])
 108
 109     def test_06_service_path_create_rdm_check(self):
 110         response = test_utils.check_node_attribute_request("ROADMA01", "interface", "DEG1-TTP-TXRX-713:720")
 111         self.assertEqual(response['status_code'], requests.codes.ok)
 112         # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
 113         self.assertDictEqual(
 114             dict({
 115                 'name': 'DEG1-TTP-TXRX-713:720',
 116                 'administrative-state': 'inService',
 117                 'supporting-circuit-pack-name': '2/0',
 118                 'type': 'org-openroadm-interfaces:opticalChannel',
 119                 'supporting-port': 'L1'
 120             }, **response['interface'][0]), response['interface'][0])
 121         self.assertDictEqual(
 122             {'wavelength-number': 7},
 123             response['interface'][0]['org-openroadm-optical-channel-interfaces:och'])
 124
 125     def test_07_service_path_create_rdm_check(self):
 126         response = test_utils.check_node_attribute_request("ROADMA01", "interface", "SRG1-PP7-TXRX-713:720")
 127         self.assertEqual(response['status_code'], requests.codes.ok)
 128         # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
 129         self.assertDictEqual(
 130             dict({
 131                 'name': 'SRG1-PP7-TXRX-713:720',
 132                 'administrative-state': 'inService',
 133                 'supporting-circuit-pack-name': '4/0',
 134                 'type': 'org-openroadm-interfaces:opticalChannel',
 135                 'supporting-port': 'C7'
 136             }, **response['interface'][0]), response['interface'][0])
 137         self.assertDictEqual(
 138             {'wavelength-number': 7},
 139             response['interface'][0]['org-openroadm-optical-channel-interfaces:och'])
 140
 141     def test_08_service_path_create_rdm_check(self):
 142         response = test_utils.check_node_attribute_request(
 143             "ROADMA01", "roadm-connections", "SRG1-PP7-TXRX-DEG1-TTP-TXRX-713:720")
 144         self.assertEqual(response['status_code'], requests.codes.ok)
 145         # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
 146         self.assertDictEqual(
 147             dict({
 148                 'connection-number': 'SRG1-PP7-TXRX-DEG1-TTP-TXRX-713:720',
 149                 'wavelength-number': 7,
 150                 'opticalControlMode': 'off'
 151             }, **response['roadm-connections'][0]), response['roadm-connections'][0])
 152         self.assertDictEqual({'src-if': 'SRG1-PP7-TXRX-713:720'}, response['roadm-connections'][0]['source'])
 153         self.assertDictEqual({'dst-if': 'DEG1-TTP-TXRX-713:720'}, response['roadm-connections'][0]['destination'])
 154
 155     def test_09_service_path_create_xpdr_check(self):
 156         response = test_utils.check_node_attribute_request("XPDRA01", "interface", "XPDR1-NETWORK1-713:720")
 157         self.assertEqual(response['status_code'], requests.codes.ok)
 158         # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
 159         self.assertDictEqual(
 160             dict({
 161                 'name': 'XPDR1-NETWORK1-713:720',
 162                 'administrative-state': 'inService',
 163                 'supporting-circuit-pack-name': '1/0/1-PLUG-NET',
 164                 'type': 'org-openroadm-interfaces:opticalChannel',
 165                 'supporting-port': '1'
 166             }, **response['interface'][0]), response['interface'][0])
 167         intf = response['interface'][0]['org-openroadm-optical-channel-interfaces:och']
 168         self.assertEqual(intf['rate'], 'org-openroadm-optical-channel-interfaces:R100G')
 169         self.assertEqual(intf['modulation-format'], 'dp-qpsk')
 170         self.assertEqual(intf['wavelength-number'], 7)
 171         self.assertEqual(float(intf['transmit-power']), -5)
 172
 173     def test_10_service_path_create_xpdr_check(self):
 174         response = test_utils.check_node_attribute_request("XPDRA01", "interface", "XPDR1-NETWORK1-OTU")
 175         self.assertEqual(response['status_code'], requests.codes.ok)
 176         # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
 177         self.assertDictEqual(
 178             dict({
 179                 'name': 'XPDR1-NETWORK1-OTU',
 180                 'administrative-state': 'inService',
 181                 'supporting-circuit-pack-name': '1/0/1-PLUG-NET',
 182                 'type': 'org-openroadm-interfaces:otnOtu',
 183                 'supporting-port': '1',
 184                 'supporting-interface': 'XPDR1-NETWORK1-713:720'
 185             }, **response['interface'][0]), response['interface'][0])
 186         self.assertDictEqual(
 187             {'rate': 'org-openroadm-otn-otu-interfaces:OTU4', 'fec': 'scfec'},
 188             response['interface'][0]['org-openroadm-otn-otu-interfaces:otu'])
 189
 190     def test_11_service_path_create_xpdr_check(self):
 191         response = test_utils.check_node_attribute_request("XPDRA01", "interface", "XPDR1-NETWORK1-ODU")
 192         self.assertEqual(response['status_code'], requests.codes.ok)
 193         # the 2 following statements replace self.assertDictContainsSubset deprecated in python 3.2
 194         self.assertDictEqual(
 195             dict({
 196                 'name': 'XPDR1-NETWORK1-ODU',
 197                 'administrative-state': 'inService',
 198                 'supporting-circuit-pack-name': '1/0/1-PLUG-NET',
 199                 'type': 'org-openroadm-interfaces:otnOdu',
 200                 'supporting-port': '1',
 201                 'supporting-interface': 'XPDR1-NETWORK1-OTU'
 202             }, **response['interface'][0]), response['interface'][0])
 203         self.assertDictEqual(
 204             dict({
 205                 'rate': 'org-openroadm-otn-odu-interfaces:ODU4',
 206                 'monitoring-mode': 'terminated'
 207             }, **response['interface'][0]['org-openroadm-otn-odu-interfaces:odu']),
 208             response['interface'][0]['org-openroadm-otn-odu-interfaces:odu']
 209         )
 210         self.assertDictEqual({'exp-payload-type': '07', 'payload-type': '07'},
 211                              response['interface'][0]['org-openroadm-otn-odu-interfaces:odu']['opu'])
 212
 213     def test_12_service_path_create_xpdr_check(self):
 214         response = test_utils.check_node_attribute_request("XPDRA01", "interface", "XPDR1-CLIENT1-ETHERNET")
 215         self.assertEqual(response['status_code'], requests.codes.ok)
 216         # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
 217         self.assertDictEqual(
 218             dict({
 219                 'name': 'XPDR1-CLIENT1-ETHERNET',
 220                 'administrative-state': 'inService',
 221                 'supporting-circuit-pack-name': '1/0/C1-PLUG-CLIENT',
 222                 'type': 'org-openroadm-interfaces:ethernetCsmacd',
 223                 'supporting-port': 'C1'
 224             }, **response['interface'][0]), response['interface'][0])
 225         self.assertDictEqual(
 226             {'speed': 100000,
 227              'mtu': 9000,
 228              'auto-negotiation': 'enabled',
 229              'duplex': 'full',
 230              'fec': 'off'},
 231             response['interface'][0]['org-openroadm-ethernet-interfaces:ethernet'])
 232
 233     def test_13_service_path_create_xpdr_check(self):
 234         response = test_utils.check_node_attribute_request("XPDRA01", "circuit-packs", "1%2F0%2F1-PLUG-NET")
 235         # FIXME: https://jira.opendaylight.org/browse/TRNSPRTPCE-591
 236         self.assertEqual(response['status_code'], requests.codes.ok)
 237         self.assertIn('not-reserved-inuse', response['circuit-packs'][0]["equipment-state"])
 238
 239     def test_14_service_path_delete(self):
 240         response = test_utils.transportpce_api_rpc_request(
 241             'transportpce-device-renderer', 'service-path',
 242             {
 243                 'service-name': 'service_test',
 244                 'wave-number': '7',
 245                 'modulation-format': 'dp-qpsk',
 246                 'operation': 'delete',
 247                 'nodes': [{'node-id': 'ROADMA01', 'src-tp': 'SRG1-PP7-TXRX', 'dest-tp': 'DEG1-TTP-TXRX'},
 248                           {'node-id': 'XPDRA01', 'src-tp': 'XPDR1-CLIENT1', 'dest-tp': 'XPDR1-NETWORK1'}],
 249                 'center-freq': 195.8,
 250                 'nmc-width': 40,
 251                 'min-freq': 195.775,
 252                 'max-freq': 195.825,
 253                 'lower-spectral-slot-number': 713,
 254                 'higher-spectral-slot-number': 720
 255             })
 256         self.assertEqual(response['status_code'], requests.codes.ok)
 257         self.assertDictEqual(response['output'], {'result': 'Request processed', 'success': True})
 258
 259     def test_15_service_path_delete_rdm_check(self):
 260         response = test_utils.check_node_attribute_request("ROADMA01", "interface", "DEG1-TTP-TXRX-713:720")
 261         self.assertEqual(response['status_code'], requests.codes.conflict)
 262
 263     def test_16_service_path_delete_rdm_check(self):
 264         response = test_utils.check_node_attribute_request("ROADMA01", "interface", "SRG1-PP7-TXRX-713:720")
 265         self.assertEqual(response['status_code'], requests.codes.conflict)
 266
 267     def test_17_service_path_delete_rdm_check(self):
 268         response = test_utils.check_node_attribute_request(
 269             "ROADMA01", "roadm-connections", "SRG1-PP7-TXRX-DEG1-TTP-TXRX-713:720")
 270         self.assertEqual(response['status_code'], requests.codes.conflict)
 271
 272     def test_18_service_path_delete_xpdr_check(self):
 273         response = test_utils.check_node_attribute_request("XPDRA01", "interface", "XPDR1-NETWORK1-713:720")
 274         self.assertEqual(response['status_code'], requests.codes.conflict)
 275
 276     def test_19_service_path_delete_xpdr_check(self):
 277         response = test_utils.check_node_attribute_request("XPDRA01", "interface", "XPDR1-NETWORK1-OTU")
 278         self.assertEqual(response['status_code'], requests.codes.conflict)
 279
 280     def test_20_service_path_delete_xpdr_check(self):
 281         response = test_utils.check_node_attribute_request("XPDRA01", "interface", "XPDR1-NETWORK1-ODU")
 282         self.assertEqual(response['status_code'], requests.codes.conflict)
 283
 284     def test_21_service_path_delete_xpdr_check(self):
 285         response = test_utils.check_node_attribute_request("XPDRA01", "interface", "XPDR1-CLIENT1-ETHERNET")
 286         self.assertEqual(response['status_code'], requests.codes.conflict)
 287
 288     def test_22_service_path_delete_xpdr_check(self):
 289         response = test_utils.check_node_attribute_request("XPDRA01", "circuit-packs", "1%2F0%2F1-PLUG-NET")
 290         self.assertEqual(response['status_code'], requests.codes.ok)
 291         self.assertEqual('not-reserved-available', response["circuit-packs"][0]['equipment-state'])
 292
 293     def test_23_rdm_device_disconnected(self):
 294         response = test_utils.unmount_device("ROADMA01")
 295         self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
 296
 297     def test_24_xpdr_device_disconnected(self):
 298         response = test_utils.unmount_device("XPDRA01")
 299         self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
 300
 301
 302 if __name__ == "__main__":
 303     unittest.main(verbosity=2, failfast=False)