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