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_rfc8040  # 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_rfc8040.start_tpce()
  34         cls.processes = test_utils_rfc8040.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_rfc8040.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_rfc8040.mount_device("ROADM-A1", ('roadma', self.NODE_VERSION))
  51         self.assertEqual(response.status_code, requests.codes.created, test_utils_rfc8040.CODE_SHOULD_BE_201)
  52
  53     def test_02_rdm_device_connected(self):
  54         response = test_utils_rfc8040.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_rfc8040.get_portmapping_node_info("ROADM-A1")
  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_rfc8040.portmapping_request("ROADM-A1", "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_rfc8040.portmapping_request("ROADM-A1", "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_rfc8040.portmapping_request("ROADM-A1", "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_rfc8040.portmapping_request("ROADM-A1", "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_rfc8040.mount_device("XPDR-A1", ('xpdra', self.NODE_VERSION))
 110         self.assertEqual(response.status_code, requests.codes.created, test_utils_rfc8040.CODE_SHOULD_BE_201)
 111
 112     def test_09_xpdr_device_connected(self):
 113         response = test_utils_rfc8040.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_rfc8040.get_portmapping_node_info("XPDR-A1")
 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_rfc8040.portmapping_request("XPDR-A1", "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', 'xponder-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_rfc8040.portmapping_request("XPDR-A1", "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', 'xponder-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_rfc8040.portmapping_request("XPDR-A1", "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', 'xponder-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_rfc8040.portmapping_request("XPDR-A1", "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', 'xponder-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_rfc8040.mount_device("SPDR-SA1", ('spdra', self.NODE_VERSION))
 181         self.assertEqual(response.status_code, requests.codes.created, test_utils_rfc8040.CODE_SHOULD_BE_201)
 182
 183     def test_16_spdr_device_connected(self):
 184         response = test_utils_rfc8040.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_rfc8040.get_portmapping_node_info("SPDR-SA1")
 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_rfc8040.portmapping_switching_pool_request("SPDR-SA1", "1")
 202         self.assertEqual(response['status_code'], requests.codes.ok)
 203         self.assertEqual("blocking",
 204                          response['switching_pool'][0]['switching-pool-type'])
 205         self.assertEqual(4,
 206                          len(response['switching_pool'][0]['non-blocking-list']))
 207         self.assertIn(
 208             {'nbl-number': 11,
 209              'lcp-list': ['XPDR1-CLIENT1', 'XPDR1-NETWORK1'],
 210              'interconnect-bandwidth-unit': 1000000000,
 211              'interconnect-bandwidth': 0},
 212             response['switching_pool'][0]['non-blocking-list'])
 213
 214     def test_19_spdr_switching_pool_2(self):
 215         response = test_utils_rfc8040.portmapping_switching_pool_request("SPDR-SA1", "2")
 216         self.assertEqual(response['status_code'], requests.codes.ok)
 217         self.assertEqual("non-blocking",
 218                          response['switching_pool'][0]['switching-pool-type'])
 219         self.assertEqual(1,
 220                          len(response['switching_pool'][0]['non-blocking-list']))
 221         self.assertIn(
 222             {'nbl-number': 2,
 223              'lcp-list': ['XPDR2-NETWORK2', 'XPDR2-NETWORK3', 'XPDR2-CLIENT1',
 224                           'XPDR2-CLIENT2', 'XPDR2-NETWORK1', 'XPDR2-CLIENT3',
 225                           'XPDR2-CLIENT4', 'XPDR2-NETWORK4'],
 226              'interconnect-bandwidth-unit': 1000000000,
 227              'interconnect-bandwidth': 0},
 228             response['switching_pool'][0]['non-blocking-list'])
 229
 230     def test_20_spdr_switching_pool_3(self):
 231         response = test_utils_rfc8040.portmapping_switching_pool_request("SPDR-SA1", "3")
 232         self.assertEqual(response['status_code'], requests.codes.ok)
 233         self.assertEqual("blocking",
 234                          response['switching_pool'][0]['switching-pool-type'])
 235         self.assertEqual(4,
 236                          len(response['switching_pool'][0]['non-blocking-list']))
 237         self.assertIn(
 238             {'nbl-number': 83,
 239              'interconnect-bandwidth': 0,
 240              'interconnect-bandwidth-unit': 1000000000,
 241              'lcp-list': ['XPDR3-NETWORK1', 'XPDR3-CLIENT3']},
 242             response['switching_pool'][0]['non-blocking-list'])
 243
 244     def test_21_spdr_portmapping_mappings(self):
 245         response = test_utils_rfc8040.get_portmapping("SPDR-SA1")
 246         self.assertEqual(response['status_code'], requests.codes.ok)
 247         self.assertEqual(18, len(response['nodes'][0]['mapping']))
 248
 249     def test_22_spdr_portmapping_XPDR1_CLIENT1(self):
 250         response = test_utils_rfc8040.portmapping_request("SPDR-SA1", "XPDR1-CLIENT1")
 251         self.assertEqual(response['status_code'], requests.codes.ok)
 252         expected_subset_response = {
 253             "logical-connection-point": "XPDR1-CLIENT1",
 254             "supporting-circuit-pack-name": "CP1-SFP1",
 255             "supporting-port": "CP1-SFP1-P1",
 256             "lcp-hash-val": "FqlcrxV7p30=",
 257             "port-direction": "bidirectional",
 258             "port-admin-state": "InService",
 259             "xponder-type": "mpdr",
 260             "port-qual": "xpdr-client",
 261             "port-oper-state": "InService"}
 262         expected_sorted_list = ["org-openroadm-port-types:if-10GE-ODU2",
 263                                 "org-openroadm-port-types:if-10GE-ODU2e"]
 264         subset = {k: v for k, v in response['mapping'][0].items() if k in expected_subset_response}
 265         self.assertDictEqual(subset, expected_subset_response)
 266         self.assertEqual(sorted(response['mapping'][0]['supported-interface-capability']), expected_sorted_list)
 267
 268     def test_23_spdr_portmapping_XPDR1_NETWORK1(self):
 269         response = test_utils_rfc8040.portmapping_request("SPDR-SA1", "XPDR1-NETWORK1")
 270         self.assertEqual(response['status_code'], requests.codes.ok)
 271         self.assertIn(
 272             {"logical-connection-point": "XPDR1-NETWORK1",
 273              "supporting-circuit-pack-name": "CP1-CFP0",
 274              "supported-interface-capability": ["org-openroadm-port-types:if-OCH-OTU4-ODU4"],
 275              "supporting-port": "CP1-CFP0-P1",
 276              "lcp-hash-val": "Swfw02qXGyI=",
 277              "port-direction": "bidirectional",
 278              "port-admin-state": "InService",
 279              "xponder-type": "mpdr",
 280              "port-qual": "xpdr-network",
 281              "port-oper-state": "InService"},
 282             response['mapping'])
 283
 284     def test_24_spdr_portmapping_XPDR2_CLIENT2(self):
 285         response = test_utils_rfc8040.portmapping_request("SPDR-SA1", "XPDR2-CLIENT2")
 286         self.assertEqual(response['status_code'], requests.codes.ok)
 287         expected_subset_response = {
 288             'logical-connection-point': 'XPDR2-CLIENT2',
 289             'supporting-port': 'CP2-QSFP2-P1',
 290             'lcp-hash-val': 'AN/WSSRXne3t',
 291             'port-direction': 'bidirectional',
 292             'xponder-type': 'switch',
 293             'port-qual': 'switch-client',
 294             'supporting-circuit-pack-name': 'CP2-QSFP2',
 295             'port-admin-state': 'InService',
 296             'port-oper-state': 'InService'}
 297         expected_sorted_list = ['org-openroadm-port-types:if-100GE',
 298                                 'org-openroadm-port-types:if-100GE-ODU4']
 299         subset = {k: v for k, v in response['mapping'][0].items() if k in expected_subset_response}
 300         self.assertDictEqual(subset, expected_subset_response)
 301         self.assertEqual(sorted(response['mapping'][0]['supported-interface-capability']), expected_sorted_list)
 302
 303     def test_25_spdr_portmapping_XPDR2_NETWORK2(self):
 304         response = test_utils_rfc8040.portmapping_request("SPDR-SA1", "XPDR2-NETWORK2")
 305         self.assertEqual(response['status_code'], requests.codes.ok)
 306         self.assertIn(
 307             {"logical-connection-point": "XPDR2-NETWORK2",
 308              "supporting-circuit-pack-name": "CP6-CFP",
 309              "supported-interface-capability": ["org-openroadm-port-types:if-OCH-OTU4-ODU4"],
 310              "supporting-port": "CP6-CFP-P1",
 311              "lcp-hash-val": "exT821pFtOQ=",
 312              "port-direction": "bidirectional",
 313              "port-admin-state": "InService",
 314              "xponder-type": "switch",
 315              "port-qual": "switch-network",
 316              "port-oper-state": "InService"},
 317             response['mapping'])
 318
 319     def test_26_spdr_portmapping_XPDR3_CLIENT3(self):
 320         response = test_utils_rfc8040.portmapping_request("SPDR-SA1", "XPDR3-CLIENT3")
 321         self.assertEqual(response['status_code'], requests.codes.ok)
 322         self.assertIn(
 323             {"logical-connection-point": "XPDR3-CLIENT3",
 324              "supporting-circuit-pack-name": "CP3-SFP3",
 325              "supported-interface-capability": ["org-openroadm-port-types:if-1GE-ODU0"],
 326              "supporting-port": "CP3-SFP3-P1",
 327              "lcp-hash-val": "AKsQ/HRQdtdN",
 328              "port-direction": "bidirectional",
 329              "port-admin-state": "InService",
 330              "xponder-type": "mpdr",
 331              "port-qual": "xpdr-client",
 332              "port-oper-state": "InService"},
 333             response['mapping'])
 334
 335     def test_27_spdr_portmapping_XPDR3_NETWORK1(self):
 336         response = test_utils_rfc8040.portmapping_request("SPDR-SA1", "XPDR3-NETWORK1")
 337         self.assertEqual(response['status_code'], requests.codes.ok)
 338         self.assertIn(
 339             {"logical-connection-point": "XPDR3-NETWORK1",
 340              "supporting-circuit-pack-name": "CP3-CFP0",
 341              "supported-interface-capability": ["org-openroadm-port-types:if-OCH-OTU4-ODU4"],
 342              "supporting-port": "CP3-CFP0-P1",
 343              "lcp-hash-val": "ANnxoi7K8q30",
 344              "port-direction": "bidirectional",
 345              "port-admin-state": "InService",
 346              "xponder-type": "mpdr",
 347              "port-qual": "xpdr-network",
 348              "port-oper-state": "InService"},
 349             response['mapping'])
 350
 351     def test_28_spdr_device_disconnection(self):
 352         response = test_utils_rfc8040.unmount_device("SPDR-SA1")
 353         self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
 354
 355     def test_29_xpdr_device_disconnected(self):
 356         response = test_utils_rfc8040.check_device_connection("SPDR-SA1")
 357         self.assertEqual(response['status_code'], requests.codes.conflict)
 358         self.assertIn(response['connection-status']['error-type'], ('protocol', 'application'))
 359         self.assertEqual(response['connection-status']['error-tag'], 'data-missing')
 360         self.assertEqual(response['connection-status']['error-message'],
 361                          'Request could not be completed because the relevant data model content does not exist')
 362
 363     def test_30_xpdr_device_disconnection(self):
 364         response = test_utils_rfc8040.unmount_device("XPDR-A1")
 365         self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
 366
 367     def test_31_xpdr_device_disconnected(self):
 368         response = test_utils_rfc8040.check_device_connection("XPDR-A1")
 369         self.assertEqual(response['status_code'], requests.codes.conflict)
 370         self.assertIn(response['connection-status']['error-type'], ('protocol', 'application'))
 371         self.assertEqual(response['connection-status']['error-tag'], 'data-missing')
 372         self.assertEqual(response['connection-status']['error-message'],
 373                          'Request could not be completed because the relevant data model content does not exist')
 374
 375     def test_32_xpdr_device_not_connected(self):
 376         response = test_utils_rfc8040.get_portmapping_node_info("XPDR-A1")
 377         self.assertEqual(response['status_code'], requests.codes.conflict)
 378         self.assertIn(response['node-info']['error-type'], ('protocol', 'application'))
 379         self.assertEqual(response['node-info']['error-tag'], 'data-missing')
 380         self.assertEqual(response['node-info']['error-message'],
 381                          'Request could not be completed because the relevant data model content does not exist')
 382
 383     def test_33_rdm_device_disconnection(self):
 384         response = test_utils_rfc8040.unmount_device("ROADM-A1")
 385         self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
 386
 387     def test_34_rdm_device_disconnected(self):
 388         response = test_utils_rfc8040.check_device_connection("ROADM-A1")
 389         self.assertEqual(response['status_code'], requests.codes.conflict)
 390         self.assertIn(response['connection-status']['error-type'], ('protocol', 'application'))
 391         self.assertEqual(response['connection-status']['error-tag'], 'data-missing')
 392         self.assertEqual(response['connection-status']['error-message'],
 393                          'Request could not be completed because the relevant data model content does not exist')
 394
 395     def test_35_rdm_device_not_connected(self):
 396         response = test_utils_rfc8040.get_portmapping_node_info("ROADM-A1")
 397         self.assertEqual(response['status_code'], requests.codes.conflict)
 398         self.assertIn(response['node-info']['error-type'], ('protocol', 'application'))
 399         self.assertEqual(response['node-info']['error-tag'], 'data-missing')
 400         self.assertEqual(response['node-info']['error-message'],
 401                          'Request could not be completed because the relevant data model content does not exist')
 402
 403
 404 if __name__ == "__main__":
 405     unittest.main(verbosity=2)