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