tests/transportpce_tests/2.2.1/test05_flex_grid.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
  14 import unittest
  15 import time
  16 import requests
  17 # pylint: disable=wrong-import-order
  18 import sys
  19 sys.path.append('transportpce_tests/common/')
  20 # pylint: disable=wrong-import-position
  21 # pylint: disable=import-error
  22 import test_utils  # nopep8
  23
  24
  25 class TransportPCEPortMappingTesting(unittest.TestCase):
  26
  27     processes = None
  28     NODE_VERSION = '2.2.1'
  29
  30     @classmethod
  31     def setUpClass(cls):
  32         cls.processes = test_utils.start_tpce()
  33         cls.processes = test_utils.start_sims([('roadmd', cls.NODE_VERSION)])
  34
  35     @classmethod
  36     def tearDownClass(cls):
  37         # pylint: disable=not-an-iterable
  38         for process in cls.processes:
  39             test_utils.shutdown_process(process)
  40         print("all processes killed")
  41
  42     def setUp(self):
  43         # pylint: disable=consider-using-f-string
  44         print("execution of {}".format(self.id().split(".")[-1]))
  45         time.sleep(10)
  46
  47     def test_01_rdm_device_connection(self):
  48         response = test_utils.mount_device("ROADM-D1", ('roadmd', self.NODE_VERSION))
  49         self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
  50
  51     def test_02_rdm_device_connected(self):
  52         response = test_utils.check_device_connection("ROADM-D1")
  53         self.assertEqual(response['status_code'], requests.codes.ok)
  54         self.assertEqual(response['connection-status'], 'connected')
  55         time.sleep(10)
  56
  57     def test_03_rdm_portmapping_info(self):
  58         response = test_utils.get_portmapping_node_attr("ROADM-D1", "node-info", None)
  59         self.assertEqual(response['status_code'], requests.codes.ok)
  60         self.assertEqual(
  61             {'node-type': 'rdm',
  62              'node-ip-address': '127.0.0.14',
  63              'node-clli': 'NodeD',
  64              'openroadm-version': '2.2.1',
  65              'node-vendor': 'vendorD',
  66              'node-model': 'model2'},
  67             response['node-info'])
  68         time.sleep(3)
  69
  70     def test_04_rdm_deg1_lcp(self):
  71         # pylint: disable=line-too-long
  72         response = test_utils.get_portmapping_node_attr("ROADM-D1", "mc-capabilities", "DEG1-TTP")
  73         self.assertEqual(response['status_code'], requests.codes.ok)
  74         self.assertIn(response['mc-capabilities'],
  75                       [[{'mc-node-name': 'DEG1-TTP', 'center-freq-granularity': '6.25', 'slot-width-granularity': '12.5'}],
  76                        [{'mc-node-name': 'DEG1-TTP', 'center-freq-granularity': 6.25, 'slot-width-granularity': 12.5}]])
  77         time.sleep(3)
  78
  79     def test_05_rdm_deg2_lcp(self):
  80         # pylint: disable=line-too-long
  81         response = test_utils.get_portmapping_node_attr("ROADM-D1", "mc-capabilities", "DEG2-TTP")
  82         self.assertEqual(response['status_code'], requests.codes.ok)
  83         self.assertIn(response['mc-capabilities'],
  84                       [[{'mc-node-name': 'DEG2-TTP', 'center-freq-granularity': '6.25', 'slot-width-granularity': '12.5'}],
  85                        [{'mc-node-name': 'DEG2-TTP', 'center-freq-granularity': 6.25, 'slot-width-granularity': 12.5}]])
  86         time.sleep(3)
  87
  88     def test_06_rdm_srg1_lcp(self):
  89         # pylint: disable=line-too-long
  90         response = test_utils.get_portmapping_node_attr("ROADM-D1", "mc-capabilities", "SRG1-PP")
  91         self.assertEqual(response['status_code'], requests.codes.ok)
  92         self.assertIn(response['mc-capabilities'],
  93                       [[{'mc-node-name': 'SRG1-PP', 'center-freq-granularity': '6.25', 'slot-width-granularity': '12.5'}],
  94                        [{'mc-node-name': 'SRG1-PP', 'center-freq-granularity': 6.25, 'slot-width-granularity': 12.5}]])
  95         time.sleep(3)
  96
  97     # Renderer interface creations
  98     def test_07_device_renderer(self):
  99         response = test_utils.transportpce_api_rpc_request(
 100             'transportpce-device-renderer', 'service-path',
 101             {
 102                 'modulation-format': 'dp-qpsk',
 103                 'operation': 'create',
 104                 'service-name': 'testNMC-MC',
 105                 'wave-number': '0',
 106                 'center-freq': '196.05',
 107                 'nmc-width': '80',
 108                 'nodes': [
 109                     {
 110                         'node-id': 'ROADM-D1',
 111                         'src-tp': 'SRG1-PP1-TXRX',
 112                         'dest-tp': 'DEG1-TTP-TXRX'
 113                     }
 114                 ],
 115                 'min-freq': 196.00625,
 116                 'max-freq': 196.09375,
 117                 'lower-spectral-slot-number': 749,
 118                 'higher-spectral-slot-number': 763
 119             })
 120         self.assertEqual(response['status_code'], requests.codes.ok)
 121         self.assertIn('Interfaces created successfully for nodes: ', response['output']['result'])
 122         time.sleep(10)
 123
 124     # Get Degree MC interface and check
 125     def test_08_degree_mc_interface(self):
 126         response = test_utils.check_node_attribute_request("ROADM-D1", "interface", "DEG1-TTP-TXRX-mc-749:763")
 127         self.assertEqual(response['status_code'], requests.codes.ok)
 128         self.assertDictEqual(
 129             dict({"name": "DEG1-TTP-TXRX-mc-749:763",
 130                   "supporting-interface": "OMS-DEG1-TTP-TXRX",
 131                   "supporting-circuit-pack-name": "1/0",
 132                   "circuit-id": "TBD",
 133                   "description": "TBD",
 134                   "supporting-port": "L1",
 135                   "type": "org-openroadm-interfaces:mediaChannelTrailTerminationPoint"},
 136                  **response['interface'][0]), response['interface'][0])
 137         # Check the mc-ttp max and min-freq
 138         mcttp = response['interface'][0]['org-openroadm-media-channel-interfaces:mc-ttp']
 139         self.assertEqual(float(mcttp['min-freq']), 196.00625)
 140         self.assertEqual(float(mcttp['max-freq']), 196.09375)
 141         time.sleep(3)
 142
 143     # get DEG-NMC interface and check
 144     def test_09_degree_nmc_interface(self):
 145         response = test_utils.check_node_attribute_request("ROADM-D1", "interface", "DEG1-TTP-TXRX-nmc-749:763")
 146         self.assertEqual(response['status_code'], requests.codes.ok)
 147         self.assertDictEqual(
 148             dict({"name": "DEG1-TTP-TXRX-nmc-749:763",
 149                   "supporting-interface": "DEG1-TTP-TXRX-mc-749:763",
 150                   "supporting-circuit-pack-name": "1/0",
 151                   "circuit-id": "TBD",
 152                   "description": "TBD",
 153                   "supporting-port": "L1",
 154                   "type": "org-openroadm-interfaces:networkMediaChannelConnectionTerminationPoint"},
 155                  **response['interface'][0]), response['interface'][0])
 156         nmcctp = response['interface'][0]['org-openroadm-network-media-channel-interfaces:nmc-ctp']
 157         self.assertEqual(float(nmcctp['frequency']), 196.05)
 158         self.assertEqual(float(nmcctp['width']), 80)
 159         time.sleep(3)
 160
 161     # get SRG-NMC interface
 162     def test_10_srg_nmc_interface(self):
 163         response = test_utils.check_node_attribute_request("ROADM-D1", "interface", "SRG1-PP1-TXRX-nmc-749:763")
 164         self.assertEqual(response['status_code'], requests.codes.ok)
 165         self.assertEqual(
 166             dict({"name": "SRG1-PP1-TXRX-nmc-749:763",
 167                   "supporting-circuit-pack-name": "3/0",
 168                   "circuit-id": "TBD",
 169                   "description": "TBD",
 170                   "supporting-port": "C1",
 171                   "type": "org-openroadm-interfaces:networkMediaChannelConnectionTerminationPoint"},
 172                  **response['interface'][0]), response['interface'][0])
 173         nmcctp = response['interface'][0]['org-openroadm-network-media-channel-interfaces:nmc-ctp']
 174         self.assertEqual(float(nmcctp['frequency']), 196.05)
 175         self.assertEqual(float(nmcctp['width']), 80)
 176         time.sleep(3)
 177
 178     # Create ROADM-connection
 179     def test_11_roadm_connection(self):
 180         response = test_utils.check_node_attribute_request(
 181             "ROADM-D1", "roadm-connections", "SRG1-PP1-TXRX-DEG1-TTP-TXRX-749:763")
 182         self.assertEqual(response['status_code'], requests.codes.ok)
 183         self.assertEqual("SRG1-PP1-TXRX-DEG1-TTP-TXRX-749:763",
 184                          response['roadm-connections'][0]['connection-name'])
 185         self.assertEqual("SRG1-PP1-TXRX-nmc-749:763",
 186                          response['roadm-connections'][0]['source']['src-if'])
 187         self.assertEqual("DEG1-TTP-TXRX-nmc-749:763",
 188                          response['roadm-connections'][0]['destination']['dst-if'])
 189         time.sleep(3)
 190
 191     # Delete ROADM connections and interfaces
 192
 193     # delete ROADM connection
 194     def test_12_delete_roadm_connection(self):
 195         response = test_utils.del_node_attribute_request(
 196             "ROADM-D1", "roadm-connections", "SRG1-PP1-TXRX-DEG1-TTP-TXRX-749:763")
 197         self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
 198         time.sleep(3)
 199
 200     # Delete NMC SRG interface
 201     def test_13_delete_srg_interface(self):
 202         response = test_utils.del_node_attribute_request("ROADM-D1", "interface", "SRG1-PP1-TXRX-nmc-749:763")
 203         self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
 204         time.sleep(3)
 205
 206     # Delete NMC Degree interface
 207     def test_14_delete_degree_interface(self):
 208         response = test_utils.del_node_attribute_request("ROADM-D1", "interface", "DEG1-TTP-TXRX-nmc-749:763")
 209         self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
 210         time.sleep(3)
 211
 212     # Delete MC Degree interface
 213     def test_15_delete_degree_interface(self):
 214         response = test_utils.del_node_attribute_request("ROADM-D1", "interface", "DEG1-TTP-TXRX-mc-749:763")
 215         self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
 216         time.sleep(3)
 217
 218     def test_16_disconnect_ROADM_D1(self):
 219         response = test_utils.unmount_device("ROADM-D1")
 220         self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
 221
 222
 223 if __name__ == "__main__":
 224     unittest.main(verbosity=2)