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         response = test_utils.service_path_request("create", "service_test", "7",
  79                                                    [{"renderer:node-id": "ROADMA01",
  80                                                      "renderer:src-tp": "SRG1-PP7-TXRX", "renderer:dest-tp": "DEG1-TTP-TXRX"},
  81                                                     {"renderer:node-id": "XPDRA01",
  82                                                      "renderer:src-tp": "XPDR1-CLIENT1", "renderer:dest-tp": "XPDR1-NETWORK1"}])
  83         self.assertEqual(response.status_code, requests.codes.ok)
  84         res = response.json()
  85         self.assertIn('Roadm-connection successfully created for nodes: ROADMA01', res["output"]["result"])
  86
  87     def test_06_service_path_create_rdm_check(self):
  88         response = test_utils.check_netconf_node_request("ROADMA01", "interface/DEG1-TTP-TXRX-7")
  89         self.assertEqual(response.status_code, requests.codes.ok)
  90         res = response.json()
  91         # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
  92         self.assertDictEqual(
  93             dict({
  94                 'name': 'DEG1-TTP-TXRX-7',
  95                 'administrative-state': 'inService',
  96                 'supporting-circuit-pack-name': '2/0',
  97                 'type': 'org-openroadm-interfaces:opticalChannel',
  98                 'supporting-port': 'L1'
  99             }, **res['interface'][0]),
 100             res['interface'][0]
 101         )
 102         self.assertDictEqual(
 103             {'wavelength-number': 7},
 104             res['interface'][0]['org-openroadm-optical-channel-interfaces:och'])
 105
 106     def test_07_service_path_create_rdm_check(self):
 107         response = test_utils.check_netconf_node_request("ROADMA01", "interface/SRG1-PP7-TXRX-7")
 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': 'SRG1-PP7-TXRX-7',
 114                 'administrative-state': 'inService',
 115                 'supporting-circuit-pack-name': '4/0',
 116                 'type': 'org-openroadm-interfaces:opticalChannel',
 117                 'supporting-port': 'C7'
 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_08_service_path_create_rdm_check(self):
 126         response = test_utils.check_netconf_node_request("ROADMA01", "roadm-connections/SRG1-PP7-TXRX-DEG1-TTP-TXRX-7")
 127         self.assertEqual(response.status_code, requests.codes.ok)
 128         res = response.json()
 129         # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
 130         self.assertDictEqual(
 131             dict({
 132                 'connection-number': 'SRG1-PP7-TXRX-DEG1-TTP-TXRX-7',
 133                 'wavelength-number': 7,
 134                 'opticalControlMode': 'off'
 135             }, **res['roadm-connections'][0]),
 136             res['roadm-connections'][0]
 137         )
 138         self.assertDictEqual(
 139             {'src-if': 'SRG1-PP7-TXRX-7'},
 140             res['roadm-connections'][0]['source'])
 141         self.assertDictEqual(
 142             {'dst-if': 'DEG1-TTP-TXRX-7'},
 143             res['roadm-connections'][0]['destination'])
 144
 145     def test_09_service_path_create_xpdr_check(self):
 146         response = test_utils.check_netconf_node_request("XPDRA01", "interface/XPDR1-NETWORK1-7")
 147         self.assertEqual(response.status_code, requests.codes.ok)
 148         res = response.json()
 149         # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
 150         self.assertDictEqual(
 151             dict({
 152                 'name': 'XPDR1-NETWORK1-7',
 153                 'administrative-state': 'inService',
 154                 'supporting-circuit-pack-name': '1/0/1-PLUG-NET',
 155                 'type': 'org-openroadm-interfaces:opticalChannel',
 156                 'supporting-port': '1'
 157             }, **res['interface'][0]),
 158             res['interface'][0]
 159         )
 160         self.assertDictEqual(
 161             {u'rate': u'org-openroadm-optical-channel-interfaces:R100G',
 162              u'transmit-power': -5,
 163              u'wavelength-number': 7,
 164              u'modulation-format': u'dp-qpsk'},
 165             res['interface'][0]['org-openroadm-optical-channel-interfaces:och'])
 166
 167     def test_10_service_path_create_xpdr_check(self):
 168         response = test_utils.check_netconf_node_request("XPDRA01", "interface/XPDR1-NETWORK1-OTU")
 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                 'name': 'XPDR1-NETWORK1-OTU',
 175                 'administrative-state': 'inService',
 176                 'supporting-circuit-pack-name': '1/0/1-PLUG-NET',
 177                 'type': 'org-openroadm-interfaces:otnOtu',
 178                 'supporting-port': '1',
 179                 'supporting-interface': 'XPDR1-NETWORK1-7'
 180             }, **res['interface'][0]),
 181             res['interface'][0]
 182         )
 183         self.assertDictEqual(
 184             {u'rate': u'org-openroadm-otn-otu-interfaces:OTU4',
 185              u'fec': u'scfec'},
 186             res['interface'][0]['org-openroadm-otn-otu-interfaces:otu'])
 187
 188     def test_11_service_path_create_xpdr_check(self):
 189         response = test_utils.check_netconf_node_request("XPDRA01", "interface/XPDR1-NETWORK1-ODU")
 190         self.assertEqual(response.status_code, requests.codes.ok)
 191         res = response.json()
 192         # the 2 following statements replace self.assertDictContainsSubset deprecated in python 3.2
 193         self.assertDictEqual(
 194             dict({
 195                 'name': 'XPDR1-NETWORK1-ODU',
 196                 'administrative-state': 'inService',
 197                 'supporting-circuit-pack-name': '1/0/1-PLUG-NET',
 198                 'type': 'org-openroadm-interfaces:otnOdu',
 199                 'supporting-port': '1',
 200                 'supporting-interface': 'XPDR1-NETWORK1-OTU'
 201             }, **res['interface'][0]),
 202             res['interface'][0]
 203         )
 204         self.assertDictEqual(
 205             dict({
 206                 'rate': 'org-openroadm-otn-odu-interfaces:ODU4',
 207                 u'monitoring-mode': u'terminated'
 208             }, **res['interface'][0]['org-openroadm-otn-odu-interfaces:odu']),
 209             res['interface'][0]['org-openroadm-otn-odu-interfaces:odu']
 210         )
 211         self.assertDictEqual({u'exp-payload-type': u'07', u'payload-type': u'07'},
 212                              res['interface'][0]['org-openroadm-otn-odu-interfaces:odu']['opu'])
 213
 214     def test_12_service_path_create_xpdr_check(self):
 215         response = test_utils.check_netconf_node_request("XPDRA01", "interface/XPDR1-CLIENT1-ETHERNET")
 216         self.assertEqual(response.status_code, requests.codes.ok)
 217         res = response.json()
 218         # the following statement replaces self.assertDictContainsSubset deprecated in python 3.2
 219         self.assertDictEqual(
 220             dict({
 221                 'name': 'XPDR1-CLIENT1-ETHERNET',
 222                 'administrative-state': 'inService',
 223                 'supporting-circuit-pack-name': '1/0/C1-PLUG-CLIENT',
 224                 'type': 'org-openroadm-interfaces:ethernetCsmacd',
 225                 'supporting-port': 'C1'
 226             }, **res['interface'][0]),
 227             res['interface'][0]
 228         )
 229         self.assertDictEqual(
 230             {'speed': 100000,
 231              'mtu': 9000,
 232              'auto-negotiation': 'enabled',
 233              'duplex': 'full',
 234              'fec': 'off'},
 235             res['interface'][0]['org-openroadm-ethernet-interfaces:ethernet'])
 236
 237     def test_13_service_path_create_xpdr_check(self):
 238         response = test_utils.check_netconf_node_request("XPDRA01", "circuit-packs/1%2F0%2F1-PLUG-NET")
 239         self.assertEqual(response.status_code, requests.codes.ok)
 240         res = response.json()
 241         self.assertIn('not-reserved-inuse', res['circuit-packs'][0]["equipment-state"])
 242
 243     def test_14_service_path_delete(self):
 244         response = test_utils.service_path_request("delete", "service_test", "7",
 245                                                    [{"renderer:node-id": "ROADMA01",
 246                                                      "renderer:src-tp": "SRG1-PP7-TXRX", "renderer:dest-tp": "DEG1-TTP-TXRX"},
 247                                                     {"renderer:node-id": "XPDRA01",
 248                                                      "renderer:src-tp": "XPDR1-CLIENT1", "renderer:dest-tp": "XPDR1-NETWORK1"}])
 249         self.assertEqual(response.status_code, requests.codes.ok)
 250         self.assertEqual(response.json(), {
 251             'output': {'result': 'Request processed', 'success': True}})
 252
 253     def test_15_service_path_delete_rdm_check(self):
 254         response = test_utils.check_netconf_node_request("ROADMA01", "interface/DEG1-TTP-TXRX-7")
 255         self.assertEqual(response.status_code, requests.codes.conflict)
 256         res = response.json()
 257         self.assertIn(
 258             {"error-type": "application", "error-tag": "data-missing",
 259              "error-message": "Request could not be completed because the relevant data model content does not exist"},
 260             res['errors']['error'])
 261
 262     def test_16_service_path_delete_rdm_check(self):
 263         response = test_utils.check_netconf_node_request("ROADMA01", "interface/SRG1-PP7-TXRX-7")
 264         self.assertEqual(response.status_code, requests.codes.conflict)
 265         res = response.json()
 266         self.assertIn(
 267             {"error-type": "application", "error-tag": "data-missing",
 268                 "error-message": "Request could not be completed because the relevant data model content does not exist"},
 269             res['errors']['error'])
 270
 271     def test_17_service_path_delete_rdm_check(self):
 272         response = test_utils.check_netconf_node_request("ROADMA01", "roadm-connections/SRG1-PP7-TXRX-DEG1-TTP-TXRX-7")
 273         self.assertEqual(response.status_code, requests.codes.conflict)
 274         res = response.json()
 275         self.assertIn(
 276             {"error-type": "application", "error-tag": "data-missing",
 277                 "error-message": "Request could not be completed because the relevant data model content does not exist"},
 278             res['errors']['error'])
 279
 280     def test_18_service_path_delete_xpdr_check(self):
 281         response = test_utils.check_netconf_node_request("XPDRA01", "interface/XPDR1-NETWORK1-7")
 282         self.assertEqual(response.status_code, requests.codes.conflict)
 283         res = response.json()
 284         self.assertIn(
 285             {"error-type": "application", "error-tag": "data-missing",
 286                 "error-message": "Request could not be completed because the relevant data model content does not exist"},
 287             res['errors']['error'])
 288
 289     def test_19_service_path_delete_xpdr_check(self):
 290         response = test_utils.check_netconf_node_request("XPDRA01", "interface/XPDR1-NETWORK1-OTU")
 291         self.assertEqual(response.status_code, requests.codes.conflict)
 292         res = response.json()
 293         self.assertIn(
 294             {"error-type": "application", "error-tag": "data-missing",
 295                 "error-message": "Request could not be completed because the relevant data model content does not exist"},
 296             res['errors']['error'])
 297
 298     def test_20_service_path_delete_xpdr_check(self):
 299         response = test_utils.check_netconf_node_request("XPDRA01", "interface/XPDR1-NETWORK1-ODU")
 300         self.assertEqual(response.status_code, requests.codes.conflict)
 301         res = response.json()
 302         self.assertIn(
 303             {"error-type": "application", "error-tag": "data-missing",
 304                 "error-message": "Request could not be completed because the relevant data model content does not exist"},
 305             res['errors']['error'])
 306
 307     def test_21_service_path_delete_xpdr_check(self):
 308         response = test_utils.check_netconf_node_request("XPDRA01", "interface/XPDR1-CLIENT1-ETHERNET")
 309         self.assertEqual(response.status_code, requests.codes.conflict)
 310         res = response.json()
 311         self.assertIn(
 312             {"error-type": "application", "error-tag": "data-missing",
 313                 "error-message": "Request could not be completed because the relevant data model content does not exist"},
 314             res['errors']['error'])
 315
 316     def test_22_service_path_delete_xpdr_check(self):
 317         response = test_utils.check_netconf_node_request("XPDRA01", "circuit-packs/1%2F0%2F1-PLUG-NET")
 318         self.assertEqual(response.status_code, requests.codes.ok)
 319         res = response.json()
 320         self.assertEqual('not-reserved-available', res["circuit-packs"][0]['equipment-state'])
 321
 322     def test_23_rdm_device_disconnected(self):
 323         response = test_utils.unmount_device("ROADMA01")
 324         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 325
 326     def test_24_xpdr_device_disconnected(self):
 327         response = test_utils.unmount_device("XPDRA01")
 328         self.assertEqual(response.status_code, requests.codes.ok, test_utils.CODE_SHOULD_BE_200)
 329
 330
 331 if __name__ == "__main__":
 332     unittest.main(verbosity=2, failfast=False)