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 import unittest
  13 import json
  14 import time
  15 import requests
  16 from common import test_utils
  17
  18
  19 class TransportPCERendererTesting(unittest.TestCase):
  20
  21     processes = None
  22
  23     @classmethod
  24     def setUpClass(cls):
  25         cls.processes = test_utils.start_tpce()
  26         cls.processes = test_utils.start_sims(['xpdra', 'roadma'])
  27
  28     @classmethod
  29     def tearDownClass(cls):
  30         for process in cls.processes:
  31             test_utils.shutdown_process(process)
  32         print("all processes killed")
  33
  34     def setUp(self):
  35         print("execution of {}".format(self.id().split(".")[-1]))
  36         time.sleep(10)
  37
  38     def test_01_rdm_device_connected(self):
  39         response = test_utils.mount_device("ROADMA01", 'roadma')
  40         self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
  41
  42     def test_02_xpdr_device_connected(self):
  43         response = test_utils.mount_device("XPDRA01", 'xpdra')
  44         self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
  45
  46     def test_03_rdm_portmapping(self):
  47         url = ("{}/config/transportpce-portmapping:network/"
  48                "nodes/ROADMA01"
  49                )
  50         response = test_utils.get_request(url)
  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             res['nodes'][0]['mapping'])
  57         self.assertIn(
  58             {'supporting-port': 'C7', 'supporting-circuit-pack-name': '4/0',
  59              'logical-connection-point': 'SRG1-PP7-TXRX', 'port-direction': 'bidirectional'},
  60             res['nodes'][0]['mapping'])
  61
  62     def test_04_xpdr_portmapping(self):
  63         url = ("{}/config/transportpce-portmapping:network/"
  64                "nodes/XPDRA01"
  65                )
  66         response = test_utils.get_request(url)
  67         self.assertEqual(response.status_code, requests.codes.ok)
  68         res = response.json()
  69         self.assertIn(
  70             {'supporting-port': '1', 'supporting-circuit-pack-name': '1/0/1-PLUG-NET',
  71              'logical-connection-point': 'XPDR1-NETWORK1', 'port-direction': 'bidirectional',
  72              'connection-map-lcp': 'XPDR1-CLIENT1', 'port-qual': 'xpdr-network',
  73              'lcp-hash-val': 'OSvMgUyP+mE='},
  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             res['nodes'][0]['mapping'])
  82
  83     def test_05_service_path_create(self):
  84         url = "{}/operations/transportpce-device-renderer:service-path"
  85         data = {"renderer:input": {
  86             "renderer:service-name": "service_test",
  87             "renderer:wave-number": "7",
  88             "renderer:modulation-format": "qpsk",
  89             "renderer:operation": "create",
  90             "renderer:nodes": [
  91                 {"renderer:node-id": "ROADMA01",
  92                  "renderer:src-tp": "SRG1-PP7-TXRX",
  93                  "renderer:dest-tp": "DEG1-TTP-TXRX"},
  94                 {"renderer:node-id": "XPDRA01",
  95                  "renderer:src-tp": "XPDR1-CLIENT1",
  96                  "renderer:dest-tp": "XPDR1-NETWORK1"}]}}
  97         response = test_utils.post_request(url, data)
  98         self.assertEqual(response.status_code, requests.codes.ok)
  99         res = response.json()
 100         self.assertIn('Roadm-connection successfully created for nodes: ROADMA01', res["output"]["result"])
 101
 102     def test_06_service_path_create_rdm_check(self):
 103         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 104                "node/ROADMA01/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 105                "interface/DEG1-TTP-TXRX-7"
 106                )
 107         response = test_utils.get_request(url)
 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-7',
 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         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 127                "node/ROADMA01/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 128                "interface/SRG1-PP7-TXRX-7"
 129                )
 130         response = test_utils.get_request(url)
 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                 'name': 'SRG1-PP7-TXRX-7',
 137                 'administrative-state': 'inService',
 138                 'supporting-circuit-pack-name': '4/0',
 139                 'type': 'org-openroadm-interfaces:opticalChannel',
 140                 'supporting-port': 'C7'
 141             }, **res['interface'][0]),
 142             res['interface'][0]
 143         )
 144         self.assertDictEqual(
 145             {'wavelength-number': 7},
 146             res['interface'][0]['org-openroadm-optical-channel-interfaces:och'])
 147
 148     def test_08_service_path_create_rdm_check(self):
 149         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 150                "node/ROADMA01/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 151                "roadm-connections/SRG1-PP7-TXRX-DEG1-TTP-TXRX-7"
 152                )
 153         response = test_utils.get_request(url)
 154         self.assertEqual(response.status_code, requests.codes.ok)
 155         res = response.json()
 156         # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
 157         self.assertDictEqual(
 158             dict({
 159                 'connection-number': 'SRG1-PP7-TXRX-DEG1-TTP-TXRX-7',
 160                 'wavelength-number': 7,
 161                 'opticalControlMode': 'off'
 162             }, **res['roadm-connections'][0]),
 163             res['roadm-connections'][0]
 164         )
 165         self.assertDictEqual(
 166             {'src-if': 'SRG1-PP7-TXRX-7'},
 167             res['roadm-connections'][0]['source'])
 168         self.assertDictEqual(
 169             {'dst-if': 'DEG1-TTP-TXRX-7'},
 170             res['roadm-connections'][0]['destination'])
 171
 172     def test_09_service_path_create_xpdr_check(self):
 173         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 174                "node/XPDRA01/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 175                "interface/XPDR1-NETWORK1-7"
 176                )
 177         response = test_utils.get_request(url)
 178         self.assertEqual(response.status_code, requests.codes.ok)
 179         res = response.json()
 180         # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
 181         self.assertDictEqual(
 182             dict({
 183                 'name': 'XPDR1-NETWORK1-7',
 184                 'administrative-state': 'inService',
 185                 'supporting-circuit-pack-name': '1/0/1-PLUG-NET',
 186                 'type': 'org-openroadm-interfaces:opticalChannel',
 187                 'supporting-port': '1'
 188             }, **res['interface'][0]),
 189             res['interface'][0]
 190         )
 191         self.assertDictEqual(
 192             {u'rate': u'org-openroadm-optical-channel-interfaces:R100G',
 193              u'transmit-power': -5,
 194              u'wavelength-number': 7,
 195              u'modulation-format': u'dp-qpsk'},
 196             res['interface'][0]['org-openroadm-optical-channel-interfaces:och'])
 197
 198     def test_10_service_path_create_xpdr_check(self):
 199         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 200                "node/XPDRA01/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 201                "interface/XPDR1-NETWORK1-OTU"
 202                )
 203         response = test_utils.get_request(url)
 204         self.assertEqual(response.status_code, requests.codes.ok)
 205         res = response.json()
 206         # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
 207         self.assertDictEqual(
 208             dict({
 209                 'name': 'XPDR1-NETWORK1-OTU',
 210                 'administrative-state': 'inService',
 211                 'supporting-circuit-pack-name': '1/0/1-PLUG-NET',
 212                 'type': 'org-openroadm-interfaces:otnOtu',
 213                 'supporting-port': '1',
 214                 'supporting-interface': 'XPDR1-NETWORK1-7'
 215             }, **res['interface'][0]),
 216             res['interface'][0]
 217         )
 218         self.assertDictEqual(
 219             {u'rate': u'org-openroadm-otn-otu-interfaces:OTU4',
 220              u'fec': u'scfec'},
 221             res['interface'][0]['org-openroadm-otn-otu-interfaces:otu'])
 222
 223     def test_11_service_path_create_xpdr_check(self):
 224         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 225                "node/XPDRA01/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 226                "interface/XPDR1-NETWORK1-ODU"
 227                )
 228         response = test_utils.get_request(url)
 229         self.assertEqual(response.status_code, requests.codes.ok)
 230         res = response.json()
 231         # the 2 following statements replace self.assertDictContainsSubset deprecated in python 3.2
 232         self.assertDictEqual(
 233             dict({
 234                 'name': 'XPDR1-NETWORK1-ODU',
 235                 'administrative-state': 'inService',
 236                 'supporting-circuit-pack-name': '1/0/1-PLUG-NET',
 237                 'type': 'org-openroadm-interfaces:otnOdu',
 238                 'supporting-port': '1',
 239                 'supporting-interface': 'XPDR1-NETWORK1-OTU'
 240             }, **res['interface'][0]),
 241             res['interface'][0]
 242         )
 243         self.assertDictEqual(
 244             dict({
 245                 'rate': 'org-openroadm-otn-odu-interfaces:ODU4',
 246                 u'monitoring-mode': u'terminated'
 247             }, **res['interface'][0]['org-openroadm-otn-odu-interfaces:odu']),
 248             res['interface'][0]['org-openroadm-otn-odu-interfaces:odu']
 249         )
 250         self.assertDictEqual({u'exp-payload-type': u'07', u'payload-type': u'07'},
 251                              res['interface'][0]['org-openroadm-otn-odu-interfaces:odu']['opu'])
 252
 253     def test_12_service_path_create_xpdr_check(self):
 254         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 255                "node/XPDRA01/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 256                "interface/XPDR1-CLIENT1-ETHERNET"
 257                )
 258         response = test_utils.get_request(url)
 259         self.assertEqual(response.status_code, requests.codes.ok)
 260         res = response.json()
 261         # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
 262         self.assertDictEqual(
 263             dict({
 264                 'name': 'XPDR1-CLIENT1-ETHERNET',
 265                 'administrative-state': 'inService',
 266                 'supporting-circuit-pack-name': '1/0/C1-PLUG-CLIENT',
 267                 'type': 'org-openroadm-interfaces:ethernetCsmacd',
 268                 'supporting-port': 'C1'
 269             }, **res['interface'][0]),
 270             res['interface'][0]
 271         )
 272         self.assertDictEqual(
 273             {'speed': 100000,
 274              'mtu': 9000,
 275              'auto-negotiation': 'enabled',
 276              'duplex': 'full',
 277              'fec': 'off'},
 278             res['interface'][0]['org-openroadm-ethernet-interfaces:ethernet'])
 279
 280     def test_13_service_path_create_xpdr_check(self):
 281         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 282                "node/XPDRA01/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 283                "circuit-packs/1%2F0%2F1-PLUG-NET"
 284                )
 285         response = test_utils.get_request(url)
 286         self.assertEqual(response.status_code, requests.codes.ok)
 287         res = response.json()
 288         self.assertIn('not-reserved-inuse', res['circuit-packs'][0]["equipment-state"])
 289
 290     def test_14_service_path_delete(self):
 291         url = "{}/operations/transportpce-device-renderer:service-path"
 292         data = {"renderer:input": {
 293             "renderer:service-name": "service_test",
 294             "renderer:wave-number": "7",
 295             "renderer:operation": "delete",
 296             "renderer:nodes": [
 297                 {"renderer:node-id": "ROADMA01",
 298                  "renderer:src-tp": "SRG1-PP7-TXRX",
 299                  "renderer:dest-tp": "DEG1-TTP-TXRX"},
 300                 {"renderer:node-id": "XPDRA01",
 301                  "renderer:src-tp": "XPDR1-CLIENT1",
 302                  "renderer:dest-tp": "XPDR1-NETWORK1"}]}}
 303         response = test_utils.post_request(url, data)
 304         self.assertEqual(response.status_code, requests.codes.ok)
 305         self.assertEqual(response.json(), {
 306             'output': {'result': 'Request processed', 'success': True}})
 307
 308     def test_15_service_path_delete_rdm_check(self):
 309         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 310                "node/ROADMA01/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 311                "interface/DEG1-TTP-TXRX-7"
 312                )
 313         response = test_utils.get_request(url)
 314         self.assertEqual(response.status_code, requests.codes.not_found)
 315         res = response.json()
 316         self.assertIn(
 317             {"error-type": "application", "error-tag": "data-missing",
 318              "error-message": "Request could not be completed because the relevant data model content does not exist"},
 319             res['errors']['error'])
 320
 321     def test_16_service_path_delete_rdm_check(self):
 322         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 323                "node/ROADMA01/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 324                "interface/SRG1-PP7-TXRX-7"
 325                )
 326         response = test_utils.get_request(url)
 327         self.assertEqual(response.status_code, requests.codes.not_found)
 328         res = response.json()
 329         self.assertIn(
 330             {"error-type": "application", "error-tag": "data-missing",
 331                 "error-message": "Request could not be completed because the relevant data model content does not exist"},
 332             res['errors']['error'])
 333
 334     def test_17_service_path_delete_rdm_check(self):
 335         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 336                "node/ROADMA01/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 337                "roadm-connections/SRG1-PP7-TXRX-DEG1-TTP-TXRX-7"
 338                )
 339         response = test_utils.get_request(url)
 340         self.assertEqual(response.status_code, requests.codes.not_found)
 341         res = response.json()
 342         self.assertIn(
 343             {"error-type": "application", "error-tag": "data-missing",
 344                 "error-message": "Request could not be completed because the relevant data model content does not exist"},
 345             res['errors']['error'])
 346
 347     def test_18_service_path_delete_xpdr_check(self):
 348         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 349                "node/XPDRA01/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 350                "interface/XPDR1-NETWORK1-7"
 351                )
 352         response = test_utils.get_request(url)
 353         self.assertEqual(response.status_code, requests.codes.not_found)
 354         res = response.json()
 355         self.assertIn(
 356             {"error-type": "application", "error-tag": "data-missing",
 357                 "error-message": "Request could not be completed because the relevant data model content does not exist"},
 358             res['errors']['error'])
 359
 360     def test_19_service_path_delete_xpdr_check(self):
 361         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 362                "node/XPDRA01/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 363                "interface/XPDR1-NETWORK1-OTU"
 364                )
 365         response = test_utils.get_request(url)
 366         self.assertEqual(response.status_code, requests.codes.not_found)
 367         res = response.json()
 368         self.assertIn(
 369             {"error-type": "application", "error-tag": "data-missing",
 370                 "error-message": "Request could not be completed because the relevant data model content does not exist"},
 371             res['errors']['error'])
 372
 373     def test_20_service_path_delete_xpdr_check(self):
 374         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 375                "node/XPDRA01/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 376                "interface/XPDR1-NETWORK1-ODU"
 377                )
 378         response = test_utils.get_request(url)
 379         self.assertEqual(response.status_code, requests.codes.not_found)
 380         res = response.json()
 381         self.assertIn(
 382             {"error-type": "application", "error-tag": "data-missing",
 383                 "error-message": "Request could not be completed because the relevant data model content does not exist"},
 384             res['errors']['error'])
 385
 386     def test_21_service_path_delete_xpdr_check(self):
 387         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 388                "node/XPDRA01/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 389                "interface/XPDR1-CLIENT1-ETHERNET"
 390                )
 391         response = test_utils.get_request(url)
 392         self.assertEqual(response.status_code, requests.codes.not_found)
 393         res = response.json()
 394         self.assertIn(
 395             {"error-type": "application", "error-tag": "data-missing",
 396                 "error-message": "Request could not be completed because the relevant data model content does not exist"},
 397             res['errors']['error'])
 398
 399     def test_22_service_path_delete_xpdr_check(self):
 400         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/"
 401                "node/XPDRA01/yang-ext:mount/org-openroadm-device:org-openroadm-device/"
 402                "circuit-packs/1%2F0%2F1-PLUG-NET"
 403                )
 404         response = test_utils.get_request(url)
 405         self.assertEqual(response.status_code, requests.codes.ok)
 406         res = response.json()
 407         self.assertEqual('not-reserved-available', res["circuit-packs"][0]['equipment-state'])
 408
 409     def test_23_rdm_device_disconnected(self):
 410         response = test_utils.unmount_device("ROADMA01")
 411         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 412
 413     def test_24_xpdr_device_disconnected(self):
 414         response = test_utils.unmount_device("XPDRA01")
 415         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 416
 417
 418 if __name__ == "__main__":
 419     unittest.main(verbosity=2, failfast=False)