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         self.assertIn(
 253             {"logical-connection-point": "XPDR1-CLIENT1",
 254              "supporting-circuit-pack-name": "CP1-SFP1",
 255              "supported-interface-capability": ["org-openroadm-port-types:if-10GE-ODU2e",
 256                                                 "org-openroadm-port-types:if-10GE-ODU2"],
 257              "supporting-port": "CP1-SFP1-P1",
 258              "lcp-hash-val": "FqlcrxV7p30=",
 259              "port-direction": "bidirectional",
 260              "port-admin-state": "InService",
 261              "xponder-type": "mpdr",
 262              "port-qual": "xpdr-client",
 263              "port-oper-state": "InService"},
 264             response['mapping'])
 265
 266     def test_23_spdr_portmapping_XPDR1_NETWORK1(self):
 267         response = test_utils_rfc8040.portmapping_request("SPDR-SA1", "XPDR1-NETWORK1")
 268         self.assertEqual(response['status_code'], requests.codes.ok)
 269         self.assertIn(
 270             {"logical-connection-point": "XPDR1-NETWORK1",
 271              "supporting-circuit-pack-name": "CP1-CFP0",
 272              "supported-interface-capability": ["org-openroadm-port-types:if-OCH-OTU4-ODU4"],
 273              "supporting-port": "CP1-CFP0-P1",
 274              "lcp-hash-val": "Swfw02qXGyI=",
 275              "port-direction": "bidirectional",
 276              "port-admin-state": "InService",
 277              "xponder-type": "mpdr",
 278              "port-qual": "xpdr-network",
 279              "port-oper-state": "InService"},
 280             response['mapping'])
 281
 282     def test_24_spdr_portmapping_XPDR2_CLIENT2(self):
 283         response = test_utils_rfc8040.portmapping_request("SPDR-SA1", "XPDR2-CLIENT2")
 284         self.assertEqual(response['status_code'], requests.codes.ok)
 285         self.assertIn(
 286             {'logical-connection-point': 'XPDR2-CLIENT2',
 287              'supporting-port': 'CP2-QSFP2-P1',
 288              'lcp-hash-val': 'AN/WSSRXne3t',
 289              'port-direction': 'bidirectional',
 290              'xponder-type': 'switch',
 291              'port-qual': 'switch-client',
 292              'supporting-circuit-pack-name': 'CP2-QSFP2',
 293              'port-admin-state': 'InService',
 294              'port-oper-state': 'InService',
 295              'supported-interface-capability': ['org-openroadm-port-types:if-100GE',
 296                                                 'org-openroadm-port-types:if-100GE-ODU4']},
 297             response['mapping'])
 298
 299     def test_25_spdr_portmapping_XPDR2_NETWORK2(self):
 300         response = test_utils_rfc8040.portmapping_request("SPDR-SA1", "XPDR2-NETWORK2")
 301         self.assertEqual(response['status_code'], requests.codes.ok)
 302         self.assertIn(
 303             {"logical-connection-point": "XPDR2-NETWORK2",
 304              "supporting-circuit-pack-name": "CP6-CFP",
 305              "supported-interface-capability": ["org-openroadm-port-types:if-OCH-OTU4-ODU4"],
 306              "supporting-port": "CP6-CFP-P1",
 307              "lcp-hash-val": "exT821pFtOQ=",
 308              "port-direction": "bidirectional",
 309              "port-admin-state": "InService",
 310              "xponder-type": "switch",
 311              "port-qual": "switch-network",
 312              "port-oper-state": "InService"},
 313             response['mapping'])
 314
 315     def test_26_spdr_portmapping_XPDR3_CLIENT3(self):
 316         response = test_utils_rfc8040.portmapping_request("SPDR-SA1", "XPDR3-CLIENT3")
 317         self.assertEqual(response['status_code'], requests.codes.ok)
 318         self.assertIn(
 319             {"logical-connection-point": "XPDR3-CLIENT3",
 320              "supporting-circuit-pack-name": "CP3-SFP3",
 321              "supported-interface-capability": ["org-openroadm-port-types:if-1GE-ODU0"],
 322              "supporting-port": "CP3-SFP3-P1",
 323              "lcp-hash-val": "AKsQ/HRQdtdN",
 324              "port-direction": "bidirectional",
 325              "port-admin-state": "InService",
 326              "xponder-type": "mpdr",
 327              "port-qual": "xpdr-client",
 328              "port-oper-state": "InService"},
 329             response['mapping'])
 330
 331     def test_27_spdr_portmapping_XPDR3_NETWORK1(self):
 332         response = test_utils_rfc8040.portmapping_request("SPDR-SA1", "XPDR3-NETWORK1")
 333         self.assertEqual(response['status_code'], requests.codes.ok)
 334         self.assertIn(
 335             {"logical-connection-point": "XPDR3-NETWORK1",
 336              "supporting-circuit-pack-name": "CP3-CFP0",
 337              "supported-interface-capability": ["org-openroadm-port-types:if-OCH-OTU4-ODU4"],
 338              "supporting-port": "CP3-CFP0-P1",
 339              "lcp-hash-val": "ANnxoi7K8q30",
 340              "port-direction": "bidirectional",
 341              "port-admin-state": "InService",
 342              "xponder-type": "mpdr",
 343              "port-qual": "xpdr-network",
 344              "port-oper-state": "InService"},
 345             response['mapping'])
 346
 347     def test_28_spdr_device_disconnection(self):
 348         response = test_utils_rfc8040.unmount_device("SPDR-SA1")
 349         self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
 350
 351     def test_29_xpdr_device_disconnected(self):
 352         response = test_utils_rfc8040.check_device_connection("SPDR-SA1")
 353         self.assertEqual(response['status_code'], requests.codes.conflict)
 354         self.assertIn(response['connection-status']['error-type'], ('protocol', 'application'))
 355         self.assertEqual(response['connection-status']['error-tag'], 'data-missing')
 356         self.assertEqual(response['connection-status']['error-message'],
 357                          'Request could not be completed because the relevant data model content does not exist')
 358
 359     def test_30_xpdr_device_disconnection(self):
 360         response = test_utils_rfc8040.unmount_device("XPDR-A1")
 361         self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
 362
 363     def test_31_xpdr_device_disconnected(self):
 364         response = test_utils_rfc8040.check_device_connection("XPDR-A1")
 365         self.assertEqual(response['status_code'], requests.codes.conflict)
 366         self.assertIn(response['connection-status']['error-type'], ('protocol', 'application'))
 367         self.assertEqual(response['connection-status']['error-tag'], 'data-missing')
 368         self.assertEqual(response['connection-status']['error-message'],
 369                          'Request could not be completed because the relevant data model content does not exist')
 370
 371     def test_32_xpdr_device_not_connected(self):
 372         response = test_utils_rfc8040.get_portmapping_node_info("XPDR-A1")
 373         self.assertEqual(response['status_code'], requests.codes.conflict)
 374         self.assertIn(response['node-info']['error-type'], ('protocol', 'application'))
 375         self.assertEqual(response['node-info']['error-tag'], 'data-missing')
 376         self.assertEqual(response['node-info']['error-message'],
 377                          'Request could not be completed because the relevant data model content does not exist')
 378
 379     def test_33_rdm_device_disconnection(self):
 380         response = test_utils_rfc8040.unmount_device("ROADM-A1")
 381         self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
 382
 383     def test_34_rdm_device_disconnected(self):
 384         response = test_utils_rfc8040.check_device_connection("ROADM-A1")
 385         self.assertEqual(response['status_code'], requests.codes.conflict)
 386         self.assertIn(response['connection-status']['error-type'], ('protocol', 'application'))
 387         self.assertEqual(response['connection-status']['error-tag'], 'data-missing')
 388         self.assertEqual(response['connection-status']['error-message'],
 389                          'Request could not be completed because the relevant data model content does not exist')
 390
 391     def test_35_rdm_device_not_connected(self):
 392         response = test_utils_rfc8040.get_portmapping_node_info("ROADM-A1")
 393         self.assertEqual(response['status_code'], requests.codes.conflict)
 394         self.assertIn(response['node-info']['error-type'], ('protocol', 'application'))
 395         self.assertEqual(response['node-info']['error-tag'], 'data-missing')
 396         self.assertEqual(response['node-info']['error-message'],
 397                          'Request could not be completed because the relevant data model content does not exist')
 398
 399
 400 if __name__ == "__main__":
 401     unittest.main(verbosity=2)