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