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