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     testtools_process1 = None
  26     testtools_process2 = None
  27     odl_process = None
  28     restconf_baseurl = "http://localhost:8181/restconf"
  29
  30     @classmethod
  31     def __start_testtools(cls):
  32         executable = ("./netconf/netconf/tools/netconf-testtool/target/"
  33                       "netconf-testtool-1.3.1-executable.jar")
  34         if os.path.isfile(executable):
  35             with open('testtools1.log', 'w') as outfile:
  36                 cls.testtools_process1 = subprocess.Popen(
  37                     ["java", "-jar", executable, "--schemas-dir", "schemas",
  38                      "--initial-config-xml", "sample_configs/ord_1.2.1/sample-config-ROADM.xml",
  39                      "--starting-port", "17830"],
  40                     stdout=outfile)
  41             with open('testtools2.log', 'w') as outfile:
  42                 cls.testtools_process2 = subprocess.Popen(
  43                     ["java", "-jar", executable, "--schemas-dir", "schemas",
  44                      "--initial-config-xml", "sample_configs/ord_1.2.1/sample-config-XPDR.xml",
  45                      "--starting-port", "17831"],
  46                     stdout=outfile)
  47
  48     @classmethod
  49     def __start_odl(cls):
  50         executable = "../karaf/target/assembly/bin/karaf"
  51         with open('odl.log', 'w') as outfile:
  52             cls.odl_process = subprocess.Popen(
  53                 ["bash", executable, "server"], stdout=outfile,
  54                 stdin=open(os.devnull))
  55
  56     @classmethod
  57     def setUpClass(cls):
  58         cls.__start_testtools()
  59         cls.__start_odl()
  60         time.sleep(60)
  61
  62     @classmethod
  63     def tearDownClass(cls):
  64         cls.testtools_process1.send_signal(signal.SIGINT)
  65         cls.testtools_process1.wait()
  66         cls.testtools_process2.send_signal(signal.SIGINT)
  67         cls.testtools_process2.wait()
  68         for child in psutil.Process(cls.odl_process.pid).children():
  69             child.send_signal(signal.SIGINT)
  70             child.wait()
  71         cls.odl_process.send_signal(signal.SIGINT)
  72         cls.odl_process.wait()
  73
  74     def setUp(self):
  75         print ("execution of {}".format(self.id().split(".")[-1]))
  76         time.sleep(10)
  77
  78     def test_01_restconfAPI(self):
  79         url = ("{}/operational/network-topology:network-topology/topology/"
  80         "topology-netconf/node/controller-config".format(self.restconf_baseurl))
  81         headers = {'content-type': 'application/json'}
  82         response = requests.request("GET", url, headers=headers, auth=('admin', 'admin'))
  83         self.assertEqual(response.status_code, requests.codes.ok)
  84         res = response.json()
  85         self.assertEqual(res['node'] [0] ['netconf-node-topology:connection-status'],
  86                          'connected')
  87
  88     def test_02_restconfAPI(self):
  89         url = ("{}/config/portmapping:network/nodes/controller-config"
  90                .format(self.restconf_baseurl))
  91         headers = {'content-type': 'application/json'}
  92         response = requests.request(
  93             "GET", url, headers=headers, auth=('admin', 'admin'))
  94         self.assertEqual(response.status_code, requests.codes.not_found)
  95         res = response.json()
  96         self.assertIn(
  97             {"error-type":"application", "error-tag":"data-missing",
  98              "error-message":"Request could not be completed because the relevant data model content does not exist "},
  99             res['errors']['error'])
 100
 101     def test_03_rdm_device_connected(self):
 102         url = ("{}/config/network-topology:"
 103                "network-topology/topology/topology-netconf/node/ROADMA"
 104               .format(self.restconf_baseurl))
 105         data = {"node": [{
 106             "node-id": "ROADMA",
 107             "netconf-node-topology:username": "admin",
 108             "netconf-node-topology:password": "admin",
 109             "netconf-node-topology:host": "127.0.0.1",
 110             "netconf-node-topology:port": "17830",
 111             "netconf-node-topology:tcp-only": "false",
 112             "netconf-node-topology:pass-through": {}}]}
 113         headers = {'content-type': 'application/json'}
 114         response = requests.request(
 115             "PUT", url, data=json.dumps(data), headers=headers,
 116             auth=('admin', 'admin'))
 117         self.assertEqual(response.status_code, requests.codes.created)
 118         time.sleep(20)
 119
 120     def test_04_rdm_device_connected(self):
 121         url = ("{}/operational/network-topology:"
 122                "network-topology/topology/topology-netconf/node/ROADMA"
 123                .format(self.restconf_baseurl))
 124         headers = {'content-type': 'application/json'}
 125         response = requests.request(
 126             "GET", url, headers=headers, auth=('admin', 'admin'))
 127         self.assertEqual(response.status_code, requests.codes.ok)
 128         res = response.json()
 129         self.assertEqual(
 130             res['node'][0]['netconf-node-topology:connection-status'],
 131             'connected')
 132         time.sleep(10)
 133
 134     def test_05_rdm_portmapping_DEG1_TTP_TXRX(self):
 135         url = ("{}/config/portmapping:network/"
 136                "nodes/ROADMA/mapping/DEG1-TTP-TXRX"
 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.assertIn(
 144             {'supporting-port': 'L1', 'supporting-circuit-pack-name': '2/0',
 145              'logical-connection-point': 'DEG1-TTP-TXRX'},
 146             res['mapping'])
 147
 148     def test_06_rdm_portmapping_SRG1_PP7_TXRX(self):
 149         url = ("{}/config/portmapping:network/"
 150                "nodes/ROADMA/mapping/SRG1-PP7-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': 'C7', 'supporting-circuit-pack-name': '4/0',
 159              'logical-connection-point': 'SRG1-PP7-TXRX'},
 160             res['mapping'])
 161
 162     def test_07_rdm_portmapping_SRG3_PP1_TXRX(self):
 163         url = ("{}/config/portmapping:network/"
 164                "nodes/ROADMA/mapping/SRG3-PP1-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': 'C1', 'supporting-circuit-pack-name': '5/0',
 173              'logical-connection-point': 'SRG3-PP1-TXRX'},
 174             res['mapping'])
 175
 176     def test_08_xpdr_device_connected(self):
 177         url = ("{}/config/network-topology:"
 178                "network-topology/topology/topology-netconf/node/XPDRA"
 179               .format(self.restconf_baseurl))
 180         data = {"node": [{
 181             "node-id": "XPDRA",
 182             "netconf-node-topology:username": "admin",
 183             "netconf-node-topology:password": "admin",
 184             "netconf-node-topology:host": "127.0.0.1",
 185             "netconf-node-topology:port": "17831",
 186             "netconf-node-topology:tcp-only": "false",
 187             "netconf-node-topology:pass-through": {}}]}
 188         headers = {'content-type': 'application/json'}
 189         response = requests.request(
 190             "PUT", url, data=json.dumps(data), headers=headers,
 191             auth=('admin', 'admin'))
 192         self.assertEqual(response.status_code, requests.codes.created)
 193         time.sleep(20)
 194
 195     def test_09_xpdr_device_connected(self):
 196         url = ("{}/operational/network-topology:"
 197                "network-topology/topology/topology-netconf/node/XPDRA"
 198                .format(self.restconf_baseurl))
 199         headers = {'content-type': 'application/json'}
 200         response = requests.request(
 201             "GET", url, headers=headers, auth=('admin', 'admin'))
 202         self.assertEqual(response.status_code, requests.codes.ok)
 203         res = response.json()
 204         self.assertEqual(
 205             res['node'][0]['netconf-node-topology:connection-status'],
 206             'connected')
 207         time.sleep(10)
 208
 209     def test_10_xpdr_portmapping_NETWORK1(self):
 210         url = ("{}/config/portmapping:network/"
 211                "nodes/XPDRA/mapping/XPDR-LINE1"
 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.assertIn(
 219             {'supporting-port': '1', 'supporting-circuit-pack-name': '1/0/1-PLUG-NET',
 220              'logical-connection-point': 'XPDR-LINE1'},
 221             res['mapping'])
 222
 223     def test_11_xpdr_portmapping_CLIENT1(self):
 224         url = ("{}/config/portmapping:network/"
 225                "nodes/XPDRA/mapping/XPDR-CLNT1"
 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',
 234              'supporting-circuit-pack-name': '1/0/2-PLUG-CLIENT',
 235              'logical-connection-point': 'XPDR-CLNT1'},
 236             res['mapping'])
 237
 238     def test_12_xpdr_device_disconnected(self):
 239         url = ("{}/config/network-topology:"
 240                 "network-topology/topology/topology-netconf/node/XPDRA"
 241                .format(self.restconf_baseurl))
 242         headers = {'content-type': 'application/json'}
 243         response = requests.request(
 244              "DELETE", url, headers=headers,
 245              auth=('admin', 'admin'))
 246         self.assertEqual(response.status_code, requests.codes.ok)
 247         time.sleep(20)
 248
 249     def test_13_xpdr_device_disconnected(self):
 250         url = ("{}/operational/network-topology:network-topology/topology/"
 251                "topology-netconf/node/XPDRA".format(self.restconf_baseurl))
 252         headers = {'content-type': 'application/json'}
 253         response = requests.request(
 254             "GET", url, headers=headers, auth=('admin', 'admin'))
 255         self.assertEqual(response.status_code, requests.codes.not_found)
 256         res = response.json()
 257         self.assertIn(
 258             {"error-type":"application", "error-tag":"data-missing",
 259              "error-message":"Request could not be completed because the relevant data model content does not exist "},
 260             res['errors']['error'])
 261
 262     def test_14_xpdr_device_disconnected(self):
 263         url = ("{}/config/portmapping:network/"
 264                "nodes/XPDRA"
 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_15_rdm_device_disconnected(self):
 277         url = ("{}/config/network-topology:"
 278                 "network-topology/topology/topology-netconf/node/ROADMA"
 279                .format(self.restconf_baseurl))
 280         headers = {'content-type': 'application/json'}
 281         response = requests.request(
 282              "DELETE", url, headers=headers,
 283              auth=('admin', 'admin'))
 284         self.assertEqual(response.status_code, requests.codes.ok)
 285         time.sleep(20)
 286
 287     def test_16_rdm_device_disconnected(self):
 288         url = ("{}/operational/network-topology:network-topology/topology/"
 289                "topology-netconf/node/ROADMA".format(self.restconf_baseurl))
 290         headers = {'content-type': 'application/json'}
 291         response = requests.request(
 292             "GET", url, headers=headers, auth=('admin', 'admin'))
 293         self.assertEqual(response.status_code, requests.codes.not_found)
 294         res = response.json()
 295         self.assertIn(
 296             {"error-type":"application", "error-tag":"data-missing",
 297              "error-message":"Request could not be completed because the relevant data model content does not exist "},
 298             res['errors']['error'])
 299
 300     def test_17_rdm_device_disconnected(self):
 301         url = ("{}/config/portmapping:network/nodes/ROADMA"
 302                .format(self.restconf_baseurl))
 303         headers = {'content-type': 'application/json'}
 304         response = requests.request(
 305             "GET", url, headers=headers, auth=('admin', 'admin'))
 306         self.assertEqual(response.status_code, requests.codes.not_found)
 307         res = response.json()
 308         self.assertIn(
 309             {"error-type":"application", "error-tag":"data-missing",
 310              "error-message":"Request could not be completed because the relevant data model content does not exist "},
 311             res['errors']['error'])
 312
 313
 314 if __name__ == "__main__":
 315     unittest.main(verbosity=2)