tests/transportpce_tests/2.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 #from unittest.result import failfast
  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 test_01_rdm_device_connected(self):
  35         response = test_utils.mount_device("ROADM-A1", 'roadma')
  36         self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
  37
  38     def test_02_xpdr_device_connected(self):
  39         response = test_utils.mount_device("XPDR-A1", 'xpdra')
  40         self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
  41
  42     def test_03_rdm_portmapping(self):
  43         url = "{}/config/transportpce-portmapping:network/nodes/ROADM-A1"
  44         response = test_utils.get_request(url)
  45         self.assertEqual(response.status_code, requests.codes.ok)
  46         res = response.json()
  47         self.assertIn(
  48             {'supporting-port': 'L1', 'supporting-circuit-pack-name': '1/0',
  49              'logical-connection-point': 'DEG1-TTP-TXRX', 'port-direction': 'bidirectional'},
  50             res['nodes'][0]['mapping'])
  51         self.assertIn(
  52             {'supporting-port': 'C3', 'supporting-circuit-pack-name': '3/0',
  53              'logical-connection-point': 'SRG1-PP3-TXRX', 'port-direction': 'bidirectional'},
  54             res['nodes'][0]['mapping'])
  55
  56     def test_04_xpdr_portmapping(self):
  57         url = "{}/config/transportpce-portmapping:network/nodes/XPDR-A1"
  58         response = test_utils.get_request(url)
  59         self.assertEqual(response.status_code, requests.codes.ok)
  60         res = response.json()
  61         self.assertIn(
  62             {'supported-interface-capability': ['org-openroadm-port-types:if-OCH'],
  63              'supporting-port': '1', 'supporting-circuit-pack-name': '1/0/1-PLUG-NET',
  64              'logical-connection-point': 'XPDR1-NETWORK1', 'port-qual': 'xpdr-network',
  65              'port-direction': 'bidirectional', 'connection-map-lcp': 'XPDR1-CLIENT1',
  66              'lcp-hash-val': 'AMkDwQ7xTmRI'},
  67             res['nodes'][0]['mapping'])
  68         self.assertIn(
  69             {'supported-interface-capability': ['org-openroadm-port-types:if-100GE'],
  70              'supporting-port': 'C1',
  71              'supporting-circuit-pack-name': '1/0/1-PLUG-CLIENT',
  72              'logical-connection-point': 'XPDR1-CLIENT1', 'port-direction': 'bidirectional',
  73              'connection-map-lcp': 'XPDR1-NETWORK1', 'port-qual': 'xpdr-client',
  74              'lcp-hash-val': 'AJUUr6I5fALj'},
  75             res['nodes'][0]['mapping'])
  76
  77     def test_05_service_path_create(self):
  78         url = "{}/operations/transportpce-device-renderer:service-path"
  79         data = {"renderer:input": {
  80             "renderer:service-name": "service_test",
  81             "renderer:wave-number": "7",
  82             "renderer:modulation-format": "qpsk",
  83             "renderer:operation": "create",
  84             "renderer:nodes": [
  85                 {"renderer:node-id": "ROADM-A1",
  86                  "renderer:src-tp": "SRG1-PP3-TXRX",
  87                  "renderer:dest-tp": "DEG1-TTP-TXRX"},
  88                 {"renderer:node-id": "XPDR-A1",
  89                  "renderer:src-tp": "XPDR1-CLIENT1",
  90                  "renderer:dest-tp": "XPDR1-NETWORK1"}]}}
  91         response = test_utils.post_request(url, data)
  92         self.assertEqual(response.status_code, requests.codes.ok)
  93         res = response.json()
  94         self.assertIn('Roadm-connection successfully created for nodes: ROADM-A1', res["output"]["result"])
  95
  96     def test_06_service_path_create_rdm_check(self):
  97         response = test_utils.check_netconf_node_request("ROADM-A1", "interface/DEG1-TTP-TXRX-nmc-7")
  98         self.assertEqual(response.status_code, requests.codes.ok)
  99         res = response.json()
 100         # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
 101         self.assertDictEqual(
 102             dict({
 103                  'name': 'DEG1-TTP-TXRX-nmc-7',
 104                  'administrative-state': 'inService',
 105                  'supporting-circuit-pack-name': '1/0',
 106                  'type': 'org-openroadm-interfaces:networkMediaChannelConnectionTerminationPoint',
 107                  'supporting-port': 'L1'
 108                  }, **res['interface'][0]),
 109             res['interface'][0]
 110         )
 111         self.assertDictEqual(
 112             {u'frequency': 195.8, u'width': 40},
 113             res['interface'][0]['org-openroadm-network-media-channel-interfaces:nmc-ctp'])
 114
 115     def test_07_service_path_create_rdm_check(self):
 116         response = test_utils.check_netconf_node_request("ROADM-A1", "interface/DEG1-TTP-TXRX-mc-7")
 117         self.assertEqual(response.status_code, requests.codes.ok)
 118         res = response.json()
 119         # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
 120         self.assertDictEqual(
 121             dict({
 122                  'name': 'DEG1-TTP-TXRX-mc-7',
 123                  'administrative-state': 'inService',
 124                  'supporting-circuit-pack-name': '1/0',
 125                  'type': 'org-openroadm-interfaces:mediaChannelTrailTerminationPoint',
 126                  'supporting-port': 'L1'
 127                  }, **res['interface'][0]),
 128             res['interface'][0]
 129         )
 130         self.assertDictEqual(
 131             {u'min-freq': 195.775, u'max-freq': 195.825},
 132             res['interface'][0]['org-openroadm-media-channel-interfaces:mc-ttp'])
 133
 134     def test_08_service_path_create_rdm_check(self):
 135         response = test_utils.check_netconf_node_request("ROADM-A1", "interface/SRG1-PP3-TXRX-nmc-7")
 136         self.assertEqual(response.status_code, requests.codes.ok)
 137         res = response.json()
 138         # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
 139         self.assertDictEqual(
 140             dict({
 141                  'name': 'SRG1-PP3-TXRX-nmc-7',
 142                  'administrative-state': 'inService',
 143                  'supporting-circuit-pack-name': '3/0',
 144                  'type': 'org-openroadm-interfaces:networkMediaChannelConnectionTerminationPoint',
 145                  'supporting-port': 'C3'
 146                  }, **res['interface'][0]),
 147             res['interface'][0]
 148         )
 149         self.assertDictEqual(
 150             {u'frequency': 195.8, u'width': 40},
 151             res['interface'][0]['org-openroadm-network-media-channel-interfaces:nmc-ctp'])
 152
 153     # -mc supporting interfaces must not be created for SRG, only degrees
 154     def test_09_service_path_create_rdm_check(self):
 155         response = test_utils.check_netconf_node_request("ROADM-A1", "interface/SRG1-PP3-TXRX-mc-7")
 156         self.assertEqual(response.status_code, requests.codes.not_found)
 157         res = response.json()
 158         self.assertIn(
 159             {"error-type": "application", "error-tag": "data-missing",
 160              "error-message": "Request could not be completed because the relevant data model content does not exist"},
 161             res['errors']['error'])
 162
 163     def test_10_service_path_create_rdm_check(self):
 164         response = test_utils.check_netconf_node_request("ROADM-A1", "roadm-connections/SRG1-PP3-TXRX-DEG1-TTP-TXRX-7")
 165         self.assertEqual(response.status_code, requests.codes.ok)
 166         res = response.json()
 167         # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
 168         self.assertDictEqual(
 169             dict({
 170                  'connection-name': 'SRG1-PP3-TXRX-DEG1-TTP-TXRX-7',
 171                  'opticalControlMode': 'off'
 172                  }, **res['roadm-connections'][0]),
 173             res['roadm-connections'][0]
 174         )
 175         self.assertDictEqual(
 176             {'src-if': 'SRG1-PP3-TXRX-nmc-7'},
 177             res['roadm-connections'][0]['source'])
 178         self.assertDictEqual(
 179             {'dst-if': 'DEG1-TTP-TXRX-nmc-7'},
 180             res['roadm-connections'][0]['destination'])
 181
 182     def test_11_service_path_create_xpdr_check(self):
 183         response = test_utils.check_netconf_node_request("XPDR-A1", "interface/XPDR1-NETWORK1-7")
 184         self.assertEqual(response.status_code, requests.codes.ok)
 185         res = response.json()
 186         # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
 187         self.assertDictEqual(
 188             dict({
 189                  'name': 'XPDR1-NETWORK1-7',
 190                  'administrative-state': 'inService',
 191                  'supporting-circuit-pack-name': '1/0/1-PLUG-NET',
 192                  'type': 'org-openroadm-interfaces:opticalChannel',
 193                  'supporting-port': '1'
 194                  }, **res['interface'][0]),
 195             res['interface'][0]
 196         )
 197         self.assertDictEqual(
 198             {u'rate': u'org-openroadm-common-types:R100G',
 199              u'transmit-power': -5,
 200              u'frequency': 195.8},
 201             res['interface'][0]['org-openroadm-optical-channel-interfaces:och'])
 202
 203     def test_12_service_path_create_xpdr_check(self):
 204         response = test_utils.check_netconf_node_request("XPDR-A1", "interface/XPDR1-NETWORK1-OTU")
 205         self.assertEqual(response.status_code, requests.codes.ok)
 206         res = response.json()
 207         # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
 208         self.assertDictEqual(
 209             dict({
 210                  'name': 'XPDR1-NETWORK1-OTU',
 211                  'administrative-state': 'inService',
 212                  'supporting-circuit-pack-name': '1/0/1-PLUG-NET',
 213                  'type': 'org-openroadm-interfaces:otnOtu',
 214                  'supporting-port': '1',
 215                  'supporting-interface': 'XPDR1-NETWORK1-7'
 216                  }, **res['interface'][0]),
 217             res['interface'][0]
 218         )
 219         input_dict_2 = {'tx-sapi': 'AMkDwQ7xTmRI',
 220                         'expected-dapi': 'AMkDwQ7xTmRI',
 221                         'rate': 'org-openroadm-otn-common-types:OTU4',
 222                         'fec': 'scfec'}
 223         self.assertDictEqual(input_dict_2,
 224                              res['interface'][0]['org-openroadm-otn-otu-interfaces:otu'])
 225
 226     def test_13_service_path_create_xpdr_check(self):
 227         response = test_utils.check_netconf_node_request("XPDR-A1", "interface/XPDR1-NETWORK1-ODU")
 228         self.assertEqual(response.status_code, requests.codes.ok)
 229         res = response.json()
 230         # the 2 following statements replace self.assertDictContainsSubset deprecated in python 3.2
 231         self.assertDictEqual(
 232             dict({
 233                  'name': 'XPDR1-NETWORK1-ODU',
 234                  'administrative-state': 'inService',
 235                  'supporting-circuit-pack-name': '1/0/1-PLUG-NET',
 236                  'type': 'org-openroadm-interfaces:otnOdu',
 237                  'supporting-port': '1',
 238                  'supporting-interface': 'XPDR1-NETWORK1-OTU'
 239                  }, **res['interface'][0]),
 240             res['interface'][0]
 241         )
 242         self.assertDictEqual(
 243             dict({
 244                  'rate': 'org-openroadm-otn-common-types:ODU4',
 245                  u'monitoring-mode': u'terminated'
 246                  }, **res['interface'][0]['org-openroadm-otn-odu-interfaces:odu']),
 247             res['interface'][0]['org-openroadm-otn-odu-interfaces:odu']
 248         )
 249         self.assertDictEqual({u'exp-payload-type': u'07', u'payload-type': u'07'},
 250                              res['interface'][0]['org-openroadm-otn-odu-interfaces:odu']['opu'])
 251
 252     def test_14_service_path_create_xpdr_check(self):
 253         response = test_utils.check_netconf_node_request("XPDR-A1", "interface/XPDR1-CLIENT1-ETHERNET")
 254         self.assertEqual(response.status_code, requests.codes.ok)
 255         res = response.json()
 256         # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
 257         self.assertDictEqual(
 258             dict({
 259                  'name': 'XPDR1-CLIENT1-ETHERNET',
 260                  'administrative-state': 'inService',
 261                  'supporting-circuit-pack-name': '1/0/1-PLUG-CLIENT',
 262                  'type': 'org-openroadm-interfaces:ethernetCsmacd',
 263                  'supporting-port': 'C1'
 264                  }, **res['interface'][0]),
 265             res['interface'][0]
 266         )
 267         self.assertDictEqual(
 268             {u'fec': u'off', u'speed': 100000},
 269             res['interface'][0]['org-openroadm-ethernet-interfaces:ethernet'])
 270
 271     def test_15_service_path_create_xpdr_check(self):
 272         response = test_utils.check_netconf_node_request("XPDR-A1", "circuit-packs/1%2F0%2F1-PLUG-NET")
 273         self.assertEqual(response.status_code, requests.codes.ok)
 274         res = response.json()
 275         self.assertIn('not-reserved-inuse', res['circuit-packs'][0]["equipment-state"])
 276
 277     def test_16_service_path_create_xpdr_check(self):
 278         response = test_utils.check_netconf_node_request("XPDR-A1", "circuit-packs/1%2F0%2F1-PLUG-CLIENT")
 279         self.assertEqual(response.status_code, requests.codes.ok)
 280         res = response.json()
 281         self.assertIn('not-reserved-inuse', res['circuit-packs'][0]["equipment-state"])
 282
 283     def test_17_service_path_delete(self):
 284         url = "{}/operations/transportpce-device-renderer:service-path"
 285         data = {"renderer:input": {
 286             "renderer:service-name": "service_test",
 287             "renderer:wave-number": "7",
 288             "renderer:operation": "delete",
 289             "renderer:nodes": [
 290                 {"renderer:node-id": "ROADM-A1",
 291                  "renderer:src-tp": "SRG1-PP3-TXRX",
 292                  "renderer:dest-tp": "DEG1-TTP-TXRX"},
 293                 {"renderer:node-id": "XPDR-A1",
 294                  "renderer:src-tp": "XPDR1-CLIENT1",
 295                  "renderer:dest-tp": "XPDR1-NETWORK1"}]}}
 296         response = test_utils.post_request(url, data)
 297         self.assertEqual(response.status_code, requests.codes.ok)
 298         self.assertEqual(response.json(), {
 299             'output': {'result': 'Request processed', 'success': True}})
 300
 301     def test_18_service_path_delete_rdm_check(self):
 302         response = test_utils.check_netconf_node_request("ROADM-A1", "interface/DEG1-TTP-TXRX-mc-7")
 303         self.assertEqual(response.status_code, requests.codes.not_found)
 304         res = response.json()
 305         self.assertIn(
 306             {"error-type": "application", "error-tag": "data-missing",
 307              "error-message": "Request could not be completed because the relevant data model content does not exist"},
 308             res['errors']['error'])
 309
 310     def test_19_service_path_delete_rdm_check(self):
 311         response = test_utils.check_netconf_node_request("ROADM-A1", "interface/DEG1-TTP-TXRX-nmc-7")
 312         self.assertEqual(response.status_code, requests.codes.not_found)
 313         res = response.json()
 314         self.assertIn(
 315             {"error-type": "application", "error-tag": "data-missing",
 316              "error-message": "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_rdm_check(self):
 320         response = test_utils.check_netconf_node_request("ROADM-A1", "interface/SRG1-PP3-TXRX-mc-7")
 321         self.assertEqual(response.status_code, requests.codes.not_found)
 322         res = response.json()
 323         self.assertIn({
 324             "error-type": "application",
 325             "error-tag": "data-missing",
 326             "error-message": "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_rdm_check(self):
 330         response = test_utils.check_netconf_node_request("ROADM-A1", "interface/SRG1-PP3-TXRX-nmc-7")
 331         self.assertEqual(response.status_code, requests.codes.not_found)
 332         res = response.json()
 333         self.assertIn({
 334             "error-type": "application",
 335             "error-tag": "data-missing",
 336             "error-message": "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_rdm_check(self):
 340         response = test_utils.check_netconf_node_request("ROADM-A1", "interface/SRG1-PP3-TXRX-DEG1-TTP-TXRX-7")
 341         self.assertEqual(response.status_code, requests.codes.not_found)
 342         res = response.json()
 343         self.assertIn({
 344             "error-type": "application",
 345             "error-tag": "data-missing",
 346             "error-message": "Request could not be completed because the relevant data model content does not exist"},
 347             res['errors']['error'])
 348
 349     def test_23_service_path_delete_xpdr_check(self):
 350         response = test_utils.check_netconf_node_request("XPDR-A1", "interface/XPDR1-NETWORK1-7")
 351         self.assertEqual(response.status_code, requests.codes.not_found)
 352         res = response.json()
 353         self.assertIn({
 354             "error-type": "application",
 355             "error-tag": "data-missing",
 356             "error-message": "Request could not be completed because the relevant data model content does not exist"},
 357             res['errors']['error'])
 358
 359     def test_24_service_path_delete_xpdr_check(self):
 360         response = test_utils.check_netconf_node_request("XPDR-A1", "interface/XPDR1-NETWORK1-OTU")
 361         self.assertEqual(response.status_code, requests.codes.not_found)
 362         res = response.json()
 363         self.assertIn({
 364             "error-type": "application",
 365             "error-tag": "data-missing",
 366             "error-message": "Request could not be completed because the relevant data model content does not exist"},
 367             res['errors']['error'])
 368
 369     def test_25_service_path_delete_xpdr_check(self):
 370         response = test_utils.check_netconf_node_request("XPDR-A1", "interface/XPDR1-NETWORK1-ODU")
 371         self.assertEqual(response.status_code, requests.codes.not_found)
 372         res = response.json()
 373         self.assertIn({
 374             "error-type": "application",
 375             "error-tag": "data-missing",
 376             "error-message": "Request could not be completed because the relevant data model content does not exist"},
 377             res['errors']['error'])
 378
 379     def test_26_service_path_delete_xpdr_check(self):
 380         response = test_utils.check_netconf_node_request("XPDR-A1", "interface/XPDR1-CLIENT1-ETHERNET")
 381         self.assertEqual(response.status_code, requests.codes.not_found)
 382         res = response.json()
 383         self.assertIn({
 384             "error-type": "application",
 385             "error-tag": "data-missing",
 386             "error-message": "Request could not be completed because the relevant data model content does not exist"},
 387             res['errors']['error'])
 388
 389     def test_27_service_path_delete_xpdr_check(self):
 390         response = test_utils.check_netconf_node_request("XPDR-A1", "circuit-packs/1%2F0%2F1-PLUG-NET")
 391         self.assertEqual(response.status_code, requests.codes.ok)
 392         res = response.json()
 393         self.assertEqual('not-reserved-available', res["circuit-packs"][0]['equipment-state'])
 394
 395     def test_28_service_path_delete_xpdr_check(self):
 396         response = test_utils.check_netconf_node_request("XPDR-A1", "circuit-packs/1%2F0%2F1-PLUG-CLIENT")
 397         self.assertEqual(response.status_code, requests.codes.ok)
 398         res = response.json()
 399         self.assertEqual('not-reserved-available', res["circuit-packs"][0]['equipment-state'])
 400
 401     def test_29_rdm_device_disconnected(self):
 402         response = test_utils.unmount_device("ROADM-A1")
 403         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 404
 405     def test_30_xpdr_device_disconnected(self):
 406         response = test_utils.unmount_device("XPDR-A1")
 407         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 408
 409
 410 if __name__ == "__main__":
 411     unittest.main(verbosity=2, failfast=False)