tests/transportpce_tests/1.2.1/test04_renderer_service_path_nominal.py

   1 #!/usr/bin/env python
   2
   3 #############################################################################
   4 # Copyright (c) 2017 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 TransportPCERendererTesting(unittest.TestCase):
  27
  28     processes = None
  29     NODE_VERSION = '1.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
  36     @classmethod
  37     def tearDownClass(cls):
  38         # pylint: disable=not-an-iterable
  39         for process in cls.processes:
  40             test_utils.shutdown_process(process)
  41         print("all processes killed")
  42
  43     def setUp(self):
  44         # pylint: disable=consider-using-f-string
  45         print("execution of {}".format(self.id().split(".")[-1]))
  46         time.sleep(10)
  47
  48     def test_01_rdm_device_connected(self):
  49         response = test_utils.mount_device("ROADMA01", ('roadma', self.NODE_VERSION))
  50         self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
  51
  52     def test_02_xpdr_device_connected(self):
  53         response = test_utils.mount_device("XPDRA01", ('xpdra', self.NODE_VERSION))
  54         self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
  55
  56     def test_03_rdm_portmapping(self):
  57         response = test_utils.portmapping_request("ROADMA01")
  58         self.assertEqual(response.status_code, requests.codes.ok)
  59         res = response.json()
  60         self.assertIn(
  61             {'supporting-port': 'L1', 'supporting-circuit-pack-name': '2/0',
  62              'logical-connection-point': 'DEG1-TTP-TXRX', 'port-direction': 'bidirectional',
  63              'port-admin-state': 'InService', 'port-oper-state': 'InService'},
  64             res['nodes'][0]['mapping'])
  65         self.assertIn(
  66             {'supporting-port': 'C7', 'supporting-circuit-pack-name': '4/0',
  67              'logical-connection-point': 'SRG1-PP7-TXRX', 'port-direction': 'bidirectional',
  68              'port-admin-state': 'InService', 'port-oper-state': 'InService'},
  69             res['nodes'][0]['mapping'])
  70
  71     def test_04_xpdr_portmapping(self):
  72         response = test_utils.portmapping_request("XPDRA01")
  73         self.assertEqual(response.status_code, requests.codes.ok)
  74         res = response.json()
  75         self.assertIn(
  76             {'supporting-port': '1', 'supporting-circuit-pack-name': '1/0/1-PLUG-NET',
  77              'logical-connection-point': 'XPDR1-NETWORK1', 'port-direction': 'bidirectional',
  78              'connection-map-lcp': 'XPDR1-CLIENT1', 'port-qual': 'xpdr-network',
  79              'lcp-hash-val': 'OSvMgUyP+mE=',
  80              'port-admin-state': 'InService', 'port-oper-state': 'InService'},
  81             res['nodes'][0]['mapping'])
  82         self.assertIn(
  83             {'supporting-port': 'C1',
  84              'supporting-circuit-pack-name': '1/0/C1-PLUG-CLIENT',
  85              'logical-connection-point': 'XPDR1-CLIENT1', 'port-direction': 'bidirectional',
  86              'connection-map-lcp': 'XPDR1-NETWORK1', 'port-qual': 'xpdr-client',
  87              'lcp-hash-val': 'AO9UFkY/TLYw',
  88              'port-admin-state': 'InService', 'port-oper-state': 'InService'},
  89             res['nodes'][0]['mapping'])
  90
  91     def test_05_service_path_create(self):
  92         response = test_utils.service_path_request("create", "service_test", "7",
  93                                                    [{"renderer:node-id": "ROADMA01",
  94                                                      "renderer:src-tp": "SRG1-PP7-TXRX",
  95                                                      "renderer:dest-tp": "DEG1-TTP-TXRX"},
  96                                                     {"renderer:node-id": "XPDRA01",
  97                                                      "renderer:src-tp": "XPDR1-CLIENT1",
  98                                                      "renderer:dest-tp": "XPDR1-NETWORK1"}],
  99                                                    195.8, 40, 195.775, 195.825, 713,
 100                                                    720)
 101         print(response.json())
 102         self.assertEqual(response.status_code, requests.codes.ok)
 103         res = response.json()
 104         self.assertIn('Roadm-connection successfully created for nodes: ROADMA01', res["output"]["result"])
 105
 106     def test_06_service_path_create_rdm_check(self):
 107         response = test_utils.check_netconf_node_request("ROADMA01", "interface/DEG1-TTP-TXRX-713:720")
 108         self.assertEqual(response.status_code, requests.codes.ok)
 109         res = response.json()
 110         # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
 111         self.assertDictEqual(
 112             dict({
 113                 'name': 'DEG1-TTP-TXRX-713:720',
 114                 'administrative-state': 'inService',
 115                 'supporting-circuit-pack-name': '2/0',
 116                 'type': 'org-openroadm-interfaces:opticalChannel',
 117                 'supporting-port': 'L1'
 118             }, **res['interface'][0]),
 119             res['interface'][0]
 120         )
 121         self.assertDictEqual(
 122             {'wavelength-number': 7},
 123             res['interface'][0]['org-openroadm-optical-channel-interfaces:och'])
 124
 125     def test_07_service_path_create_rdm_check(self):
 126         response = test_utils.check_netconf_node_request("ROADMA01", "interface/SRG1-PP7-TXRX-713:720")
 127         self.assertEqual(response.status_code, requests.codes.ok)
 128         res = response.json()
 129         # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
 130         self.assertDictEqual(
 131             dict({
 132                 'name': 'SRG1-PP7-TXRX-713:720',
 133                 'administrative-state': 'inService',
 134                 'supporting-circuit-pack-name': '4/0',
 135                 'type': 'org-openroadm-interfaces:opticalChannel',
 136                 'supporting-port': 'C7'
 137             }, **res['interface'][0]),
 138             res['interface'][0]
 139         )
 140         self.assertDictEqual(
 141             {'wavelength-number': 7},
 142             res['interface'][0]['org-openroadm-optical-channel-interfaces:och'])
 143
 144     def test_08_service_path_create_rdm_check(self):
 145         response = test_utils.check_netconf_node_request(
 146             "ROADMA01", "roadm-connections/SRG1-PP7-TXRX-DEG1-TTP-TXRX-713:720")
 147         self.assertEqual(response.status_code, requests.codes.ok)
 148         res = response.json()
 149         # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
 150         self.assertDictEqual(
 151             dict({
 152                 'connection-number': 'SRG1-PP7-TXRX-DEG1-TTP-TXRX-713:720',
 153                 'wavelength-number': 7,
 154                 'opticalControlMode': 'off'
 155             }, **res['roadm-connections'][0]),
 156             res['roadm-connections'][0]
 157         )
 158         self.assertDictEqual(
 159             {'src-if': 'SRG1-PP7-TXRX-713:720'},
 160             res['roadm-connections'][0]['source'])
 161         self.assertDictEqual(
 162             {'dst-if': 'DEG1-TTP-TXRX-713:720'},
 163             res['roadm-connections'][0]['destination'])
 164
 165     def test_09_service_path_create_xpdr_check(self):
 166         response = test_utils.check_netconf_node_request("XPDRA01", "interface/XPDR1-NETWORK1-713:720")
 167         self.assertEqual(response.status_code, requests.codes.ok)
 168         res = response.json()
 169         # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
 170         self.assertDictEqual(
 171             dict({
 172                 'name': 'XPDR1-NETWORK1-713:720',
 173                 'administrative-state': 'inService',
 174                 'supporting-circuit-pack-name': '1/0/1-PLUG-NET',
 175                 'type': 'org-openroadm-interfaces:opticalChannel',
 176                 'supporting-port': '1'
 177             }, **res['interface'][0]),
 178             res['interface'][0]
 179         )
 180         self.assertDictEqual(
 181             {'rate': 'org-openroadm-optical-channel-interfaces:R100G',
 182              'transmit-power': -5,
 183              'wavelength-number': 7,
 184              'modulation-format': 'dp-qpsk'},
 185             res['interface'][0]['org-openroadm-optical-channel-interfaces:och'])
 186
 187     def test_10_service_path_create_xpdr_check(self):
 188         response = test_utils.check_netconf_node_request("XPDRA01", "interface/XPDR1-NETWORK1-OTU")
 189         self.assertEqual(response.status_code, requests.codes.ok)
 190         res = response.json()
 191         # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
 192         self.assertDictEqual(
 193             dict({
 194                 'name': 'XPDR1-NETWORK1-OTU',
 195                 'administrative-state': 'inService',
 196                 'supporting-circuit-pack-name': '1/0/1-PLUG-NET',
 197                 'type': 'org-openroadm-interfaces:otnOtu',
 198                 'supporting-port': '1',
 199                 'supporting-interface': 'XPDR1-NETWORK1-713:720'
 200             }, **res['interface'][0]),
 201             res['interface'][0]
 202         )
 203         self.assertDictEqual(
 204             {'rate': 'org-openroadm-otn-otu-interfaces:OTU4',
 205              'fec': 'scfec'},
 206             res['interface'][0]['org-openroadm-otn-otu-interfaces:otu'])
 207
 208     def test_11_service_path_create_xpdr_check(self):
 209         response = test_utils.check_netconf_node_request("XPDRA01", "interface/XPDR1-NETWORK1-ODU")
 210         self.assertEqual(response.status_code, requests.codes.ok)
 211         res = response.json()
 212         # the 2 following statements replace self.assertDictContainsSubset deprecated in python 3.2
 213         self.assertDictEqual(
 214             dict({
 215                 'name': 'XPDR1-NETWORK1-ODU',
 216                 'administrative-state': 'inService',
 217                 'supporting-circuit-pack-name': '1/0/1-PLUG-NET',
 218                 'type': 'org-openroadm-interfaces:otnOdu',
 219                 'supporting-port': '1',
 220                 'supporting-interface': 'XPDR1-NETWORK1-OTU'
 221             }, **res['interface'][0]),
 222             res['interface'][0]
 223         )
 224         self.assertDictEqual(
 225             dict({
 226                 'rate': 'org-openroadm-otn-odu-interfaces:ODU4',
 227                 'monitoring-mode': 'terminated'
 228             }, **res['interface'][0]['org-openroadm-otn-odu-interfaces:odu']),
 229             res['interface'][0]['org-openroadm-otn-odu-interfaces:odu']
 230         )
 231         self.assertDictEqual({'exp-payload-type': '07', 'payload-type': '07'},
 232                              res['interface'][0]['org-openroadm-otn-odu-interfaces:odu']['opu'])
 233
 234     def test_12_service_path_create_xpdr_check(self):
 235         response = test_utils.check_netconf_node_request("XPDRA01", "interface/XPDR1-CLIENT1-ETHERNET")
 236         self.assertEqual(response.status_code, requests.codes.ok)
 237         res = response.json()
 238         # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
 239         self.assertDictEqual(
 240             dict({
 241                 'name': 'XPDR1-CLIENT1-ETHERNET',
 242                 'administrative-state': 'inService',
 243                 'supporting-circuit-pack-name': '1/0/C1-PLUG-CLIENT',
 244                 'type': 'org-openroadm-interfaces:ethernetCsmacd',
 245                 'supporting-port': 'C1'
 246             }, **res['interface'][0]),
 247             res['interface'][0]
 248         )
 249         self.assertDictEqual(
 250             {'speed': 100000,
 251              'mtu': 9000,
 252              'auto-negotiation': 'enabled',
 253              'duplex': 'full',
 254              'fec': 'off'},
 255             res['interface'][0]['org-openroadm-ethernet-interfaces:ethernet'])
 256
 257     def test_13_service_path_create_xpdr_check(self):
 258         response = test_utils.check_netconf_node_request("XPDRA01", "circuit-packs/1%2F0%2F1-PLUG-NET")
 259         self.assertEqual(response.status_code, requests.codes.ok)
 260         res = response.json()
 261         self.assertIn('not-reserved-inuse', res['circuit-packs'][0]["equipment-state"])
 262
 263     def test_14_service_path_delete(self):
 264         response = test_utils.service_path_request("delete", "service_test", "7",
 265                                                    [{"renderer:node-id": "ROADMA01",
 266                                                      "renderer:src-tp": "SRG1-PP7-TXRX",
 267                                                      "renderer:dest-tp": "DEG1-TTP-TXRX"},
 268                                                     {"renderer:node-id": "XPDRA01",
 269                                                      "renderer:src-tp": "XPDR1-CLIENT1",
 270                                                      "renderer:dest-tp": "XPDR1-NETWORK1"}],
 271                                                    195.8, 40, 195.775, 195.825, 713,
 272                                                    720)
 273         self.assertEqual(response.status_code, requests.codes.ok)
 274         self.assertEqual(response.json(), {
 275             'output': {'result': 'Request processed', 'success': True}})
 276
 277     def test_15_service_path_delete_rdm_check(self):
 278         response = test_utils.check_netconf_node_request("ROADMA01", "interface/DEG1-TTP-TXRX-713:720")
 279         self.assertEqual(response.status_code, requests.codes.conflict)
 280         res = response.json()
 281         self.assertIn(
 282             {"error-type": "application", "error-tag": "data-missing",
 283              "error-message": "Request could not be completed because the relevant data model content does not exist"},
 284             res['errors']['error'])
 285
 286     def test_16_service_path_delete_rdm_check(self):
 287         response = test_utils.check_netconf_node_request("ROADMA01", "interface/SRG1-PP7-TXRX-713:720")
 288         self.assertEqual(response.status_code, requests.codes.conflict)
 289         res = response.json()
 290         self.assertIn(
 291             {"error-type": "application", "error-tag": "data-missing",
 292                 "error-message":
 293                     "Request could not be completed because the relevant data model content does not exist"},
 294             res['errors']['error'])
 295
 296     def test_17_service_path_delete_rdm_check(self):
 297         response = test_utils.check_netconf_node_request(
 298             "ROADMA01", "roadm-connections/SRG1-PP7-TXRX-DEG1-TTP-TXRX-713:720")
 299         self.assertEqual(response.status_code, requests.codes.conflict)
 300         res = response.json()
 301         self.assertIn(
 302             {"error-type": "application", "error-tag": "data-missing",
 303                 "error-message":
 304                     "Request could not be completed because the relevant data model content does not exist"},
 305             res['errors']['error'])
 306
 307     def test_18_service_path_delete_xpdr_check(self):
 308         response = test_utils.check_netconf_node_request("XPDRA01", "interface/XPDR1-NETWORK1-713:720")
 309         self.assertEqual(response.status_code, requests.codes.conflict)
 310         res = response.json()
 311         self.assertIn(
 312             {"error-type": "application", "error-tag": "data-missing",
 313                 "error-message":
 314                     "Request could not be completed because the relevant data model content does not exist"},
 315             res['errors']['error'])
 316
 317     def test_19_service_path_delete_xpdr_check(self):
 318         response = test_utils.check_netconf_node_request("XPDRA01", "interface/XPDR1-NETWORK1-OTU")
 319         self.assertEqual(response.status_code, requests.codes.conflict)
 320         res = response.json()
 321         self.assertIn(
 322             {"error-type": "application", "error-tag": "data-missing",
 323                 "error-message":
 324                     "Request could not be completed because the relevant data model content does not exist"},
 325             res['errors']['error'])
 326
 327     def test_20_service_path_delete_xpdr_check(self):
 328         response = test_utils.check_netconf_node_request("XPDRA01", "interface/XPDR1-NETWORK1-ODU")
 329         self.assertEqual(response.status_code, requests.codes.conflict)
 330         res = response.json()
 331         self.assertIn(
 332             {"error-type": "application", "error-tag": "data-missing",
 333                 "error-message":
 334                     "Request could not be completed because the relevant data model content does not exist"},
 335             res['errors']['error'])
 336
 337     def test_21_service_path_delete_xpdr_check(self):
 338         response = test_utils.check_netconf_node_request("XPDRA01", "interface/XPDR1-CLIENT1-ETHERNET")
 339         self.assertEqual(response.status_code, requests.codes.conflict)
 340         res = response.json()
 341         self.assertIn(
 342             {"error-type": "application", "error-tag": "data-missing",
 343                 "error-message":
 344                     "Request could not be completed because the relevant data model content does not exist"},
 345             res['errors']['error'])
 346
 347     def test_22_service_path_delete_xpdr_check(self):
 348         response = test_utils.check_netconf_node_request("XPDRA01", "circuit-packs/1%2F0%2F1-PLUG-NET")
 349         self.assertEqual(response.status_code, requests.codes.ok)
 350         res = response.json()
 351         self.assertEqual('not-reserved-available', res["circuit-packs"][0]['equipment-state'])
 352
 353     def test_23_rdm_device_disconnected(self):
 354         response = test_utils.unmount_device("ROADMA01")
 355         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 356
 357     def test_24_xpdr_device_disconnected(self):
 358         response = test_utils.unmount_device("XPDRA01")
 359         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 360
 361
 362 if __name__ == "__main__":
 363     unittest.main(verbosity=2, failfast=False)