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_rfc8040  # 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_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 test_01_rdm_device_connected(self):
  44         response = test_utils_rfc8040.mount_device("ROADM-A1", ('roadma', self.NODE_VERSION))
  45         self.assertEqual(response.status_code, requests.codes.created, test_utils_rfc8040.CODE_SHOULD_BE_201)
  46
  47     def test_02_xpdr_device_connected(self):
  48         response = test_utils_rfc8040.mount_device("XPDR-A1", ('xpdra', self.NODE_VERSION))
  49         self.assertEqual(response.status_code, requests.codes.created, test_utils_rfc8040.CODE_SHOULD_BE_201)
  50
  51     def test_03_rdm_portmapping(self):
  52         response = test_utils_rfc8040.get_portmapping("ROADM-A1")
  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_rfc8040.get_portmapping("XPDR-A1")
  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', 'xponder-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', 'xponder-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_rfc8040.device_renderer_service_path_request(
  88             {
  89                 'service-name': 'service_test',
  90                 'wave-number': '7',
  91                 'modulation-format': 'dp-qpsk',
  92                 'operation': 'create',
  93                 'nodes': [{'node-id': 'ROADM-A1', 'src-tp': 'SRG1-PP3-TXRX', 'dest-tp': 'DEG1-TTP-TXRX'},
  94                           {'node-id': 'XPDR-A1', 'src-tp': 'XPDR1-CLIENT1', 'dest-tp': 'XPDR1-NETWORK1'}],
  95                 'center-freq': 195.8,
  96                 'nmc-width': 40,
  97                 'min-freq': 195.775,
  98                 'max-freq': 195.825,
  99                 'lower-spectral-slot-number': 713,
 100                 'higher-spectral-slot-number': 720
 101             })
 102         self.assertEqual(response['status_code'], requests.codes.ok)
 103         self.assertIn('Interfaces created successfully for nodes: ROADM-A1', response['output']['result'])
 104
 105     def test_06_service_path_create_rdm_check(self):
 106         response = test_utils_rfc8040.check_node_attribute_request("ROADM-A1", "interface", "DEG1-TTP-TXRX-nmc-713:720")
 107         self.assertEqual(response['status_code'], requests.codes.ok)
 108         # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
 109         self.assertDictEqual(
 110             dict({
 111                  'name': 'DEG1-TTP-TXRX-nmc-713:720',
 112                  'administrative-state': 'inService',
 113                  'supporting-circuit-pack-name': '1/0',
 114                  'type': 'org-openroadm-interfaces:networkMediaChannelConnectionTerminationPoint',
 115                  'supporting-port': 'L1'
 116                  }, **response['interface'][0]), response['interface'][0]
 117         )
 118         self.assertIn(response['interface'][0]['org-openroadm-network-media-channel-interfaces:nmc-ctp'],
 119                       [{'frequency': '195.8000', 'width': '40'},
 120                        {'frequency': 195.8, 'width': 40}])
 121
 122     def test_07_service_path_create_rdm_check(self):
 123         response = test_utils_rfc8040.check_node_attribute_request("ROADM-A1", "interface", "DEG1-TTP-TXRX-mc-713:720")
 124         self.assertEqual(response['status_code'], requests.codes.ok)
 125         # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
 126         self.assertDictEqual(
 127             dict({
 128                  'name': 'DEG1-TTP-TXRX-mc-7',
 129                  'administrative-state': 'inService',
 130                  'supporting-circuit-pack-name': '1/0',
 131                  'type': 'org-openroadm-interfaces:mediaChannelTrailTerminationPoint',
 132                  'supporting-port': 'L1'
 133                  }, **response['interface'][0]), response['interface'][0]
 134         )
 135         self.assertIn(response['interface'][0]['org-openroadm-media-channel-interfaces:mc-ttp'],
 136                       [{'min-freq': '195.7750', 'max-freq': '195.8250'},
 137                        {'min-freq': 195.775, 'max-freq': 195.825}])
 138
 139     def test_08_service_path_create_rdm_check(self):
 140         response = test_utils_rfc8040.check_node_attribute_request("ROADM-A1", "interface", "SRG1-PP3-TXRX-nmc-713:720")
 141         self.assertEqual(response['status_code'], requests.codes.ok)
 142         # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
 143         self.assertDictEqual(
 144             dict({
 145                  'name': 'SRG1-PP3-TXRX-nmc-713:720',
 146                  'administrative-state': 'inService',
 147                  'supporting-circuit-pack-name': '3/0',
 148                  'type': 'org-openroadm-interfaces:networkMediaChannelConnectionTerminationPoint',
 149                  'supporting-port': 'C3'
 150                  }, **response['interface'][0]), response['interface'][0]
 151         )
 152         self.assertIn(response['interface'][0]['org-openroadm-network-media-channel-interfaces:nmc-ctp'],
 153                       [{'frequency': '195.8000', 'width': '40'},
 154                        {'frequency': 195.8, 'width': 40}])
 155
 156     # -mc supporting interfaces must not be created for SRG, only degrees
 157     def test_09_service_path_create_rdm_check(self):
 158         response = test_utils_rfc8040.check_node_attribute_request("ROADM-A1", "interface", "SRG1-PP3-TXRX-mc-713:720")
 159         self.assertEqual(response['status_code'], requests.codes.conflict)
 160
 161     def test_10_service_path_create_rdm_check(self):
 162         response = test_utils_rfc8040.check_node_attribute_request(
 163             "ROADM-A1", "roadm-connections", "SRG1-PP3-TXRX-DEG1-TTP-TXRX-713:720")
 164         self.assertEqual(response['status_code'], requests.codes.ok)
 165         # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
 166         self.assertDictEqual(
 167             dict({
 168                  'connection-name': 'SRG1-PP3-TXRX-DEG1-TTP-TXRX-713:720',
 169                  'opticalControlMode': 'off'
 170                  }, **response['roadm-connections'][0]), response['roadm-connections'][0])
 171         self.assertDictEqual({'src-if': 'SRG1-PP3-TXRX-nmc-713:720'}, response['roadm-connections'][0]['source'])
 172         self.assertDictEqual({'dst-if': 'DEG1-TTP-TXRX-nmc-713:720'}, response['roadm-connections'][0]['destination'])
 173
 174     def test_11_service_path_create_xpdr_check(self):
 175         response = test_utils_rfc8040.check_node_attribute_request("XPDR-A1", "interface", "XPDR1-NETWORK1-713:720")
 176         self.assertEqual(response['status_code'], requests.codes.ok)
 177         # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
 178         self.assertDictEqual(
 179             dict({
 180                  'name': 'XPDR1-NETWORK1-713:720',
 181                  'administrative-state': 'inService',
 182                  'supporting-circuit-pack-name': '1/0/1-PLUG-NET',
 183                  'type': 'org-openroadm-interfaces:opticalChannel',
 184                  'supporting-port': '1'
 185                  }, **response['interface'][0]), response['interface'][0]
 186         )
 187         self.assertIn(
 188             response['interface'][0]['org-openroadm-optical-channel-interfaces:och'],
 189             [{'rate': 'org-openroadm-common-types:R100G', 'transmit-power': '-5',
 190               'modulation-format': 'dp-qpsk', 'frequency': '195.8000'},
 191              {'rate': 'org-openroadm-common-types:R100G', 'transmit-power': -5,
 192               'modulation-format': 'dp-qpsk', 'frequency': 195.8}])
 193
 194     def test_12_service_path_create_xpdr_check(self):
 195         response = test_utils_rfc8040.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_rfc8040.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_rfc8040.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     def test_15_service_path_create_xpdr_check(self):
 251         response = test_utils_rfc8040.check_node_attribute_request("XPDR-A1", "circuit-packs", "1%2F0%2F1-PLUG-NET")
 252         self.assertEqual(response['status_code'], requests.codes.ok)
 253         self.assertIn('not-reserved-inuse', response['circuit-packs'][0]['equipment-state'])
 254         # FIXME: https://jira.opendaylight.org/browse/TRNSPRTPCE-591
 255
 256     def test_16_service_path_create_xpdr_check(self):
 257         response = test_utils_rfc8040.check_node_attribute_request("XPDR-A1", "circuit-packs", "1%2F0%2F1-PLUG-CLIENT")
 258         self.assertEqual(response['status_code'], requests.codes.ok)
 259         self.assertIn('not-reserved-inuse', response['circuit-packs'][0]['equipment-state'])
 260         # FIXME: https://jira.opendaylight.org/browse/TRNSPRTPCE-591
 261
 262     def test_17_service_path_delete(self):
 263         response = test_utils_rfc8040.device_renderer_service_path_request(
 264             {
 265                 'service-name': 'service_test',
 266                 'wave-number': '7',
 267                 'modulation-format': 'dp-qpsk',
 268                 'operation': 'delete',
 269                 'nodes': [{'node-id': 'ROADM-A1', 'src-tp': 'SRG1-PP3-TXRX', 'dest-tp': 'DEG1-TTP-TXRX'},
 270                           {'node-id': 'XPDR-A1', 'src-tp': 'XPDR1-CLIENT1', 'dest-tp': 'XPDR1-NETWORK1'}],
 271                 'center-freq': 195.8,
 272                 'nmc-width': 40,
 273                 'min-freq': 195.775,
 274                 'max-freq': 195.825,
 275                 'lower-spectral-slot-number': 713,
 276                 'higher-spectral-slot-number': 720
 277             })
 278         self.assertEqual(response['status_code'], requests.codes.ok)
 279         self.assertDictEqual(response['output'], {'result': 'Request processed', 'success': True})
 280
 281     def test_18_service_path_delete_rdm_check(self):
 282         response = test_utils_rfc8040.check_node_attribute_request("ROADM-A1", "interface", "DEG1-TTP-TXRX-mc-713:720")
 283         self.assertEqual(response['status_code'], requests.codes.conflict)
 284
 285     def test_19_service_path_delete_rdm_check(self):
 286         response = test_utils_rfc8040.check_node_attribute_request("ROADM-A1", "interface", "DEG1-TTP-TXRX-nmc-713:720")
 287         self.assertEqual(response['status_code'], requests.codes.conflict)
 288
 289     def test_20_service_path_delete_rdm_check(self):
 290         response = test_utils_rfc8040.check_node_attribute_request("ROADM-A1", "interface", "SRG1-PP3-TXRX-mc-713:720")
 291         self.assertEqual(response['status_code'], requests.codes.conflict)
 292
 293     def test_21_service_path_delete_rdm_check(self):
 294         response = test_utils_rfc8040.check_node_attribute_request("ROADM-A1", "interface", "SRG1-PP3-TXRX-nmc-713:720")
 295         self.assertEqual(response['status_code'], requests.codes.conflict)
 296
 297     def test_22_service_path_delete_rdm_check(self):
 298         response = test_utils_rfc8040.check_node_attribute_request(
 299             "ROADM-A1", "interface", "SRG1-PP3-TXRX-DEG1-TTP-TXRX-713:720")
 300         self.assertEqual(response['status_code'], requests.codes.conflict)
 301
 302     def test_23_service_path_delete_rdm_check(self):
 303         response = test_utils_rfc8040.check_node_attribute_request(
 304             "ROADM-A1", "roadm-connections", "SRG1-PP3-TXRX-DEG1-TTP-TXRX-713:720")
 305         self.assertEqual(response['status_code'], requests.codes.conflict)
 306
 307     def test_24_service_path_delete_xpdr_check(self):
 308         response = test_utils_rfc8040.check_node_attribute_request("XPDR-A1", "interface", "XPDR1-NETWORK1-713:720")
 309         self.assertEqual(response['status_code'], requests.codes.conflict)
 310
 311     def test_25_service_path_delete_xpdr_check(self):
 312         response = test_utils_rfc8040.check_node_attribute_request("XPDR-A1", "interface", "XPDR1-NETWORK1-OTU")
 313         self.assertEqual(response['status_code'], requests.codes.conflict)
 314
 315     def test_26_service_path_delete_xpdr_check(self):
 316         response = test_utils_rfc8040.check_node_attribute_request("XPDR-A1", "interface", "XPDR1-NETWORK1-ODU")
 317         self.assertEqual(response['status_code'], requests.codes.conflict)
 318
 319     def test_27_service_path_delete_xpdr_check(self):
 320         response = test_utils_rfc8040.check_node_attribute_request("XPDR-A1", "interface", "XPDR1-CLIENT1-ETHERNET")
 321         self.assertEqual(response['status_code'], requests.codes.conflict)
 322
 323     def test_28_service_path_delete_xpdr_check(self):
 324         response = test_utils_rfc8040.check_node_attribute_request("XPDR-A1", "circuit-packs", "1%2F0%2F1-PLUG-NET")
 325         self.assertEqual(response['status_code'], requests.codes.ok)
 326         self.assertEqual('not-reserved-available', response["circuit-packs"][0]['equipment-state'])
 327
 328     def test_29_service_path_delete_xpdr_check(self):
 329         response = test_utils_rfc8040.check_node_attribute_request("XPDR-A1", "circuit-packs", "1%2F0%2F1-PLUG-CLIENT")
 330         self.assertEqual(response['status_code'], requests.codes.ok)
 331         self.assertEqual('not-reserved-available', response["circuit-packs"][0]['equipment-state'])
 332
 333     def test_30_rdm_device_disconnected(self):
 334         response = test_utils_rfc8040.unmount_device("ROADM-A1")
 335         self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
 336
 337     def test_31_xpdr_device_disconnected(self):
 338         response = test_utils_rfc8040.unmount_device("XPDR-A1")
 339         self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
 340
 341
 342 if __name__ == "__main__":
 343     unittest.main(verbosity=2, failfast=False)