tests/transportpce_tests/2.2.1/test_portmapping.py

   1 #!/usr/bin/env python
   2
   3 ##############################################################################
   4 # Copyright (c) 2019 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 json
  13 import os
  14 import psutil
  15 import requests
  16 import signal
  17 import shutil
  18 import subprocess
  19 import time
  20 import unittest
  21 import test_utils
  22
  23
  24 class TransportPCEPortMappingTesting(unittest.TestCase):
  25
  26     sim_process1 = None
  27     sim_process2 = None
  28     odl_process = None
  29     restconf_baseurl = "http://localhost:8181/restconf"
  30
  31     @classmethod
  32     def setUpClass(cls):
  33         cls.odl_process = test_utils.start_tpce()
  34         cls.sim_process1 = test_utils.start_sim('xpdra')
  35         cls.sim_process2 = test_utils.start_sim('roadma')
  36
  37     @classmethod
  38     def tearDownClass(cls):
  39         for child in psutil.Process(cls.odl_process.pid).children():
  40             child.send_signal(signal.SIGINT)
  41             child.wait()
  42         cls.odl_process.send_signal(signal.SIGINT)
  43         cls.odl_process.wait()
  44         for child in psutil.Process(cls.sim_process1.pid).children():
  45             child.send_signal(signal.SIGINT)
  46             child.wait()
  47         cls.sim_process1.send_signal(signal.SIGINT)
  48         cls.sim_process1.wait()
  49         for child in psutil.Process(cls.sim_process2.pid).children():
  50             child.send_signal(signal.SIGINT)
  51             child.wait()
  52         cls.sim_process2.send_signal(signal.SIGINT)
  53         cls.sim_process2.wait()
  54
  55     def setUp(self):
  56         print("execution of {}".format(self.id().split(".")[-1]))
  57         time.sleep(10)
  58
  59     def test_01_rdm_device_connected(self):
  60         url = ("{}/config/network-topology:"
  61                "network-topology/topology/topology-netconf/node/ROADM-A1"
  62                .format(self.restconf_baseurl))
  63         data = {"node": [{
  64             "node-id": "ROADM-A1",
  65             "netconf-node-topology:username": "admin",
  66             "netconf-node-topology:password": "admin",
  67             "netconf-node-topology:host": "127.0.0.1",
  68             "netconf-node-topology:port": test_utils.sims['roadma']['port'],
  69             "netconf-node-topology:tcp-only": "false",
  70             "netconf-node-topology:pass-through": {}}]}
  71         headers = {'content-type': 'application/json'}
  72         response = requests.request(
  73             "PUT", url, data=json.dumps(data), headers=headers,
  74             auth=('admin', 'admin'))
  75         self.assertEqual(response.status_code, requests.codes.created)
  76         time.sleep(20)
  77
  78     def test_02_rdm_device_connected(self):
  79         url = ("{}/operational/network-topology:"
  80                "network-topology/topology/topology-netconf/node/ROADM-A1"
  81                .format(self.restconf_baseurl))
  82         headers = {'content-type': 'application/json'}
  83         response = requests.request(
  84             "GET", url, headers=headers, auth=('admin', 'admin'))
  85         self.assertEqual(response.status_code, requests.codes.ok)
  86         res = response.json()
  87         self.assertEqual(
  88             res['node'][0]['netconf-node-topology:connection-status'],
  89             'connected')
  90         time.sleep(10)
  91
  92     def test_03_rdm_portmapping_info(self):
  93         url = ("{}/config/transportpce-portmapping:network/"
  94                "nodes/ROADM-A1/node-info"
  95                .format(self.restconf_baseurl))
  96         headers = {'content-type': 'application/json'}
  97         response = requests.request(
  98             "GET", url, headers=headers, auth=('admin', 'admin'))
  99         self.assertEqual(response.status_code, requests.codes.ok)
 100         res = response.json()
 101         self.assertEqual(
 102             {u'node-info': {u'node-type': u'rdm',
 103                             u'node-ip-address': u'127.0.0.11',
 104                             u'node-clli': u'NodeA',
 105                             u'openroadm-version': u'2.2.1', u'node-vendor': u'vendorA',
 106                             u'node-model': u'model2'}},
 107             res)
 108         time.sleep(3)
 109
 110     def test_04_rdm_portmapping_DEG1_TTP_TXRX(self):
 111         url = ("{}/config/transportpce-portmapping:network/"
 112                "nodes/ROADM-A1/mapping/DEG1-TTP-TXRX"
 113                .format(self.restconf_baseurl))
 114         headers = {'content-type': 'application/json'}
 115         response = requests.request(
 116             "GET", url, headers=headers, auth=('admin', 'admin'))
 117         self.assertEqual(response.status_code, requests.codes.ok)
 118         res = response.json()
 119         self.assertIn(
 120             {'supporting-port': 'L1', 'supporting-circuit-pack-name': '1/0',
 121              'logical-connection-point': 'DEG1-TTP-TXRX', 'port-direction': 'bidirectional'},
 122             res['mapping'])
 123
 124     def test_05_rdm_portmapping_DEG2_TTP_TXRX_with_ots_oms(self):
 125         url = ("{}/config/transportpce-portmapping:network/"
 126                "nodes/ROADM-A1/mapping/DEG2-TTP-TXRX"
 127                .format(self.restconf_baseurl))
 128         headers = {'content-type': 'application/json'}
 129         response = requests.request(
 130             "GET", url, headers=headers, auth=('admin', 'admin'))
 131         self.assertEqual(response.status_code, requests.codes.ok)
 132         res = response.json()
 133         self.assertIn(
 134             {'supporting-port': 'L1', 'supporting-circuit-pack-name': '2/0',
 135              'logical-connection-point': 'DEG2-TTP-TXRX',
 136              'supporting-oms': 'OMS-DEG2-TTP-TXRX', 'supporting-ots': 'OTS-DEG2-TTP-TXRX',
 137              'port-direction': 'bidirectional'},
 138             res['mapping'])
 139
 140     def test_06_rdm_portmapping_SRG1_PP3_TXRX(self):
 141         url = ("{}/config/transportpce-portmapping:network/"
 142                "nodes/ROADM-A1/mapping/SRG1-PP3-TXRX"
 143                .format(self.restconf_baseurl))
 144         headers = {'content-type': 'application/json'}
 145         response = requests.request(
 146             "GET", url, headers=headers, auth=('admin', 'admin'))
 147         self.assertEqual(response.status_code, requests.codes.ok)
 148         res = response.json()
 149         self.assertIn(
 150             {'supporting-port': 'C3', 'supporting-circuit-pack-name': '3/0',
 151              'logical-connection-point': 'SRG1-PP3-TXRX', 'port-direction': 'bidirectional'},
 152             res['mapping'])
 153
 154     def test_07_rdm_portmapping_SRG3_PP1_TXRX(self):
 155         url = ("{}/config/transportpce-portmapping:network/"
 156                "nodes/ROADM-A1/mapping/SRG3-PP1-TXRX"
 157                .format(self.restconf_baseurl))
 158         headers = {'content-type': 'application/json'}
 159         response = requests.request(
 160             "GET", url, headers=headers, auth=('admin', 'admin'))
 161         self.assertEqual(response.status_code, requests.codes.ok)
 162         res = response.json()
 163         self.assertIn(
 164             {'supporting-port': 'C1', 'supporting-circuit-pack-name': '5/0',
 165              'logical-connection-point': 'SRG3-PP1-TXRX', 'port-direction': 'bidirectional'},
 166             res['mapping'])
 167
 168     def test_08_xpdr_device_connected(self):
 169         url = ("{}/config/network-topology:"
 170                "network-topology/topology/topology-netconf/node/XPDR-A1"
 171                .format(self.restconf_baseurl))
 172         data = {"node": [{
 173             "node-id": "XPDR-A1",
 174             "netconf-node-topology:username": "admin",
 175             "netconf-node-topology:password": "admin",
 176             "netconf-node-topology:host": "127.0.0.1",
 177             "netconf-node-topology:port": test_utils.sims['xpdra']['port'],
 178             "netconf-node-topology:tcp-only": "false",
 179             "netconf-node-topology:pass-through": {}}]}
 180         headers = {'content-type': 'application/json'}
 181         response = requests.request(
 182             "PUT", url, data=json.dumps(data), headers=headers,
 183             auth=('admin', 'admin'))
 184         self.assertEqual(response.status_code, requests.codes.created)
 185         time.sleep(20)
 186
 187     def test_09_xpdr_device_connected(self):
 188         url = ("{}/operational/network-topology:"
 189                "network-topology/topology/topology-netconf/node/XPDR-A1"
 190                .format(self.restconf_baseurl))
 191         headers = {'content-type': 'application/json'}
 192         response = requests.request(
 193             "GET", url, headers=headers, auth=('admin', 'admin'))
 194         self.assertEqual(response.status_code, requests.codes.ok)
 195         res = response.json()
 196         self.assertEqual(
 197             res['node'][0]['netconf-node-topology:connection-status'],
 198             'connected')
 199         time.sleep(10)
 200
 201     def test_10_xpdr_portmapping_info(self):
 202         url = ("{}/config/transportpce-portmapping:network/"
 203                "nodes/XPDR-A1/node-info"
 204                .format(self.restconf_baseurl))
 205         headers = {'content-type': 'application/json'}
 206         response = requests.request(
 207             "GET", url, headers=headers, auth=('admin', 'admin'))
 208         self.assertEqual(response.status_code, requests.codes.ok)
 209         res = response.json()
 210         self.assertEqual(
 211             {u'node-info': {u'node-type': u'xpdr',
 212                             u'node-ip-address': u'1.2.3.4',
 213                             u'node-clli': u'NodeA',
 214                             u'openroadm-version': u'2.2.1', u'node-vendor': u'vendorA',
 215                             u'node-model': u'model2'}},
 216             res)
 217         time.sleep(3)
 218
 219     def test_11_xpdr_portmapping_NETWORK1(self):
 220         url = ("{}/config/transportpce-portmapping:network/"
 221                "nodes/XPDR-A1/mapping/XPDR1-NETWORK1"
 222                .format(self.restconf_baseurl))
 223         headers = {'content-type': 'application/json'}
 224         response = requests.request(
 225             "GET", url, headers=headers, auth=('admin', 'admin'))
 226         self.assertEqual(response.status_code, requests.codes.ok)
 227         res = response.json()
 228         self.assertIn(
 229             {'supported-interface-capability': ['org-openroadm-port-types:if-OCH'],
 230              'supporting-port': '1', 'supporting-circuit-pack-name': '1/0/1-PLUG-NET',
 231              'logical-connection-point': 'XPDR1-NETWORK1', 'port-qual': 'xpdr-network',
 232              'port-direction': 'bidirectional', 'connection-map-lcp': 'XPDR1-CLIENT1',
 233              'lcp-hash-val': '8e128ba57560403cfd4ffafae38cd941'},
 234             res['mapping'])
 235
 236     def test_12_xpdr_portmapping_NETWORK2(self):
 237         url = ("{}/config/transportpce-portmapping:network/"
 238                "nodes/XPDR-A1/mapping/XPDR1-NETWORK2"
 239                .format(self.restconf_baseurl))
 240         headers = {'content-type': 'application/json'}
 241         response = requests.request(
 242             "GET", url, headers=headers, auth=('admin', 'admin'))
 243         self.assertEqual(response.status_code, requests.codes.ok)
 244         res = response.json()
 245         self.assertIn(
 246             {'supported-interface-capability': ['org-openroadm-port-types:if-OCH'],
 247              'supporting-port': '1', 'supporting-circuit-pack-name': '1/0/2-PLUG-NET',
 248              'logical-connection-point': 'XPDR1-NETWORK2', 'port-direction': 'bidirectional',
 249              'connection-map-lcp': 'XPDR1-CLIENT2', 'port-qual': 'xpdr-network',
 250              'lcp-hash-val': '8e128ba57560403cfd4ffafae38cd942'},
 251             res['mapping'])
 252
 253     def test_13_xpdr_portmapping_CLIENT1(self):
 254         url = ("{}/config/transportpce-portmapping:network/"
 255                "nodes/XPDR-A1/mapping/XPDR1-CLIENT1"
 256                .format(self.restconf_baseurl))
 257         headers = {'content-type': 'application/json'}
 258         response = requests.request(
 259             "GET", url, headers=headers, auth=('admin', 'admin'))
 260         self.assertEqual(response.status_code, requests.codes.ok)
 261         res = response.json()
 262         self.assertIn(
 263             {'supported-interface-capability': ['org-openroadm-port-types:if-100GE'],
 264              'supporting-port': 'C1',
 265              'supporting-circuit-pack-name': '1/0/1-PLUG-CLIENT',
 266              'logical-connection-point': 'XPDR1-CLIENT1', 'port-direction': 'bidirectional',
 267              'connection-map-lcp': 'XPDR1-NETWORK1', 'port-qual': 'xpdr-client',
 268              'lcp-hash-val': '3ed8ed1336784ac7c2f66c22f2f03d8'},
 269             res['mapping'])
 270
 271     def test_14_xpdr_portmapping_CLIENT2(self):
 272         url = ("{}/config/transportpce-portmapping:network/"
 273                "nodes/XPDR-A1/mapping/XPDR1-CLIENT2"
 274                .format(self.restconf_baseurl))
 275         headers = {'content-type': 'application/json'}
 276         response = requests.request(
 277             "GET", url, headers=headers, auth=('admin', 'admin'))
 278         self.assertEqual(response.status_code, requests.codes.ok)
 279         res = response.json()
 280         self.assertIn(
 281             {'supported-interface-capability': ['org-openroadm-port-types:if-100GE'],
 282              'supporting-port': 'C1',
 283              'supporting-circuit-pack-name': '1/0/2-PLUG-CLIENT',
 284              'logical-connection-point': 'XPDR1-CLIENT2', 'port-direction': 'bidirectional',
 285              'connection-map-lcp': 'XPDR1-NETWORK2', 'port-qual': 'xpdr-client',
 286              'lcp-hash-val': '3ed8ed1336784ac7c2f66c22f2f03db'},
 287             res['mapping'])
 288
 289     def test_15_xpdr_device_disconnected(self):
 290         url = ("{}/config/network-topology:"
 291                "network-topology/topology/topology-netconf/node/XPDR-A1"
 292                .format(self.restconf_baseurl))
 293         headers = {'content-type': 'application/json'}
 294         response = requests.request(
 295             "DELETE", url, headers=headers,
 296             auth=('admin', 'admin'))
 297         self.assertEqual(response.status_code, requests.codes.ok)
 298         time.sleep(20)
 299
 300     def test_16_xpdr_device_disconnected(self):
 301         url = ("{}/operational/network-topology:network-topology/topology/"
 302                "topology-netconf/node/XPDR-A1".format(self.restconf_baseurl))
 303         headers = {'content-type': 'application/json'}
 304         response = requests.request(
 305             "GET", url, headers=headers, auth=('admin', 'admin'))
 306         self.assertEqual(response.status_code, requests.codes.not_found)
 307         res = response.json()
 308         self.assertIn(
 309             {"error-type": "application", "error-tag": "data-missing",
 310              "error-message": "Request could not be completed because the relevant data model content does not exist"},
 311             res['errors']['error'])
 312
 313     def test_17_xpdr_device_disconnected(self):
 314         url = ("{}/config/transportpce-portmapping:network/nodes/XPDR-A1".format(self.restconf_baseurl))
 315         headers = {'content-type': 'application/json'}
 316         response = requests.request(
 317             "GET", url, headers=headers, auth=('admin', 'admin'))
 318         self.assertEqual(response.status_code, requests.codes.not_found)
 319         res = response.json()
 320         self.assertIn(
 321             {"error-type": "application", "error-tag": "data-missing",
 322              "error-message": "Request could not be completed because the relevant data model content does not exist"},
 323             res['errors']['error'])
 324
 325     def test_18_rdm_device_disconnected(self):
 326         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/node/ROADM-A1"
 327                .format(self.restconf_baseurl))
 328         headers = {'content-type': 'application/json'}
 329         response = requests.request(
 330             "DELETE", url, headers=headers,
 331             auth=('admin', 'admin'))
 332         self.assertEqual(response.status_code, requests.codes.ok)
 333         time.sleep(20)
 334
 335     def test_19_rdm_device_disconnected(self):
 336         url = ("{}/operational/network-topology:network-topology/topology/topology-netconf/node/ROADM-A1"
 337                .format(self.restconf_baseurl))
 338         headers = {'content-type': 'application/json'}
 339         response = requests.request(
 340             "GET", url, headers=headers, auth=('admin', 'admin'))
 341         self.assertEqual(response.status_code, requests.codes.not_found)
 342         res = response.json()
 343         self.assertIn(
 344             {"error-type": "application", "error-tag": "data-missing",
 345              "error-message": "Request could not be completed because the relevant data model content does not exist"},
 346             res['errors']['error'])
 347
 348     def test_20_rdm_device_disconnected(self):
 349         url = ("{}/config/transportpce-portmapping:network/nodes/ROADM-A1".format(self.restconf_baseurl))
 350         headers = {'content-type': 'application/json'}
 351         response = requests.request(
 352             "GET", url, headers=headers, auth=('admin', 'admin'))
 353         self.assertEqual(response.status_code, requests.codes.not_found)
 354         res = response.json()
 355         self.assertIn(
 356             {"error-type": "application", "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
 361 if __name__ == "__main__":
 362     unittest.main(verbosity=2)