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