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