tests/transportpce_tests/1.2.1/test_portmapping.py

   1 #!/usr/bin/env python
   2
   3 ##############################################################################
   4 # Copyright (c) 2017 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 #START_IGNORE_XTESTING
  31
  32     @classmethod
  33     def __start_honeynode1(cls):
  34         executable = ("./honeynode/2.1/honeynode-distribution/target/honeynode-distribution-1.18.01-hc"
  35                       "/honeynode-distribution-1.18.01/honeycomb-tpce")
  36         if os.path.isfile(executable):
  37             with open('honeynode1.log', 'w') as outfile:
  38                 cls.honeynode_process1 = subprocess.Popen(
  39                     [executable, "17831", "sample_configs/openroadm/2.1/oper-XPDRA.xml"],
  40                     stdout=outfile)
  41
  42     @classmethod
  43     def __start_honeynode2(cls):
  44         executable = ("./honeynode/2.1/honeynode-distribution/target/honeynode-distribution-1.18.01-hc"
  45                       "/honeynode-distribution-1.18.01/honeycomb-tpce")
  46         if os.path.isfile(executable):
  47             with open('honeynode2.log', 'w') as outfile:
  48                 cls.honeynode_process2 = subprocess.Popen(
  49                     [executable, "17830", "sample_configs/openroadm/2.1/oper-ROADMA.xml"],
  50                     stdout=outfile)
  51
  52     @classmethod
  53     def __start_odl(cls):
  54         executable = "../karaf/target/assembly/bin/karaf"
  55         with open('odl.log', 'w') as outfile:
  56             cls.odl_process = subprocess.Popen(
  57                 ["bash", executable, "server"], stdout=outfile,
  58                 stdin=open(os.devnull))
  59
  60     @classmethod
  61     def setUpClass(cls):
  62         cls.__start_honeynode1()
  63         time.sleep(20)
  64         cls.__start_honeynode2()
  65         time.sleep(20)
  66         cls.__start_odl()
  67         time.sleep(60)
  68
  69     @classmethod
  70     def tearDownClass(cls):
  71         for child in psutil.Process(cls.odl_process.pid).children():
  72             child.send_signal(signal.SIGINT)
  73             child.wait()
  74         cls.odl_process.send_signal(signal.SIGINT)
  75         cls.odl_process.wait()
  76         for child in psutil.Process(cls.honeynode_process1.pid).children():
  77             child.send_signal(signal.SIGINT)
  78             child.wait()
  79         cls.honeynode_process1.send_signal(signal.SIGINT)
  80         cls.honeynode_process1.wait()
  81         for child in psutil.Process(cls.honeynode_process2.pid).children():
  82             child.send_signal(signal.SIGINT)
  83             child.wait()
  84         cls.honeynode_process2.send_signal(signal.SIGINT)
  85         cls.honeynode_process2.wait()
  86
  87     def setUp(self):
  88         print ("execution of {}".format(self.id().split(".")[-1]))
  89         time.sleep(10)
  90
  91 #END_IGNORE_XTESTING
  92
  93 #    def test_01_restconfAPI(self):
  94 #        url = ("{}/operational/network-topology:network-topology/topology/"
  95 #        "topology-netconf/node/controller-config".format(self.restconf_baseurl))
  96 #        headers = {'content-type': 'application/json'}
  97 #        response = requests.request("GET", url, headers=headers, auth=('admin', 'admin'))
  98 #        self.assertEqual(response.status_code, requests.codes.ok)
  99 #        res = response.json()
 100 #        self.assertEqual(res['node'] [0] ['netconf-node-topology:connection-status'],
 101 #                         'connected')
 102
 103 #     def test_02_restconfAPI(self):
 104 #         url = ("{}/config/transportpce-portmapping:network/nodes/controller-config".format(self.restconf_baseurl))
 105 #         headers = {'content-type': 'application/json'}
 106 #         response = requests.request(
 107 #             "GET", url, headers=headers, auth=('admin', 'admin'))
 108 #         self.assertEqual(response.status_code, requests.codes.not_found)
 109 #         res = response.json()
 110 #         self.assertIn(
 111 #             {"error-type":"application", "error-tag":"data-missing",
 112 #              "error-message":"Request could not be completed because the relevant data model content does not exist "},
 113 #             res['errors']['error'])
 114
 115     def test_01_rdm_device_connected(self):
 116         url = ("{}/config/network-topology:"
 117                "network-topology/topology/topology-netconf/node/ROADMA"
 118               .format(self.restconf_baseurl))
 119         data = {"node": [{
 120             "node-id": "ROADMA",
 121             "netconf-node-topology:username": "admin",
 122             "netconf-node-topology:password": "admin",
 123             "netconf-node-topology:host": "127.0.0.1",
 124             "netconf-node-topology:port": "17830",
 125             "netconf-node-topology:tcp-only": "false",
 126             "netconf-node-topology:pass-through": {}}]}
 127         headers = {'content-type': 'application/json'}
 128         response = requests.request(
 129             "PUT", url, data=json.dumps(data), headers=headers,
 130             auth=('admin', 'admin'))
 131         self.assertEqual(response.status_code, requests.codes.created)
 132         time.sleep(20)
 133
 134     def test_02_rdm_device_connected(self):
 135         url = ("{}/operational/network-topology:"
 136                "network-topology/topology/topology-netconf/node/ROADMA"
 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.assertEqual(
 144             res['node'][0]['netconf-node-topology:connection-status'],
 145             'connected')
 146         time.sleep(10)
 147
 148     def test_03_rdm_portmapping_DEG1_TTP_TXRX(self):
 149         url = ("{}/config/transportpce-portmapping:network/"
 150                "nodes/ROADMA/mapping/DEG1-TTP-TXRX"
 151                .format(self.restconf_baseurl))
 152         headers = {'content-type': 'application/json'}
 153         response = requests.request(
 154             "GET", url, headers=headers, auth=('admin', 'admin'))
 155         self.assertEqual(response.status_code, requests.codes.ok)
 156         res = response.json()
 157         self.assertIn(
 158             {'supporting-port': 'L1', 'supporting-circuit-pack-name': '2/0',
 159              'logical-connection-point': 'DEG1-TTP-TXRX'},
 160             res['mapping'])
 161
 162     def test_04_rdm_portmapping_SRG1_PP7_TXRX(self):
 163         url = ("{}/config/transportpce-portmapping:network/"
 164                "nodes/ROADMA/mapping/SRG1-PP7-TXRX"
 165                .format(self.restconf_baseurl))
 166         headers = {'content-type': 'application/json'}
 167         response = requests.request(
 168             "GET", url, headers=headers, auth=('admin', 'admin'))
 169         self.assertEqual(response.status_code, requests.codes.ok)
 170         res = response.json()
 171         self.assertIn(
 172             {'supporting-port': 'C7', 'supporting-circuit-pack-name': '4/0',
 173              'logical-connection-point': 'SRG1-PP7-TXRX'},
 174             res['mapping'])
 175
 176     def test_05_rdm_portmapping_SRG3_PP1_TXRX(self):
 177         url = ("{}/config/transportpce-portmapping:network/"
 178                "nodes/ROADMA/mapping/SRG3-PP1-TXRX"
 179                .format(self.restconf_baseurl))
 180         headers = {'content-type': 'application/json'}
 181         response = requests.request(
 182             "GET", url, headers=headers, auth=('admin', 'admin'))
 183         self.assertEqual(response.status_code, requests.codes.ok)
 184         res = response.json()
 185         self.assertIn(
 186             {'supporting-port': 'C1', 'supporting-circuit-pack-name': '5/0',
 187              'logical-connection-point': 'SRG3-PP1-TXRX'},
 188             res['mapping'])
 189
 190     def test_06_xpdr_device_connected(self):
 191         url = ("{}/config/network-topology:"
 192                "network-topology/topology/topology-netconf/node/XPDRA"
 193               .format(self.restconf_baseurl))
 194         data = {"node": [{
 195             "node-id": "XPDRA",
 196             "netconf-node-topology:username": "admin",
 197             "netconf-node-topology:password": "admin",
 198             "netconf-node-topology:host": "127.0.0.1",
 199             "netconf-node-topology:port": "17831",
 200             "netconf-node-topology:tcp-only": "false",
 201             "netconf-node-topology:pass-through": {}}]}
 202         headers = {'content-type': 'application/json'}
 203         response = requests.request(
 204             "PUT", url, data=json.dumps(data), headers=headers,
 205             auth=('admin', 'admin'))
 206         self.assertEqual(response.status_code, requests.codes.created)
 207         time.sleep(20)
 208
 209     def test_07_xpdr_device_connected(self):
 210         url = ("{}/operational/network-topology:"
 211                "network-topology/topology/topology-netconf/node/XPDRA"
 212                .format(self.restconf_baseurl))
 213         headers = {'content-type': 'application/json'}
 214         response = requests.request(
 215             "GET", url, headers=headers, auth=('admin', 'admin'))
 216         self.assertEqual(response.status_code, requests.codes.ok)
 217         res = response.json()
 218         self.assertEqual(
 219             res['node'][0]['netconf-node-topology:connection-status'],
 220             'connected')
 221         time.sleep(10)
 222
 223     def test_08_xpdr_portmapping_NETWORK1(self):
 224         url = ("{}/config/transportpce-portmapping:network/"
 225                "nodes/XPDRA/mapping/XPDR1-NETWORK1"
 226                .format(self.restconf_baseurl))
 227         headers = {'content-type': 'application/json'}
 228         response = requests.request(
 229             "GET", url, headers=headers, auth=('admin', 'admin'))
 230         self.assertEqual(response.status_code, requests.codes.ok)
 231         res = response.json()
 232         self.assertIn(
 233             {'supporting-port': '1', 'supporting-circuit-pack-name': '1/0/1-PLUG-NET',
 234              'logical-connection-point': 'XPDR1-NETWORK1'},
 235             res['mapping'])
 236
 237     def test_09_xpdr_portmapping_NETWORK2(self):
 238         url = ("{}/config/transportpce-portmapping:network/"
 239                "nodes/XPDRA/mapping/XPDR1-NETWORK2"
 240                .format(self.restconf_baseurl))
 241         headers = {'content-type': 'application/json'}
 242         response = requests.request(
 243             "GET", url, headers=headers, auth=('admin', 'admin'))
 244         self.assertEqual(response.status_code, requests.codes.ok)
 245         res = response.json()
 246         self.assertIn(
 247             {'supporting-port': '1', 'supporting-circuit-pack-name': '1/0/2-PLUG-NET',
 248              'logical-connection-point': 'XPDR1-NETWORK2'},
 249             res['mapping'])
 250
 251     def test_10_xpdr_portmapping_CLIENT1(self):
 252         url = ("{}/config/transportpce-portmapping:network/"
 253                "nodes/XPDRA/mapping/XPDR1-CLIENT1"
 254                .format(self.restconf_baseurl))
 255         headers = {'content-type': 'application/json'}
 256         response = requests.request(
 257             "GET", url, headers=headers, auth=('admin', 'admin'))
 258         self.assertEqual(response.status_code, requests.codes.ok)
 259         res = response.json()
 260         self.assertIn(
 261             {'supporting-port': 'C1',
 262              'supporting-circuit-pack-name': '1/0/C1-PLUG-CLIENT',
 263              'logical-connection-point': 'XPDR1-CLIENT1'},
 264             res['mapping'])
 265
 266     def test_11_xpdr_portmapping_CLIENT2(self):
 267         url = ("{}/config/transportpce-portmapping:network/"
 268                "nodes/XPDRA/mapping/XPDR1-CLIENT2"
 269                .format(self.restconf_baseurl))
 270         headers = {'content-type': 'application/json'}
 271         response = requests.request(
 272             "GET", url, headers=headers, auth=('admin', 'admin'))
 273         self.assertEqual(response.status_code, requests.codes.ok)
 274         res = response.json()
 275         self.assertIn(
 276             {'supporting-port': 'C2',
 277                  'supporting-circuit-pack-name': '1/0/C2-PLUG-CLIENT',
 278                  'logical-connection-point': 'XPDR1-CLIENT2'},
 279             res['mapping'])
 280
 281     def test_12_xpdr_portmapping_CLIENT4(self):
 282         url = ("{}/config/transportpce-portmapping:network/"
 283                "nodes/XPDRA/mapping/XPDR1-CLIENT4"
 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             {'supporting-port': 'C4',
 292              'supporting-circuit-pack-name': '1/0/C4-PLUG-CLIENT',
 293              'logical-connection-point': 'XPDR1-CLIENT4'},
 294             res['mapping'])
 295
 296     def test_13_xpdr_device_disconnected(self):
 297         url = ("{}/config/network-topology:"
 298                 "network-topology/topology/topology-netconf/node/XPDRA"
 299                .format(self.restconf_baseurl))
 300         headers = {'content-type': 'application/json'}
 301         response = requests.request(
 302              "DELETE", url, headers=headers,
 303              auth=('admin', 'admin'))
 304         self.assertEqual(response.status_code, requests.codes.ok)
 305         time.sleep(20)
 306
 307     def test_14_xpdr_device_disconnected(self):
 308         url = ("{}/operational/network-topology:network-topology/topology/"
 309                "topology-netconf/node/XPDRA".format(self.restconf_baseurl))
 310         headers = {'content-type': 'application/json'}
 311         response = requests.request(
 312             "GET", url, headers=headers, auth=('admin', 'admin'))
 313         self.assertEqual(response.status_code, requests.codes.not_found)
 314         res = response.json()
 315         self.assertIn(
 316             {"error-type":"application", "error-tag":"data-missing",
 317              "error-message":"Request could not be completed because the relevant data model content does not exist"},
 318             res['errors']['error'])
 319
 320     def test_15_xpdr_device_disconnected(self):
 321         url = ("{}/config/transportpce-portmapping:network/nodes/XPDRA".format(self.restconf_baseurl))
 322         headers = {'content-type': 'application/json'}
 323         response = requests.request(
 324             "GET", url, headers=headers, auth=('admin', 'admin'))
 325         self.assertEqual(response.status_code, requests.codes.not_found)
 326         res = response.json()
 327         self.assertIn(
 328             {"error-type":"application", "error-tag":"data-missing",
 329              "error-message":"Request could not be completed because the relevant data model content does not exist"},
 330             res['errors']['error'])
 331
 332     def test_16_rdm_device_disconnected(self):
 333         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/node/ROADMA"
 334                .format(self.restconf_baseurl))
 335         headers = {'content-type': 'application/json'}
 336         response = requests.request(
 337              "DELETE", url, headers=headers,
 338              auth=('admin', 'admin'))
 339         self.assertEqual(response.status_code, requests.codes.ok)
 340         time.sleep(20)
 341
 342     def test_17_rdm_device_disconnected(self):
 343         url = ("{}/operational/network-topology:network-topology/topology/topology-netconf/node/ROADMA"
 344                .format(self.restconf_baseurl))
 345         headers = {'content-type': 'application/json'}
 346         response = requests.request(
 347             "GET", url, headers=headers, auth=('admin', 'admin'))
 348         self.assertEqual(response.status_code, requests.codes.not_found)
 349         res = response.json()
 350         self.assertIn(
 351             {"error-type":"application", "error-tag":"data-missing",
 352              "error-message":"Request could not be completed because the relevant data model content does not exist"},
 353             res['errors']['error'])
 354
 355     def test_18_rdm_device_disconnected(self):
 356         url = ("{}/config/transportpce-portmapping:network/nodes/ROADMA".format(self.restconf_baseurl))
 357         headers = {'content-type': 'application/json'}
 358         response = requests.request(
 359             "GET", url, headers=headers, auth=('admin', 'admin'))
 360         self.assertEqual(response.status_code, requests.codes.not_found)
 361         res = response.json()
 362         self.assertIn(
 363             {"error-type":"application", "error-tag":"data-missing",
 364              "error-message":"Request could not be completed because the relevant data model content does not exist"},
 365             res['errors']['error'])
 366
 367
 368 if __name__ == "__main__":
 369     unittest.main(verbosity=2)