tests/transportpce_tests/2.2.1/test01_portmapping.py

   1 #!/usr/bin/env python
   2
   3 ##############################################################################
   4 # Copyright (c) 2019 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 TransportPCEPortMappingTesting(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                                                ('spdra', cls.NODE_VERSION)])
  36
  37     @classmethod
  38     def tearDownClass(cls):
  39         # pylint: disable=not-an-iterable
  40         for process in cls.processes:
  41             test_utils.shutdown_process(process)
  42         print("all processes killed")
  43
  44     def setUp(self):
  45         # pylint: disable=consider-using-f-string
  46         print("execution of {}".format(self.id().split(".")[-1]))
  47         time.sleep(1)
  48
  49     def test_01_rdm_device_connection(self):
  50         response = test_utils.mount_device("ROADM-A1", ('roadma', self.NODE_VERSION))
  51         self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
  52
  53     def test_02_rdm_device_connected(self):
  54         response = test_utils.check_device_connection("ROADM-A1")
  55         self.assertEqual(response['status_code'], requests.codes.ok)
  56         self.assertEqual(response['connection-status'], 'connected')
  57
  58     def test_03_rdm_portmapping_info(self):
  59         response = test_utils.get_portmapping_node_attr("ROADM-A1", "node-info", None)
  60         self.assertEqual(response['status_code'], requests.codes.ok)
  61         self.assertEqual(
  62             {'node-type': 'rdm',
  63              'node-ip-address': '127.0.0.11',
  64              'node-clli': 'NodeA',
  65              'openroadm-version': '2.2.1',
  66              'node-vendor': 'vendorA',
  67              'node-model': 'model2'},
  68             response['node-info'])
  69
  70     def test_04_rdm_portmapping_DEG1_TTP_TXRX(self):
  71         response = test_utils.get_portmapping_node_attr("ROADM-A1", "mapping", "DEG1-TTP-TXRX")
  72         self.assertEqual(response['status_code'], requests.codes.ok)
  73         self.assertIn(
  74             {'supporting-port': 'L1', 'supporting-circuit-pack-name': '1/0',
  75              'logical-connection-point': 'DEG1-TTP-TXRX', 'port-direction': 'bidirectional',
  76              'port-admin-state': 'InService', 'port-oper-state': 'InService'},
  77             response['mapping'])
  78
  79     def test_05_rdm_portmapping_DEG2_TTP_TXRX_with_ots_oms(self):
  80         response = test_utils.get_portmapping_node_attr("ROADM-A1", "mapping", "DEG2-TTP-TXRX")
  81         self.assertEqual(response['status_code'], requests.codes.ok)
  82         self.assertIn(
  83             {'supporting-port': 'L1', 'supporting-circuit-pack-name': '2/0',
  84              'logical-connection-point': 'DEG2-TTP-TXRX',
  85              'supporting-oms': 'OMS-DEG2-TTP-TXRX', 'supporting-ots': 'OTS-DEG2-TTP-TXRX',
  86              'port-direction': 'bidirectional',
  87              'port-admin-state': 'InService', 'port-oper-state': 'InService'},
  88             response['mapping'])
  89
  90     def test_06_rdm_portmapping_SRG1_PP3_TXRX(self):
  91         response = test_utils.get_portmapping_node_attr("ROADM-A1", "mapping", "SRG1-PP3-TXRX")
  92         self.assertEqual(response['status_code'], requests.codes.ok)
  93         self.assertIn(
  94             {'supporting-port': 'C3', 'supporting-circuit-pack-name': '3/0',
  95              'logical-connection-point': 'SRG1-PP3-TXRX', 'port-direction': 'bidirectional',
  96              'port-admin-state': 'InService', 'port-oper-state': 'InService'},
  97             response['mapping'])
  98
  99     def test_07_rdm_portmapping_SRG3_PP1_TXRX(self):
 100         response = test_utils.get_portmapping_node_attr("ROADM-A1", "mapping", "SRG3-PP1-TXRX")
 101         self.assertEqual(response['status_code'], requests.codes.ok)
 102         self.assertIn(
 103             {'supporting-port': 'C1', 'supporting-circuit-pack-name': '5/0',
 104              'logical-connection-point': 'SRG3-PP1-TXRX', 'port-direction': 'bidirectional',
 105              'port-admin-state': 'InService', 'port-oper-state': 'InService'},
 106             response['mapping'])
 107
 108     def test_08_xpdr_device_connection(self):
 109         response = test_utils.mount_device("XPDR-A1", ('xpdra', self.NODE_VERSION))
 110         self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
 111
 112     def test_09_xpdr_device_connected(self):
 113         response = test_utils.check_device_connection("XPDR-A1")
 114         self.assertEqual(response['status_code'], requests.codes.ok)
 115         self.assertEqual(response['connection-status'], 'connected')
 116
 117     def test_10_xpdr_portmapping_info(self):
 118         response = test_utils.get_portmapping_node_attr("XPDR-A1", "node-info", None)
 119         self.assertEqual(response['status_code'], requests.codes.ok)
 120         self.assertEqual(
 121             {'node-type': 'xpdr',
 122              'node-ip-address': '1.2.3.4',
 123              'node-clli': 'NodeA',
 124              'openroadm-version': '2.2.1',
 125              'node-vendor': 'vendorA',
 126              'node-model': 'model2'},
 127             response['node-info'])
 128
 129     def test_11_xpdr_portmapping_NETWORK1(self):
 130         response = test_utils.get_portmapping_node_attr("XPDR-A1", "mapping", "XPDR1-NETWORK1")
 131         self.assertEqual(response['status_code'], requests.codes.ok)
 132         self.assertIn(
 133             {'supported-interface-capability': ['org-openroadm-port-types:if-OCH'],
 134              'supporting-port': '1', 'supporting-circuit-pack-name': '1/0/1-PLUG-NET',
 135              'logical-connection-point': 'XPDR1-NETWORK1', 'port-qual': 'xpdr-network',
 136              'port-direction': 'bidirectional', 'connection-map-lcp': 'XPDR1-CLIENT1',
 137              'lcp-hash-val': 'AMkDwQ7xTmRI', 'xpdr-type': 'tpdr',
 138              'port-admin-state': 'InService', 'port-oper-state': 'InService'},
 139             response['mapping'])
 140
 141     def test_12_xpdr_portmapping_XPDR2_NETWORK1(self):
 142         response = test_utils.get_portmapping_node_attr("XPDR-A1", "mapping", "XPDR1-NETWORK2")
 143         self.assertEqual(response['status_code'], requests.codes.ok)
 144         self.assertIn(
 145             {'supported-interface-capability': ['org-openroadm-port-types:if-OCH'],
 146              'supporting-port': '1', 'supporting-circuit-pack-name': '1/0/2-PLUG-NET',
 147              'logical-connection-point': 'XPDR1-NETWORK2', 'port-direction': 'bidirectional',
 148              'connection-map-lcp': 'XPDR1-CLIENT2', 'port-qual': 'xpdr-network',
 149              'lcp-hash-val': 'AMkDwQ7xTmRL', 'xpdr-type': 'tpdr',
 150              'port-admin-state': 'InService', 'port-oper-state': 'InService'},
 151             response['mapping'])
 152
 153     def test_13_xpdr_portmapping_XPDR1_CLIENT1(self):
 154         response = test_utils.get_portmapping_node_attr("XPDR-A1", "mapping", "XPDR1-CLIENT1")
 155         self.assertEqual(response['status_code'], requests.codes.ok)
 156         self.assertIn(
 157             {'supported-interface-capability': ['org-openroadm-port-types:if-100GE'],
 158              'supporting-port': 'C1',
 159              'supporting-circuit-pack-name': '1/0/1-PLUG-CLIENT',
 160              'logical-connection-point': 'XPDR1-CLIENT1', 'port-direction': 'bidirectional',
 161              'connection-map-lcp': 'XPDR1-NETWORK1', 'port-qual': 'xpdr-client',
 162              'lcp-hash-val': 'AJUUr6I5fALj', 'xpdr-type': 'tpdr',
 163              'port-admin-state': 'InService', 'port-oper-state': 'InService'},
 164             response['mapping'])
 165
 166     def test_14_xpdr_portmapping_XPDR1_CLIENT2(self):
 167         response = test_utils.get_portmapping_node_attr("XPDR-A1", "mapping", "XPDR1-CLIENT2")
 168         self.assertEqual(response['status_code'], requests.codes.ok)
 169         self.assertIn(
 170             {'supported-interface-capability': ['org-openroadm-port-types:if-100GE'],
 171              'supporting-port': 'C1',
 172              'supporting-circuit-pack-name': '1/0/2-PLUG-CLIENT',
 173              'logical-connection-point': 'XPDR1-CLIENT2', 'port-direction': 'bidirectional',
 174              'connection-map-lcp': 'XPDR1-NETWORK2', 'port-qual': 'xpdr-client',
 175              'lcp-hash-val': 'AJUUr6I5fALg', 'xpdr-type': 'tpdr',
 176              'port-admin-state': 'InService', 'port-oper-state': 'InService'},
 177             response['mapping'])
 178
 179     def test_15_spdr_device_connection(self):
 180         response = test_utils.mount_device("SPDR-SA1", ('spdra', self.NODE_VERSION))
 181         self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
 182
 183     def test_16_spdr_device_connected(self):
 184         response = test_utils.check_device_connection("SPDR-SA1")
 185         self.assertEqual(response['status_code'], requests.codes.ok)
 186         self.assertEqual(response['connection-status'], 'connected')
 187
 188     def test_17_spdr_portmapping_info(self):
 189         response = test_utils.get_portmapping_node_attr("SPDR-SA1", "node-info", None)
 190         self.assertEqual(response['status_code'], requests.codes.ok)
 191         self.assertEqual(
 192             {'node-type': 'xpdr',
 193              'node-ip-address': '1.2.3.4',
 194              'node-clli': 'NodeSA',
 195              'openroadm-version': '2.2.1',
 196              'node-vendor': 'vendorA',
 197              'node-model': 'universal-switchponder'},
 198             response['node-info'])
 199
 200     def test_18_spdr_switching_pool_1(self):
 201         response = test_utils.get_portmapping_node_attr("SPDR-SA1", "switching-pool-lcp", "1")
 202         self.assertEqual(response['status_code'], requests.codes.ok)
 203         self.assertEqual("blocking",
 204                          response['switching-pool-lcp'][0]['switching-pool-type'])
 205         self.assertEqual(4,
 206                          len(response['switching-pool-lcp'][0]['non-blocking-list']))
 207         self.assertIn(
 208             {'nbl-number': 11,
 209              'lcp-list': ['XPDR1-NETWORK1', 'XPDR1-CLIENT1'],
 210              'interconnect-bandwidth-unit': 1000000000,
 211              'interconnect-bandwidth': 0},
 212             response['switching-pool-lcp'][0]['non-blocking-list'])
 213
 214     def test_19_spdr_switching_pool_2(self):
 215         response = test_utils.get_portmapping_node_attr("SPDR-SA1", "switching-pool-lcp", "2")
 216         self.assertEqual(response['status_code'], requests.codes.ok)
 217         self.assertEqual("non-blocking",
 218                          response['switching-pool-lcp'][0]['switching-pool-type'])
 219         self.assertEqual(1,
 220                          len(response['switching-pool-lcp'][0]['non-blocking-list']))
 221         self.assertIn(
 222             {'nbl-number': 2,
 223              'lcp-list': ['XPDR2-NETWORK3', 'XPDR2-NETWORK4', 'XPDR2-CLIENT1', 'XPDR2-CLIENT3',
 224                           'XPDR2-CLIENT2', 'XPDR2-NETWORK1', 'XPDR2-CLIENT4', 'XPDR2-NETWORK2'],
 225              'interconnect-bandwidth-unit': 1000000000,
 226              'interconnect-bandwidth': 0},
 227             response['switching-pool-lcp'][0]['non-blocking-list'])
 228
 229     def test_20_spdr_switching_pool_3(self):
 230         response = test_utils.get_portmapping_node_attr("SPDR-SA1", "switching-pool-lcp", "3")
 231         self.assertEqual(response['status_code'], requests.codes.ok)
 232         self.assertEqual("blocking",
 233                          response['switching-pool-lcp'][0]['switching-pool-type'])
 234         self.assertEqual(4,
 235                          len(response['switching-pool-lcp'][0]['non-blocking-list']))
 236         self.assertIn(
 237             {'nbl-number': 83,
 238              'interconnect-bandwidth': 0,
 239              'interconnect-bandwidth-unit': 1000000000,
 240              'lcp-list': ['XPDR3-CLIENT3', 'XPDR3-NETWORK1']},
 241             response['switching-pool-lcp'][0]['non-blocking-list'])
 242
 243     def test_21_spdr_portmapping_mappings(self):
 244         response = test_utils.get_portmapping_node_attr("SPDR-SA1", None, None)
 245         self.assertEqual(response['status_code'], requests.codes.ok)
 246         self.assertEqual(18, len(response['nodes'][0]['mapping']))
 247
 248     def test_22_spdr_portmapping_XPDR1_CLIENT1(self):
 249         response = test_utils.get_portmapping_node_attr("SPDR-SA1", "mapping", "XPDR1-CLIENT1")
 250         self.assertEqual(response['status_code'], requests.codes.ok)
 251         expected_subset_response = {
 252             "logical-connection-point": "XPDR1-CLIENT1",
 253             "supporting-circuit-pack-name": "CP1-SFP1",
 254             "supporting-port": "CP1-SFP1-P1",
 255             "lcp-hash-val": "FqlcrxV7p30=",
 256             "port-direction": "bidirectional",
 257             "port-admin-state": "InService",
 258             "xpdr-type": "mpdr",
 259             "port-qual": "xpdr-client",
 260             "port-oper-state": "InService"}
 261         expected_sorted_list = ["org-openroadm-port-types:if-10GE-ODU2",
 262                                 "org-openroadm-port-types:if-10GE-ODU2e"]
 263         subset = {k: v for k, v in response['mapping'][0].items() if k in expected_subset_response}
 264         self.assertDictEqual(subset, expected_subset_response)
 265         self.assertEqual(sorted(response['mapping'][0]['supported-interface-capability']), expected_sorted_list)
 266
 267     def test_23_spdr_portmapping_XPDR1_NETWORK1(self):
 268         response = test_utils.get_portmapping_node_attr("SPDR-SA1", "mapping", "XPDR1-NETWORK1")
 269         self.assertEqual(response['status_code'], requests.codes.ok)
 270         self.assertIn(
 271             {"logical-connection-point": "XPDR1-NETWORK1",
 272              "supporting-circuit-pack-name": "CP1-CFP0",
 273              "supported-interface-capability": ["org-openroadm-port-types:if-OCH-OTU4-ODU4"],
 274              "supporting-port": "CP1-CFP0-P1",
 275              "lcp-hash-val": "Swfw02qXGyI=",
 276              "port-direction": "bidirectional",
 277              "port-admin-state": "InService",
 278              "xpdr-type": "mpdr",
 279              "port-qual": "xpdr-network",
 280              "port-oper-state": "InService"},
 281             response['mapping'])
 282
 283     def test_24_spdr_portmapping_XPDR2_CLIENT2(self):
 284         response = test_utils.get_portmapping_node_attr("SPDR-SA1", "mapping", "XPDR2-CLIENT2")
 285         self.assertEqual(response['status_code'], requests.codes.ok)
 286         expected_subset_response = {
 287             'logical-connection-point': 'XPDR2-CLIENT2',
 288             'supporting-port': 'CP2-QSFP2-P1',
 289             'lcp-hash-val': 'AN/WSSRXne3t',
 290             'port-direction': 'bidirectional',
 291             'xpdr-type': 'switch',
 292             'port-qual': 'switch-client',
 293             'supporting-circuit-pack-name': 'CP2-QSFP2',
 294             'port-admin-state': 'InService',
 295             'port-oper-state': 'InService'}
 296         expected_sorted_list = ['org-openroadm-port-types:if-100GE',
 297                                 'org-openroadm-port-types:if-100GE-ODU4']
 298         subset = {k: v for k, v in response['mapping'][0].items() if k in expected_subset_response}
 299         self.assertDictEqual(subset, expected_subset_response)
 300         self.assertEqual(sorted(response['mapping'][0]['supported-interface-capability']), expected_sorted_list)
 301
 302     def test_25_spdr_portmapping_XPDR2_NETWORK2(self):
 303         response = test_utils.get_portmapping_node_attr("SPDR-SA1", "mapping", "XPDR2-NETWORK2")
 304         self.assertEqual(response['status_code'], requests.codes.ok)
 305         self.assertIn(
 306             {"logical-connection-point": "XPDR2-NETWORK2",
 307              "supporting-circuit-pack-name": "CP6-CFP",
 308              "supported-interface-capability": ["org-openroadm-port-types:if-OCH-OTU4-ODU4"],
 309              "supporting-port": "CP6-CFP-P1",
 310              "lcp-hash-val": "exT821pFtOQ=",
 311              "port-direction": "bidirectional",
 312              "port-admin-state": "InService",
 313              "xpdr-type": "switch",
 314              "port-qual": "switch-network",
 315              "port-oper-state": "InService"},
 316             response['mapping'])
 317
 318     def test_26_spdr_portmapping_XPDR3_CLIENT3(self):
 319         response = test_utils.get_portmapping_node_attr("SPDR-SA1", "mapping", "XPDR3-CLIENT3")
 320         self.assertEqual(response['status_code'], requests.codes.ok)
 321         self.assertIn(
 322             {"logical-connection-point": "XPDR3-CLIENT3",
 323              "supporting-circuit-pack-name": "CP3-SFP3",
 324              "supported-interface-capability": ["org-openroadm-port-types:if-1GE-ODU0"],
 325              "supporting-port": "CP3-SFP3-P1",
 326              "lcp-hash-val": "AKsQ/HRQdtdN",
 327              "port-direction": "bidirectional",
 328              "port-admin-state": "InService",
 329              "xpdr-type": "mpdr",
 330              "port-qual": "xpdr-client",
 331              "port-oper-state": "InService"},
 332             response['mapping'])
 333
 334     def test_27_spdr_portmapping_XPDR3_NETWORK1(self):
 335         response = test_utils.get_portmapping_node_attr("SPDR-SA1", "mapping", "XPDR3-NETWORK1")
 336         self.assertEqual(response['status_code'], requests.codes.ok)
 337         self.assertIn(
 338             {"logical-connection-point": "XPDR3-NETWORK1",
 339              "supporting-circuit-pack-name": "CP3-CFP0",
 340              "supported-interface-capability": ["org-openroadm-port-types:if-OCH-OTU4-ODU4"],
 341              "supporting-port": "CP3-CFP0-P1",
 342              "lcp-hash-val": "ANnxoi7K8q30",
 343              "port-direction": "bidirectional",
 344              "port-admin-state": "InService",
 345              "xpdr-type": "mpdr",
 346              "port-qual": "xpdr-network",
 347              "port-oper-state": "InService"},
 348             response['mapping'])
 349
 350     def test_28_spdr_device_disconnection(self):
 351         response = test_utils.unmount_device("SPDR-SA1")
 352         self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
 353
 354     def test_29_xpdr_device_disconnected(self):
 355         response = test_utils.check_device_connection("SPDR-SA1")
 356         self.assertEqual(response['status_code'], requests.codes.conflict)
 357         self.assertIn(response['connection-status']['error-type'], ('protocol', 'application'))
 358         self.assertEqual(response['connection-status']['error-tag'], 'data-missing')
 359         self.assertEqual(response['connection-status']['error-message'],
 360                          'Request could not be completed because the relevant data model content does not exist')
 361
 362     def test_30_xpdr_device_disconnection(self):
 363         response = test_utils.unmount_device("XPDR-A1")
 364         self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
 365
 366     def test_31_xpdr_device_disconnected(self):
 367         response = test_utils.check_device_connection("XPDR-A1")
 368         self.assertEqual(response['status_code'], requests.codes.conflict)
 369         self.assertIn(response['connection-status']['error-type'], ('protocol', 'application'))
 370         self.assertEqual(response['connection-status']['error-tag'], 'data-missing')
 371         self.assertEqual(response['connection-status']['error-message'],
 372                          'Request could not be completed because the relevant data model content does not exist')
 373
 374     def test_32_xpdr_device_not_connected(self):
 375         response = test_utils.get_portmapping_node_attr("XPDR-A1", "node-info", None)
 376         self.assertEqual(response['status_code'], requests.codes.conflict)
 377         self.assertIn(response['node-info']['error-type'], ('protocol', 'application'))
 378         self.assertEqual(response['node-info']['error-tag'], 'data-missing')
 379         self.assertEqual(response['node-info']['error-message'],
 380                          'Request could not be completed because the relevant data model content does not exist')
 381
 382     def test_33_rdm_device_disconnection(self):
 383         response = test_utils.unmount_device("ROADM-A1")
 384         self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
 385
 386     def test_34_rdm_device_disconnected(self):
 387         response = test_utils.check_device_connection("ROADM-A1")
 388         self.assertEqual(response['status_code'], requests.codes.conflict)
 389         self.assertIn(response['connection-status']['error-type'], ('protocol', 'application'))
 390         self.assertEqual(response['connection-status']['error-tag'], 'data-missing')
 391         self.assertEqual(response['connection-status']['error-message'],
 392                          'Request could not be completed because the relevant data model content does not exist')
 393
 394     def test_35_rdm_device_not_connected(self):
 395         response = test_utils.get_portmapping_node_attr("ROADM-A1", "node-info", None)
 396         self.assertEqual(response['status_code'], requests.codes.conflict)
 397         self.assertIn(response['node-info']['error-type'], ('protocol', 'application'))
 398         self.assertEqual(response['node-info']['error-tag'], 'data-missing')
 399         self.assertEqual(response['node-info']['error-message'],
 400                          'Request could not be completed because the relevant data model content does not exist')
 401
 402
 403 if __name__ == "__main__":
 404     unittest.main(verbosity=2)