tests/transportpce_tests/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     @classmethod
  31     def __start_honeynode1(cls):
  32         executable = ("./honeynode/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, "17831", "sample_configs/ord_2.1/oper-XPDRA.xml"],
  38                     stdout=outfile)
  39
  40     @classmethod
  41     def __start_honeynode2(cls):
  42         executable = ("./honeynode/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, "17830", "sample_configs/ord_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(40)
  62         cls.__start_honeynode2()
  63         time.sleep(40)
  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_restconfAPI(self):
  90 #        url = ("{}/operational/network-topology:network-topology/topology/"
  91 #        "topology-netconf/node/controller-config".format(self.restconf_baseurl))
  92 #        headers = {'content-type': 'application/json'}
  93 #        response = requests.request("GET", url, headers=headers, auth=('admin', 'admin'))
  94 #        self.assertEqual(response.status_code, requests.codes.ok)
  95 #        res = response.json()
  96 #        self.assertEqual(res['node'] [0] ['netconf-node-topology:connection-status'],
  97 #                         'connected')
  98
  99     def test_02_restconfAPI(self):
 100         url = ("{}/config/portmapping:network/nodes/controller-config".format(self.restconf_baseurl))
 101         headers = {'content-type': 'application/json'}
 102         response = requests.request(
 103             "GET", url, headers=headers, auth=('admin', 'admin'))
 104         self.assertEqual(response.status_code, requests.codes.not_found)
 105         res = response.json()
 106         self.assertIn(
 107             {"error-type":"application", "error-tag":"data-missing",
 108              "error-message":"Request could not be completed because the relevant data model content does not exist "},
 109             res['errors']['error'])
 110
 111     def test_03_rdm_device_connected(self):
 112         url = ("{}/config/network-topology:"
 113                "network-topology/topology/topology-netconf/node/ROADMA"
 114               .format(self.restconf_baseurl))
 115         data = {"node": [{
 116             "node-id": "ROADMA",
 117             "netconf-node-topology:username": "admin",
 118             "netconf-node-topology:password": "admin",
 119             "netconf-node-topology:host": "127.0.0.1",
 120             "netconf-node-topology:port": "17830",
 121             "netconf-node-topology:tcp-only": "false",
 122             "netconf-node-topology:pass-through": {}}]}
 123         headers = {'content-type': 'application/json'}
 124         response = requests.request(
 125             "PUT", url, data=json.dumps(data), headers=headers,
 126             auth=('admin', 'admin'))
 127         self.assertEqual(response.status_code, requests.codes.created)
 128         time.sleep(20)
 129
 130     def test_04_rdm_device_connected(self):
 131         url = ("{}/operational/network-topology:"
 132                "network-topology/topology/topology-netconf/node/ROADMA"
 133                .format(self.restconf_baseurl))
 134         headers = {'content-type': 'application/json'}
 135         response = requests.request(
 136             "GET", url, headers=headers, auth=('admin', 'admin'))
 137         self.assertEqual(response.status_code, requests.codes.ok)
 138         res = response.json()
 139         self.assertEqual(
 140             res['node'][0]['netconf-node-topology:connection-status'],
 141             'connected')
 142         time.sleep(10)
 143
 144     def test_05_rdm_portmapping_DEG1_TTP_TXRX(self):
 145         url = ("{}/config/portmapping:network/"
 146                "nodes/ROADMA/mapping/DEG1-TTP-TXRX"
 147                .format(self.restconf_baseurl))
 148         headers = {'content-type': 'application/json'}
 149         response = requests.request(
 150             "GET", url, headers=headers, auth=('admin', 'admin'))
 151         self.assertEqual(response.status_code, requests.codes.ok)
 152         res = response.json()
 153         self.assertIn(
 154             {'supporting-port': 'L1', 'supporting-circuit-pack-name': '2/0',
 155              'logical-connection-point': 'DEG1-TTP-TXRX'},
 156             res['mapping'])
 157
 158     def test_06_rdm_portmapping_SRG1_PP7_TXRX(self):
 159         url = ("{}/config/portmapping:network/"
 160                "nodes/ROADMA/mapping/SRG1-PP7-TXRX"
 161                .format(self.restconf_baseurl))
 162         headers = {'content-type': 'application/json'}
 163         response = requests.request(
 164             "GET", url, headers=headers, auth=('admin', 'admin'))
 165         self.assertEqual(response.status_code, requests.codes.ok)
 166         res = response.json()
 167         self.assertIn(
 168             {'supporting-port': 'C7', 'supporting-circuit-pack-name': '4/0',
 169              'logical-connection-point': 'SRG1-PP7-TXRX'},
 170             res['mapping'])
 171
 172     def test_07_rdm_portmapping_SRG3_PP1_TXRX(self):
 173         url = ("{}/config/portmapping:network/"
 174                "nodes/ROADMA/mapping/SRG3-PP1-TXRX"
 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.assertIn(
 182             {'supporting-port': 'C1', 'supporting-circuit-pack-name': '5/0',
 183              'logical-connection-point': 'SRG3-PP1-TXRX'},
 184             res['mapping'])
 185
 186     def test_08_xpdr_device_connected(self):
 187         url = ("{}/config/network-topology:"
 188                "network-topology/topology/topology-netconf/node/XPDRA"
 189               .format(self.restconf_baseurl))
 190         data = {"node": [{
 191             "node-id": "XPDRA",
 192             "netconf-node-topology:username": "admin",
 193             "netconf-node-topology:password": "admin",
 194             "netconf-node-topology:host": "127.0.0.1",
 195             "netconf-node-topology:port": "17831",
 196             "netconf-node-topology:tcp-only": "false",
 197             "netconf-node-topology:pass-through": {}}]}
 198         headers = {'content-type': 'application/json'}
 199         response = requests.request(
 200             "PUT", url, data=json.dumps(data), headers=headers,
 201             auth=('admin', 'admin'))
 202         self.assertEqual(response.status_code, requests.codes.created)
 203         time.sleep(20)
 204
 205     def test_09_xpdr_device_connected(self):
 206         url = ("{}/operational/network-topology:"
 207                "network-topology/topology/topology-netconf/node/XPDRA"
 208                .format(self.restconf_baseurl))
 209         headers = {'content-type': 'application/json'}
 210         response = requests.request(
 211             "GET", url, headers=headers, auth=('admin', 'admin'))
 212         self.assertEqual(response.status_code, requests.codes.ok)
 213         res = response.json()
 214         self.assertEqual(
 215             res['node'][0]['netconf-node-topology:connection-status'],
 216             'connected')
 217         time.sleep(10)
 218
 219     def test_10_xpdr_portmapping_NETWORK1(self):
 220         url = ("{}/config/portmapping:network/"
 221                "nodes/XPDRA/mapping/XPDR1-NETWORK1"
 222                .format(self.restconf_baseurl))
 223         headers = {'content-type': 'application/json'}
 224         response = requests.request(
 225             "GET", url, headers=headers, auth=('admin', 'admin'))
 226         self.assertEqual(response.status_code, requests.codes.ok)
 227         res = response.json()
 228         self.assertIn(
 229             {'supporting-port': '1', 'supporting-circuit-pack-name': '1/0/1-PLUG-NET',
 230              'logical-connection-point': 'XPDR1-NETWORK1'},
 231             res['mapping'])
 232
 233     def test_11_xpdr_portmapping_NETWORK2(self):
 234         url = ("{}/config/portmapping:network/"
 235                "nodes/XPDRA/mapping/XPDR0-NETWORK2"
 236                .format(self.restconf_baseurl))
 237         headers = {'content-type': 'application/json'}
 238         response = requests.request(
 239             "GET", url, headers=headers, auth=('admin', 'admin'))
 240         self.assertEqual(response.status_code, requests.codes.ok)
 241         res = response.json()
 242         self.assertIn(
 243             {'supporting-port': '2', 'supporting-circuit-pack-name': '1/0/2-PLUG-NET',
 244              'logical-connection-point': 'XPDR0-NETWORK2'},
 245             res['mapping'])
 246
 247     def test_12_xpdr_portmapping_CLIENT1(self):
 248         url = ("{}/config/portmapping:network/"
 249                "nodes/XPDRA/mapping/XPDR1-CLIENT1"
 250                .format(self.restconf_baseurl))
 251         headers = {'content-type': 'application/json'}
 252         response = requests.request(
 253             "GET", url, headers=headers, auth=('admin', 'admin'))
 254         self.assertEqual(response.status_code, requests.codes.ok)
 255         res = response.json()
 256         self.assertIn(
 257             {'supporting-port': 'C1',
 258              'supporting-circuit-pack-name': '1/0/C1-PLUG-CLIENT',
 259              'logical-connection-point': 'XPDR1-CLIENT1'},
 260             res['mapping'])
 261
 262     def test_13_xpdr_portmapping_CLIENT2(self):
 263         url = ("{}/config/portmapping:network/"
 264                "nodes/XPDRA/mapping/XPDR1-CLIENT2"
 265                .format(self.restconf_baseurl))
 266         headers = {'content-type': 'application/json'}
 267         response = requests.request(
 268             "GET", url, headers=headers, auth=('admin', 'admin'))
 269         self.assertEqual(response.status_code, requests.codes.not_found)
 270         res = response.json()
 271         self.assertIn(
 272             {"error-type":"application","error-tag":"data-missing",
 273              "error-message":"Request could not be completed because the relevant data model content does not exist "},
 274             res['errors']['error'])
 275
 276     def test_14_xpdr_portmapping_CLIENT4(self):
 277         url = ("{}/config/portmapping:network/"
 278                "nodes/XPDRA/mapping/XPDR0-CLIENT4"
 279                .format(self.restconf_baseurl))
 280         headers = {'content-type': 'application/json'}
 281         response = requests.request(
 282             "GET", url, headers=headers, auth=('admin', 'admin'))
 283         self.assertEqual(response.status_code, requests.codes.ok)
 284         res = response.json()
 285         self.assertIn(
 286             {'supporting-port': 'C4',
 287              'supporting-circuit-pack-name': '1/0/C4-PLUG-CLIENT',
 288              'logical-connection-point': 'XPDR0-CLIENT4'},
 289             res['mapping'])
 290
 291     def test_15_xpdr_device_disconnected(self):
 292         url = ("{}/config/network-topology:"
 293                 "network-topology/topology/topology-netconf/node/XPDRA"
 294                .format(self.restconf_baseurl))
 295         headers = {'content-type': 'application/json'}
 296         response = requests.request(
 297              "DELETE", url, headers=headers,
 298              auth=('admin', 'admin'))
 299         self.assertEqual(response.status_code, requests.codes.ok)
 300         time.sleep(20)
 301
 302     def test_16_xpdr_device_disconnected(self):
 303         url = ("{}/operational/network-topology:network-topology/topology/"
 304                "topology-netconf/node/XPDRA".format(self.restconf_baseurl))
 305         headers = {'content-type': 'application/json'}
 306         response = requests.request(
 307             "GET", url, headers=headers, auth=('admin', 'admin'))
 308         self.assertEqual(response.status_code, requests.codes.not_found)
 309         res = response.json()
 310         self.assertIn(
 311             {"error-type":"application", "error-tag":"data-missing",
 312              "error-message":"Request could not be completed because the relevant data model content does not exist "},
 313             res['errors']['error'])
 314
 315     def test_17_xpdr_device_disconnected(self):
 316         url = ("{}/config/portmapping:network/nodes/XPDRA".format(self.restconf_baseurl))
 317         headers = {'content-type': 'application/json'}
 318         response = requests.request(
 319             "GET", url, headers=headers, auth=('admin', 'admin'))
 320         self.assertEqual(response.status_code, requests.codes.not_found)
 321         res = response.json()
 322         self.assertIn(
 323             {"error-type":"application", "error-tag":"data-missing",
 324              "error-message":"Request could not be completed because the relevant data model content does not exist "},
 325             res['errors']['error'])
 326
 327     def test_18_rdm_device_disconnected(self):
 328         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/node/ROADMA"
 329                .format(self.restconf_baseurl))
 330         headers = {'content-type': 'application/json'}
 331         response = requests.request(
 332              "DELETE", url, headers=headers,
 333              auth=('admin', 'admin'))
 334         self.assertEqual(response.status_code, requests.codes.ok)
 335         time.sleep(20)
 336
 337     def test_19_rdm_device_disconnected(self):
 338         url = ("{}/operational/network-topology:network-topology/topology/topology-netconf/node/ROADMA"
 339                .format(self.restconf_baseurl))
 340         headers = {'content-type': 'application/json'}
 341         response = requests.request(
 342             "GET", url, headers=headers, auth=('admin', 'admin'))
 343         self.assertEqual(response.status_code, requests.codes.not_found)
 344         res = response.json()
 345         self.assertIn(
 346             {"error-type":"application", "error-tag":"data-missing",
 347              "error-message":"Request could not be completed because the relevant data model content does not exist "},
 348             res['errors']['error'])
 349
 350     def test_20_rdm_device_disconnected(self):
 351         url = ("{}/config/portmapping:network/nodes/ROADMA".format(self.restconf_baseurl))
 352         headers = {'content-type': 'application/json'}
 353         response = requests.request(
 354             "GET", url, headers=headers, auth=('admin', 'admin'))
 355         self.assertEqual(response.status_code, requests.codes.not_found)
 356         res = response.json()
 357         self.assertIn(
 358             {"error-type":"application", "error-tag":"data-missing",
 359              "error-message":"Request could not be completed because the relevant data model content does not exist "},
 360             res['errors']['error'])
 361
 362
 363 if __name__ == "__main__":
 364     unittest.main(verbosity=2)