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