tests/transportpce_tests/2.2.1/test06_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 # from unittest.result import failfast
  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 = '2.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 test_01_rdm_device_connected(self):
  44         response = test_utils.mount_device("ROADM-A1", ('roadma', self.NODE_VERSION))
  45         self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
  46
  47     def test_02_xpdr_device_connected(self):
  48         response = test_utils.mount_device("XPDR-A1", ('xpdra', self.NODE_VERSION))
  49         self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
  50
  51     def test_03_rdm_portmapping(self):
  52         response = test_utils.get_portmapping_node_attr("ROADM-A1", None, None)
  53         self.assertEqual(response['status_code'], requests.codes.ok)
  54         self.assertIn(
  55             {'supporting-port': 'L1', 'supporting-circuit-pack-name': '1/0',
  56              'logical-connection-point': 'DEG1-TTP-TXRX', 'port-direction': 'bidirectional',
  57              'port-admin-state': 'InService', 'port-oper-state': 'InService'},
  58             response['nodes'][0]['mapping'])
  59         self.assertIn(
  60             {'supporting-port': 'C3', 'supporting-circuit-pack-name': '3/0',
  61              'logical-connection-point': 'SRG1-PP3-TXRX', 'port-direction': 'bidirectional',
  62              'port-admin-state': 'InService', 'port-oper-state': 'InService'},
  63             response['nodes'][0]['mapping'])
  64
  65     def test_04_xpdr_portmapping(self):
  66         response = test_utils.get_portmapping_node_attr("XPDR-A1", None, None)
  67         self.assertEqual(response['status_code'], requests.codes.ok)
  68         self.assertIn(
  69             {'supported-interface-capability': ['org-openroadm-port-types:if-OCH'],
  70              'supporting-port': '1', 'supporting-circuit-pack-name': '1/0/1-PLUG-NET',
  71              'logical-connection-point': 'XPDR1-NETWORK1', 'port-qual': 'xpdr-network',
  72              'port-direction': 'bidirectional', 'connection-map-lcp': 'XPDR1-CLIENT1',
  73              'lcp-hash-val': 'AMkDwQ7xTmRI', 'xpdr-type': 'tpdr',
  74              'port-admin-state': 'InService', 'port-oper-state': 'InService'},
  75             response['nodes'][0]['mapping'])
  76         self.assertIn(
  77             {'supported-interface-capability': ['org-openroadm-port-types:if-100GE'],
  78              'supporting-port': 'C1',
  79              'supporting-circuit-pack-name': '1/0/1-PLUG-CLIENT',
  80              'logical-connection-point': 'XPDR1-CLIENT1', 'port-direction': 'bidirectional',
  81              'connection-map-lcp': 'XPDR1-NETWORK1', 'port-qual': 'xpdr-client',
  82              'lcp-hash-val': 'AJUUr6I5fALj', 'xpdr-type': 'tpdr',
  83              'port-admin-state': 'InService', 'port-oper-state': 'InService'},
  84             response['nodes'][0]['mapping'])
  85
  86     def test_05_service_path_create(self):
  87         response = test_utils.transportpce_api_rpc_request(
  88             'transportpce-device-renderer', 'service-path',
  89             {
  90                 'service-name': 'service_test',
  91                 'wave-number': '7',
  92                 'modulation-format': 'dp-qpsk',
  93                 'operation': 'create',
  94                 'nodes': [{'node-id': 'ROADM-A1', 'src-tp': 'SRG1-PP3-TXRX', 'dest-tp': 'DEG1-TTP-TXRX'},
  95                           {'node-id': 'XPDR-A1', 'src-tp': 'XPDR1-CLIENT1', 'dest-tp': 'XPDR1-NETWORK1'}],
  96                 'center-freq': 195.8,
  97                 'nmc-width': 40,
  98                 'min-freq': 195.775,
  99                 'max-freq': 195.825,
 100                 'lower-spectral-slot-number': 713,
 101                 'higher-spectral-slot-number': 720
 102             })
 103         self.assertEqual(response['status_code'], requests.codes.ok)
 104         self.assertIn('Interfaces created successfully for nodes: ROADM-A1', response['output']['result'])
 105
 106     def test_06_service_path_create_rdm_check(self):
 107         response = test_utils.check_node_attribute_request("ROADM-A1", "interface", "DEG1-TTP-TXRX-nmc-713:720")
 108         self.assertEqual(response['status_code'], requests.codes.ok)
 109         # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
 110         self.assertDictEqual(
 111             dict({
 112                  'name': 'DEG1-TTP-TXRX-nmc-713:720',
 113                  'administrative-state': 'inService',
 114                  'supporting-circuit-pack-name': '1/0',
 115                  'type': 'org-openroadm-interfaces:networkMediaChannelConnectionTerminationPoint',
 116                  'supporting-port': 'L1'
 117                  }, **response['interface'][0]), response['interface'][0]
 118         )
 119         nmcctp = response['interface'][0]['org-openroadm-network-media-channel-interfaces:nmc-ctp']
 120         self.assertEqual(float(nmcctp['frequency']), 195.8)
 121         self.assertEqual(float(nmcctp['width']), 40)
 122
 123     def test_07_service_path_create_rdm_check(self):
 124         response = test_utils.check_node_attribute_request("ROADM-A1", "interface", "DEG1-TTP-TXRX-mc-713:720")
 125         self.assertEqual(response['status_code'], requests.codes.ok)
 126         # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
 127         self.assertDictEqual(
 128             dict({
 129                  'name': 'DEG1-TTP-TXRX-mc-7',
 130                  'administrative-state': 'inService',
 131                  'supporting-circuit-pack-name': '1/0',
 132                  'type': 'org-openroadm-interfaces:mediaChannelTrailTerminationPoint',
 133                  'supporting-port': 'L1'
 134                  }, **response['interface'][0]), response['interface'][0]
 135         )
 136         mcttp = response['interface'][0]['org-openroadm-media-channel-interfaces:mc-ttp']
 137         self.assertEqual(float(mcttp['min-freq']), 195.775)
 138         self.assertEqual(float(mcttp['max-freq']), 195.825)
 139
 140     def test_08_service_path_create_rdm_check(self):
 141         response = test_utils.check_node_attribute_request("ROADM-A1", "interface", "SRG1-PP3-TXRX-nmc-713:720")
 142         self.assertEqual(response['status_code'], requests.codes.ok)
 143         # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
 144         self.assertDictEqual(
 145             dict({
 146                  'name': 'SRG1-PP3-TXRX-nmc-713:720',
 147                  'administrative-state': 'inService',
 148                  'supporting-circuit-pack-name': '3/0',
 149                  'type': 'org-openroadm-interfaces:networkMediaChannelConnectionTerminationPoint',
 150                  'supporting-port': 'C3'
 151                  }, **response['interface'][0]), response['interface'][0]
 152         )
 153         nmcctp = response['interface'][0]['org-openroadm-network-media-channel-interfaces:nmc-ctp']
 154         self.assertEqual(float(nmcctp['frequency']), 195.8)
 155         self.assertEqual(float(nmcctp['width']), 40)
 156
 157     # -mc supporting interfaces must not be created for SRG, only degrees
 158     def test_09_service_path_create_rdm_check(self):
 159         response = test_utils.check_node_attribute_request("ROADM-A1", "interface", "SRG1-PP3-TXRX-mc-713:720")
 160         self.assertEqual(response['status_code'], requests.codes.conflict)
 161
 162     def test_10_service_path_create_rdm_check(self):
 163         response = test_utils.check_node_attribute_request(
 164             "ROADM-A1", "roadm-connections", "SRG1-PP3-TXRX-DEG1-TTP-TXRX-713:720")
 165         self.assertEqual(response['status_code'], requests.codes.ok)
 166         # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
 167         self.assertDictEqual(
 168             dict({
 169                  'connection-name': 'SRG1-PP3-TXRX-DEG1-TTP-TXRX-713:720',
 170                  'opticalControlMode': 'off'
 171                  }, **response['roadm-connections'][0]), response['roadm-connections'][0])
 172         self.assertDictEqual({'src-if': 'SRG1-PP3-TXRX-nmc-713:720'}, response['roadm-connections'][0]['source'])
 173         self.assertDictEqual({'dst-if': 'DEG1-TTP-TXRX-nmc-713:720'}, response['roadm-connections'][0]['destination'])
 174
 175     def test_11_service_path_create_xpdr_check(self):
 176         response = test_utils.check_node_attribute_request("XPDR-A1", "interface", "XPDR1-NETWORK1-713:720")
 177         self.assertEqual(response['status_code'], requests.codes.ok)
 178         # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
 179         self.assertDictEqual(
 180             dict({
 181                  'name': 'XPDR1-NETWORK1-713:720',
 182                  'administrative-state': 'inService',
 183                  'supporting-circuit-pack-name': '1/0/1-PLUG-NET',
 184                  'type': 'org-openroadm-interfaces:opticalChannel',
 185                  'supporting-port': '1'
 186                  }, **response['interface'][0]), response['interface'][0]
 187         )
 188         intf = response['interface'][0]['org-openroadm-optical-channel-interfaces:och']
 189         self.assertEqual(intf['rate'], 'org-openroadm-common-types:R100G')
 190         self.assertEqual(intf['modulation-format'], 'dp-qpsk')
 191         self.assertEqual(float(intf['frequency']), 195.8)
 192         self.assertEqual(float(intf['transmit-power']), -5)
 193
 194     def test_12_service_path_create_xpdr_check(self):
 195         response = test_utils.check_node_attribute_request("XPDR-A1", "interface", "XPDR1-NETWORK1-OTU")
 196         self.assertEqual(response['status_code'], requests.codes.ok)
 197         # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
 198         self.assertDictEqual(
 199             dict({
 200                  'name': 'XPDR1-NETWORK1-OTU',
 201                  'administrative-state': 'inService',
 202                  'supporting-circuit-pack-name': '1/0/1-PLUG-NET',
 203                  'type': 'org-openroadm-interfaces:otnOtu',
 204                  'supporting-port': '1',
 205                  'supporting-interface': 'XPDR1-NETWORK1-7'
 206                  }, **response['interface'][0]), response['interface'][0])
 207         self.assertDictEqual(
 208             dict({'rate': 'org-openroadm-otn-common-types:OTU4', 'fec': 'scfec'},
 209                  **response['interface'][0]['org-openroadm-otn-otu-interfaces:otu']),
 210             response['interface'][0]['org-openroadm-otn-otu-interfaces:otu'])
 211
 212     def test_13_service_path_create_xpdr_check(self):
 213         response = test_utils.check_node_attribute_request("XPDR-A1", "interface", "XPDR1-NETWORK1-ODU4")
 214         self.assertEqual(response['status_code'], requests.codes.ok)
 215         # the 2 following statements replace self.assertDictContainsSubset deprecated in python 3.2
 216         self.assertDictEqual(
 217             dict({
 218                  'name': 'XPDR1-NETWORK1-ODU4',
 219                  'administrative-state': 'inService',
 220                  'supporting-circuit-pack-name': '1/0/1-PLUG-NET',
 221                  'type': 'org-openroadm-interfaces:otnOdu',
 222                  'supporting-port': '1',
 223                  'supporting-interface': 'XPDR1-NETWORK1-OTU'
 224                  }, **response['interface'][0]), response['interface'][0])
 225         self.assertDictEqual(
 226             dict({'rate': 'org-openroadm-otn-common-types:ODU4', 'monitoring-mode': 'terminated'},
 227                  **response['interface'][0]['org-openroadm-otn-odu-interfaces:odu']),
 228             response['interface'][0]['org-openroadm-otn-odu-interfaces:odu'])
 229         self.assertDictEqual({'exp-payload-type': '07', 'payload-type': '07'},
 230                              response['interface'][0]['org-openroadm-otn-odu-interfaces:odu']['opu'])
 231
 232     def test_14_service_path_create_xpdr_check(self):
 233         response = test_utils.check_node_attribute_request("XPDR-A1", "interface", "XPDR1-CLIENT1-ETHERNET")
 234         self.assertEqual(response['status_code'], requests.codes.ok)
 235         # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
 236         self.assertDictEqual(
 237             dict({
 238                  'name': 'XPDR1-CLIENT1-ETHERNET',
 239                  'administrative-state': 'inService',
 240                  'supporting-circuit-pack-name': '1/0/1-PLUG-CLIENT',
 241                  'type': 'org-openroadm-interfaces:ethernetCsmacd',
 242                  'supporting-port': 'C1'
 243                  }, **response['interface'][0]), response['interface'][0]
 244         )
 245         self.assertDictEqual(
 246             dict({'fec': 'off', 'speed': 100000},
 247                  **response['interface'][0]['org-openroadm-ethernet-interfaces:ethernet']),
 248             response['interface'][0]['org-openroadm-ethernet-interfaces:ethernet'])
 249
 250     @unittest.skip("attributes that contains slashes seems to give some troubles to the controller")
 251     def test_15_service_path_create_xpdr_check(self):
 252         response = test_utils.check_node_attribute_request("XPDR-A1", "circuit-packs", "1%2F0%2F1-PLUG-NET")
 253         self.assertEqual(response['status_code'], requests.codes.ok)
 254         self.assertIn('not-reserved-inuse', response['circuit-packs'][0]['equipment-state'])
 255         # FIXME: https://jira.opendaylight.org/browse/TRNSPRTPCE-591
 256
 257     @unittest.skip("attributes that contains slashes seems to give some troubles to the controller")
 258     def test_16_service_path_create_xpdr_check(self):
 259         response = test_utils.check_node_attribute_request("XPDR-A1", "circuit-packs", "1%2F0%2F1-PLUG-CLIENT")
 260         self.assertEqual(response['status_code'], requests.codes.ok)
 261         self.assertIn('not-reserved-inuse', response['circuit-packs'][0]['equipment-state'])
 262         # FIXME: https://jira.opendaylight.org/browse/TRNSPRTPCE-591
 263
 264     def test_17_service_path_delete(self):
 265         response = test_utils.transportpce_api_rpc_request(
 266             'transportpce-device-renderer', 'service-path',
 267             {
 268                 'service-name': 'service_test',
 269                 'wave-number': '7',
 270                 'modulation-format': 'dp-qpsk',
 271                 'operation': 'delete',
 272                 'nodes': [{'node-id': 'ROADM-A1', 'src-tp': 'SRG1-PP3-TXRX', 'dest-tp': 'DEG1-TTP-TXRX'},
 273                           {'node-id': 'XPDR-A1', 'src-tp': 'XPDR1-CLIENT1', 'dest-tp': 'XPDR1-NETWORK1'}],
 274                 'center-freq': 195.8,
 275                 'nmc-width': 40,
 276                 'min-freq': 195.775,
 277                 'max-freq': 195.825,
 278                 'lower-spectral-slot-number': 713,
 279                 'higher-spectral-slot-number': 720
 280             })
 281         self.assertEqual(response['status_code'], requests.codes.ok)
 282         self.assertDictEqual(response['output'], {'result': 'Request processed', 'success': True})
 283
 284     def test_18_service_path_delete_rdm_check(self):
 285         response = test_utils.check_node_attribute_request("ROADM-A1", "interface", "DEG1-TTP-TXRX-mc-713:720")
 286         self.assertEqual(response['status_code'], requests.codes.conflict)
 287
 288     def test_19_service_path_delete_rdm_check(self):
 289         response = test_utils.check_node_attribute_request("ROADM-A1", "interface", "DEG1-TTP-TXRX-nmc-713:720")
 290         self.assertEqual(response['status_code'], requests.codes.conflict)
 291
 292     def test_20_service_path_delete_rdm_check(self):
 293         response = test_utils.check_node_attribute_request("ROADM-A1", "interface", "SRG1-PP3-TXRX-mc-713:720")
 294         self.assertEqual(response['status_code'], requests.codes.conflict)
 295
 296     def test_21_service_path_delete_rdm_check(self):
 297         response = test_utils.check_node_attribute_request("ROADM-A1", "interface", "SRG1-PP3-TXRX-nmc-713:720")
 298         self.assertEqual(response['status_code'], requests.codes.conflict)
 299
 300     def test_22_service_path_delete_rdm_check(self):
 301         response = test_utils.check_node_attribute_request(
 302             "ROADM-A1", "interface", "SRG1-PP3-TXRX-DEG1-TTP-TXRX-713:720")
 303         self.assertEqual(response['status_code'], requests.codes.conflict)
 304
 305     def test_23_service_path_delete_rdm_check(self):
 306         response = test_utils.check_node_attribute_request(
 307             "ROADM-A1", "roadm-connections", "SRG1-PP3-TXRX-DEG1-TTP-TXRX-713:720")
 308         self.assertEqual(response['status_code'], requests.codes.conflict)
 309
 310     def test_24_service_path_delete_xpdr_check(self):
 311         response = test_utils.check_node_attribute_request("XPDR-A1", "interface", "XPDR1-NETWORK1-713:720")
 312         self.assertEqual(response['status_code'], requests.codes.conflict)
 313
 314     def test_25_service_path_delete_xpdr_check(self):
 315         response = test_utils.check_node_attribute_request("XPDR-A1", "interface", "XPDR1-NETWORK1-OTU")
 316         self.assertEqual(response['status_code'], requests.codes.conflict)
 317
 318     def test_26_service_path_delete_xpdr_check(self):
 319         response = test_utils.check_node_attribute_request("XPDR-A1", "interface", "XPDR1-NETWORK1-ODU")
 320         self.assertEqual(response['status_code'], requests.codes.conflict)
 321
 322     def test_27_service_path_delete_xpdr_check(self):
 323         response = test_utils.check_node_attribute_request("XPDR-A1", "interface", "XPDR1-CLIENT1-ETHERNET")
 324         self.assertEqual(response['status_code'], requests.codes.conflict)
 325
 326     @unittest.skip("attributes that contains slashes seems to give some troubles to the controller")
 327     def test_28_service_path_delete_xpdr_check(self):
 328         response = test_utils.check_node_attribute_request("XPDR-A1", "circuit-packs", "1%2F0%2F1-PLUG-NET")
 329         self.assertEqual(response['status_code'], requests.codes.ok)
 330         self.assertEqual('not-reserved-available', response["circuit-packs"][0]['equipment-state'])
 331
 332     @unittest.skip("attributes that contains slashes seems to give some troubles to the controller")
 333     def test_29_service_path_delete_xpdr_check(self):
 334         response = test_utils.check_node_attribute_request("XPDR-A1", "circuit-packs", "1%2F0%2F1-PLUG-CLIENT")
 335         self.assertEqual(response['status_code'], requests.codes.ok)
 336         self.assertEqual('not-reserved-available', response["circuit-packs"][0]['equipment-state'])
 337
 338     def test_30_rdm_device_disconnected(self):
 339         response = test_utils.unmount_device("ROADM-A1")
 340         self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
 341
 342     def test_31_xpdr_device_disconnected(self):
 343         response = test_utils.unmount_device("XPDR-A1")
 344         self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
 345
 346
 347 if __name__ == "__main__":
 348     unittest.main(verbosity=2, failfast=False)