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(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_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/XPDR1-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': 'XPDR1-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.ok)
 270         res = response.json()
 271         self.assertIn(
 272             {'supporting-port': 'C2',
 273                  'supporting-circuit-pack-name': '1/0/C2-PLUG-CLIENT',
 274                  'logical-connection-point': 'XPDR1-CLIENT2'},
 275             res['mapping'])
 276
 277     def test_14_xpdr_portmapping_CLIENT4(self):
 278         url = ("{}/config/portmapping:network/"
 279                "nodes/XPDRA/mapping/XPDR1-CLIENT4"
 280                .format(self.restconf_baseurl))
 281         headers = {'content-type': 'application/json'}
 282         response = requests.request(
 283             "GET", url, headers=headers, auth=('admin', 'admin'))
 284         self.assertEqual(response.status_code, requests.codes.ok)
 285         res = response.json()
 286         self.assertIn(
 287             {'supporting-port': 'C4',
 288              'supporting-circuit-pack-name': '1/0/C4-PLUG-CLIENT',
 289              'logical-connection-point': 'XPDR1-CLIENT4'},
 290             res['mapping'])
 291
 292     def test_15_xpdr_device_disconnected(self):
 293         url = ("{}/config/network-topology:"
 294                 "network-topology/topology/topology-netconf/node/XPDRA"
 295                .format(self.restconf_baseurl))
 296         headers = {'content-type': 'application/json'}
 297         response = requests.request(
 298              "DELETE", url, headers=headers,
 299              auth=('admin', 'admin'))
 300         self.assertEqual(response.status_code, requests.codes.ok)
 301         time.sleep(20)
 302
 303     def test_16_xpdr_device_disconnected(self):
 304         url = ("{}/operational/network-topology:network-topology/topology/"
 305                "topology-netconf/node/XPDRA".format(self.restconf_baseurl))
 306         headers = {'content-type': 'application/json'}
 307         response = requests.request(
 308             "GET", url, headers=headers, auth=('admin', 'admin'))
 309         self.assertEqual(response.status_code, requests.codes.not_found)
 310         res = response.json()
 311         self.assertIn(
 312             {"error-type":"application", "error-tag":"data-missing",
 313              "error-message":"Request could not be completed because the relevant data model content does not exist "},
 314             res['errors']['error'])
 315
 316     def test_17_xpdr_device_disconnected(self):
 317         url = ("{}/config/portmapping:network/nodes/XPDRA".format(self.restconf_baseurl))
 318         headers = {'content-type': 'application/json'}
 319         response = requests.request(
 320             "GET", url, headers=headers, auth=('admin', 'admin'))
 321         self.assertEqual(response.status_code, requests.codes.not_found)
 322         res = response.json()
 323         self.assertIn(
 324             {"error-type":"application", "error-tag":"data-missing",
 325              "error-message":"Request could not be completed because the relevant data model content does not exist "},
 326             res['errors']['error'])
 327
 328     def test_18_rdm_device_disconnected(self):
 329         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/node/ROADMA"
 330                .format(self.restconf_baseurl))
 331         headers = {'content-type': 'application/json'}
 332         response = requests.request(
 333              "DELETE", url, headers=headers,
 334              auth=('admin', 'admin'))
 335         self.assertEqual(response.status_code, requests.codes.ok)
 336         time.sleep(20)
 337
 338     def test_19_rdm_device_disconnected(self):
 339         url = ("{}/operational/network-topology:network-topology/topology/topology-netconf/node/ROADMA"
 340                .format(self.restconf_baseurl))
 341         headers = {'content-type': 'application/json'}
 342         response = requests.request(
 343             "GET", url, headers=headers, auth=('admin', 'admin'))
 344         self.assertEqual(response.status_code, requests.codes.not_found)
 345         res = response.json()
 346         self.assertIn(
 347             {"error-type":"application", "error-tag":"data-missing",
 348              "error-message":"Request could not be completed because the relevant data model content does not exist "},
 349             res['errors']['error'])
 350
 351     def test_20_rdm_device_disconnected(self):
 352         url = ("{}/config/portmapping:network/nodes/ROADMA".format(self.restconf_baseurl))
 353         headers = {'content-type': 'application/json'}
 354         response = requests.request(
 355             "GET", url, headers=headers, auth=('admin', 'admin'))
 356         self.assertEqual(response.status_code, requests.codes.not_found)
 357         res = response.json()
 358         self.assertIn(
 359             {"error-type":"application", "error-tag":"data-missing",
 360              "error-message":"Request could not be completed because the relevant data model content does not exist "},
 361             res['errors']['error'])
 362
 363
 364 if __name__ == "__main__":
 365     unittest.main(verbosity=2)