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