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