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_DEG1_TTP_TXRX(self):
 123         url = ("{}/config/transportpce-portmapping:network/"
 124                "nodes/ROADM-A1/mapping/DEG1-TTP-TXRX"
 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.assertIn(
 132             {'supporting-port': 'L1', 'supporting-circuit-pack-name': '1/0',
 133              'logical-connection-point': 'DEG1-TTP-TXRX'},
 134             res['mapping'])
 135
 136     def test_04_rdm_portmapping_DEG2_TTP_TXRX_with_ots_oms(self):
 137         url = ("{}/config/transportpce-portmapping:network/"
 138                "nodes/ROADM-A1/mapping/DEG2-TTP-TXRX"
 139                .format(self.restconf_baseurl))
 140         headers = {'content-type': 'application/json'}
 141         response = requests.request(
 142             "GET", url, headers=headers, auth=('admin', 'admin'))
 143         self.assertEqual(response.status_code, requests.codes.ok)
 144         res = response.json()
 145         self.assertIn(
 146             {'supporting-port': 'L1', 'supporting-circuit-pack-name': '2/0',
 147              'logical-connection-point': 'DEG2-TTP-TXRX',
 148              'supporting-oms': 'OMS-DEG2-TTP-TXRX', 'supporting-ots': 'OTS-DEG2-TTP-TXRX'},
 149             res['mapping'])
 150
 151     def test_05_rdm_portmapping_SRG1_PP3_TXRX(self):
 152         url = ("{}/config/transportpce-portmapping:network/"
 153                "nodes/ROADM-A1/mapping/SRG1-PP3-TXRX"
 154                .format(self.restconf_baseurl))
 155         headers = {'content-type': 'application/json'}
 156         response = requests.request(
 157             "GET", url, headers=headers, auth=('admin', 'admin'))
 158         self.assertEqual(response.status_code, requests.codes.ok)
 159         res = response.json()
 160         self.assertIn(
 161             {'supporting-port': 'C3', 'supporting-circuit-pack-name': '3/0',
 162              'logical-connection-point': 'SRG1-PP3-TXRX'},
 163             res['mapping'])
 164
 165     def test_06_rdm_portmapping_SRG3_PP1_TXRX(self):
 166         url = ("{}/config/transportpce-portmapping:network/"
 167                "nodes/ROADM-A1/mapping/SRG3-PP1-TXRX"
 168                .format(self.restconf_baseurl))
 169         headers = {'content-type': 'application/json'}
 170         response = requests.request(
 171             "GET", url, headers=headers, auth=('admin', 'admin'))
 172         self.assertEqual(response.status_code, requests.codes.ok)
 173         res = response.json()
 174         self.assertIn(
 175             {'supporting-port': 'C1', 'supporting-circuit-pack-name': '5/0',
 176              'logical-connection-point': 'SRG3-PP1-TXRX'},
 177             res['mapping'])
 178
 179     def test_07_xpdr_device_connected(self):
 180         url = ("{}/config/network-topology:"
 181                "network-topology/topology/topology-netconf/node/XPDR-A1"
 182               .format(self.restconf_baseurl))
 183         data = {"node": [{
 184             "node-id": "XPDR-A1",
 185             "netconf-node-topology:username": "admin",
 186             "netconf-node-topology:password": "admin",
 187             "netconf-node-topology:host": "127.0.0.1",
 188             "netconf-node-topology:port": "17840",
 189             "netconf-node-topology:tcp-only": "false",
 190             "netconf-node-topology:pass-through": {}}]}
 191         headers = {'content-type': 'application/json'}
 192         response = requests.request(
 193             "PUT", url, data=json.dumps(data), headers=headers,
 194             auth=('admin', 'admin'))
 195         self.assertEqual(response.status_code, requests.codes.created)
 196         time.sleep(20)
 197
 198     def test_08_xpdr_device_connected(self):
 199         url = ("{}/operational/network-topology:"
 200                "network-topology/topology/topology-netconf/node/XPDR-A1"
 201                .format(self.restconf_baseurl))
 202         headers = {'content-type': 'application/json'}
 203         response = requests.request(
 204             "GET", url, headers=headers, auth=('admin', 'admin'))
 205         self.assertEqual(response.status_code, requests.codes.ok)
 206         res = response.json()
 207         self.assertEqual(
 208             res['node'][0]['netconf-node-topology:connection-status'],
 209             'connected')
 210         time.sleep(10)
 211
 212     def test_09_xpdr_portmapping_NETWORK1(self):
 213         url = ("{}/config/transportpce-portmapping:network/"
 214                "nodes/XPDR-A1/mapping/XPDR1-NETWORK1"
 215                .format(self.restconf_baseurl))
 216         headers = {'content-type': 'application/json'}
 217         response = requests.request(
 218             "GET", url, headers=headers, auth=('admin', 'admin'))
 219         self.assertEqual(response.status_code, requests.codes.ok)
 220         res = response.json()
 221         self.assertIn(
 222             {'supporting-port': '1', 'supporting-circuit-pack-name': '1/0/1-PLUG-NET',
 223              'logical-connection-point': 'XPDR1-NETWORK1'},
 224             res['mapping'])
 225
 226     def test_10_xpdr_portmapping_NETWORK2(self):
 227         url = ("{}/config/transportpce-portmapping:network/"
 228                "nodes/XPDR-A1/mapping/XPDR1-NETWORK2"
 229                .format(self.restconf_baseurl))
 230         headers = {'content-type': 'application/json'}
 231         response = requests.request(
 232             "GET", url, headers=headers, auth=('admin', 'admin'))
 233         self.assertEqual(response.status_code, requests.codes.ok)
 234         res = response.json()
 235         self.assertIn(
 236             {'supporting-port': '1', 'supporting-circuit-pack-name': '1/0/2-PLUG-NET',
 237              'logical-connection-point': 'XPDR1-NETWORK2'},
 238             res['mapping'])
 239
 240     def test_11_xpdr_portmapping_CLIENT1(self):
 241         url = ("{}/config/transportpce-portmapping:network/"
 242                "nodes/XPDR-A1/mapping/XPDR1-CLIENT1"
 243                .format(self.restconf_baseurl))
 244         headers = {'content-type': 'application/json'}
 245         response = requests.request(
 246             "GET", url, headers=headers, auth=('admin', 'admin'))
 247         self.assertEqual(response.status_code, requests.codes.ok)
 248         res = response.json()
 249         self.assertIn(
 250             {'supporting-port': 'C1',
 251              'supporting-circuit-pack-name': '1/0/1-PLUG-CLIENT',
 252              'logical-connection-point': 'XPDR1-CLIENT1'},
 253             res['mapping'])
 254
 255     def test_12_xpdr_portmapping_CLIENT2(self):
 256         url = ("{}/config/transportpce-portmapping:network/"
 257                "nodes/XPDR-A1/mapping/XPDR1-CLIENT2"
 258                .format(self.restconf_baseurl))
 259         headers = {'content-type': 'application/json'}
 260         response = requests.request(
 261             "GET", url, headers=headers, auth=('admin', 'admin'))
 262         self.assertEqual(response.status_code, requests.codes.ok)
 263         res = response.json()
 264         self.assertIn(
 265             {'supporting-port': 'C1',
 266                  'supporting-circuit-pack-name': '1/0/2-PLUG-CLIENT',
 267                  'logical-connection-point': 'XPDR1-CLIENT2'},
 268             res['mapping'])
 269
 270     def test_13_xpdr_device_disconnected(self):
 271         url = ("{}/config/network-topology:"
 272                 "network-topology/topology/topology-netconf/node/XPDR-A1"
 273                .format(self.restconf_baseurl))
 274         headers = {'content-type': 'application/json'}
 275         response = requests.request(
 276              "DELETE", url, headers=headers,
 277              auth=('admin', 'admin'))
 278         self.assertEqual(response.status_code, requests.codes.ok)
 279         time.sleep(20)
 280
 281     def test_14_xpdr_device_disconnected(self):
 282         url = ("{}/operational/network-topology:network-topology/topology/"
 283                "topology-netconf/node/XPDR-A1".format(self.restconf_baseurl))
 284         headers = {'content-type': 'application/json'}
 285         response = requests.request(
 286             "GET", url, headers=headers, auth=('admin', 'admin'))
 287         self.assertEqual(response.status_code, requests.codes.not_found)
 288         res = response.json()
 289         self.assertIn(
 290             {"error-type":"application", "error-tag":"data-missing",
 291              "error-message":"Request could not be completed because the relevant data model content does not exist"},
 292             res['errors']['error'])
 293
 294     def test_15_xpdr_device_disconnected(self):
 295         url = ("{}/config/transportpce-portmapping:network/nodes/XPDR-A1".format(self.restconf_baseurl))
 296         headers = {'content-type': 'application/json'}
 297         response = requests.request(
 298             "GET", url, headers=headers, auth=('admin', 'admin'))
 299         self.assertEqual(response.status_code, requests.codes.not_found)
 300         res = response.json()
 301         self.assertIn(
 302             {"error-type":"application", "error-tag":"data-missing",
 303              "error-message":"Request could not be completed because the relevant data model content does not exist"},
 304             res['errors']['error'])
 305
 306     def test_16_rdm_device_disconnected(self):
 307         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/node/ROADM-A1"
 308                .format(self.restconf_baseurl))
 309         headers = {'content-type': 'application/json'}
 310         response = requests.request(
 311              "DELETE", url, headers=headers,
 312              auth=('admin', 'admin'))
 313         self.assertEqual(response.status_code, requests.codes.ok)
 314         time.sleep(20)
 315
 316     def test_17_rdm_device_disconnected(self):
 317         url = ("{}/operational/network-topology:network-topology/topology/topology-netconf/node/ROADM-A1"
 318                .format(self.restconf_baseurl))
 319         headers = {'content-type': 'application/json'}
 320         response = requests.request(
 321             "GET", url, headers=headers, auth=('admin', 'admin'))
 322         self.assertEqual(response.status_code, requests.codes.not_found)
 323         res = response.json()
 324         self.assertIn(
 325             {"error-type":"application", "error-tag":"data-missing",
 326              "error-message":"Request could not be completed because the relevant data model content does not exist"},
 327             res['errors']['error'])
 328
 329     def test_18_rdm_device_disconnected(self):
 330         url = ("{}/config/transportpce-portmapping:network/nodes/ROADM-A1".format(self.restconf_baseurl))
 331         headers = {'content-type': 'application/json'}
 332         response = requests.request(
 333             "GET", url, headers=headers, auth=('admin', 'admin'))
 334         self.assertEqual(response.status_code, requests.codes.not_found)
 335         res = response.json()
 336         self.assertIn(
 337             {"error-type":"application", "error-tag":"data-missing",
 338              "error-message":"Request could not be completed because the relevant data model content does not exist"},
 339             res['errors']['error'])
 340
 341
 342 if __name__ == "__main__":
 343     unittest.main(verbosity=2)