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, "17830", "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, "17831", "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": "17831",
 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_info(self):
 149         url = ("{}/config/transportpce-portmapping:network/"
 150                "nodes/ROADMA/node-info"
 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.assertEqual(
 158             {u'node-info': {u'node-type': u'rdm',
 159              u'node-ip-address': u'127.0.0.12',
 160              u'node-clli': u'NodeA',
 161              u'openroadm-version': u'1.2.1', u'node-vendor': u'vendorA',
 162              u'node-model': u'2'}},
 163             res)
 164         time.sleep(3)
 165
 166     def test_04_rdm_portmapping_DEG1_TTP_TXRX(self):
 167         url = ("{}/config/transportpce-portmapping:network/"
 168                "nodes/ROADMA/mapping/DEG1-TTP-TXRX"
 169                .format(self.restconf_baseurl))
 170         headers = {'content-type': 'application/json'}
 171         response = requests.request(
 172             "GET", url, headers=headers, auth=('admin', 'admin'))
 173         self.assertEqual(response.status_code, requests.codes.ok)
 174         res = response.json()
 175         self.assertIn(
 176             {'supporting-port': 'L1', 'supporting-circuit-pack-name': '2/0',
 177              'logical-connection-point': 'DEG1-TTP-TXRX', 'port-direction': 'bidirectional'},
 178             res['mapping'])
 179
 180     def test_05_rdm_portmapping_SRG1_PP7_TXRX(self):
 181         url = ("{}/config/transportpce-portmapping:network/"
 182                "nodes/ROADMA/mapping/SRG1-PP7-TXRX"
 183                .format(self.restconf_baseurl))
 184         headers = {'content-type': 'application/json'}
 185         response = requests.request(
 186             "GET", url, headers=headers, auth=('admin', 'admin'))
 187         self.assertEqual(response.status_code, requests.codes.ok)
 188         res = response.json()
 189         self.assertIn(
 190             {'supporting-port': 'C7', 'supporting-circuit-pack-name': '4/0',
 191              'logical-connection-point': 'SRG1-PP7-TXRX', 'port-direction': 'bidirectional'},
 192             res['mapping'])
 193
 194     def test_06_rdm_portmapping_SRG3_PP1_TXRX(self):
 195         url = ("{}/config/transportpce-portmapping:network/"
 196                "nodes/ROADMA/mapping/SRG3-PP1-TXRX"
 197                .format(self.restconf_baseurl))
 198         headers = {'content-type': 'application/json'}
 199         response = requests.request(
 200             "GET", url, headers=headers, auth=('admin', 'admin'))
 201         self.assertEqual(response.status_code, requests.codes.ok)
 202         res = response.json()
 203         self.assertIn(
 204             {'supporting-port': 'C1', 'supporting-circuit-pack-name': '5/0',
 205              'logical-connection-point': 'SRG3-PP1-TXRX', 'port-direction': 'bidirectional'},
 206             res['mapping'])
 207
 208     def test_07_xpdr_device_connected(self):
 209         url = ("{}/config/network-topology:"
 210                "network-topology/topology/topology-netconf/node/XPDRA"
 211               .format(self.restconf_baseurl))
 212         data = {"node": [{
 213             "node-id": "XPDRA",
 214             "netconf-node-topology:username": "admin",
 215             "netconf-node-topology:password": "admin",
 216             "netconf-node-topology:host": "127.0.0.1",
 217             "netconf-node-topology:port": "17830",
 218             "netconf-node-topology:tcp-only": "false",
 219             "netconf-node-topology:pass-through": {}}]}
 220         headers = {'content-type': 'application/json'}
 221         response = requests.request(
 222             "PUT", url, data=json.dumps(data), headers=headers,
 223             auth=('admin', 'admin'))
 224         self.assertEqual(response.status_code, requests.codes.created)
 225         time.sleep(20)
 226
 227     def test_08_xpdr_device_connected(self):
 228         url = ("{}/operational/network-topology:"
 229                "network-topology/topology/topology-netconf/node/XPDRA"
 230                .format(self.restconf_baseurl))
 231         headers = {'content-type': 'application/json'}
 232         response = requests.request(
 233             "GET", url, headers=headers, auth=('admin', 'admin'))
 234         self.assertEqual(response.status_code, requests.codes.ok)
 235         res = response.json()
 236         self.assertEqual(
 237             res['node'][0]['netconf-node-topology:connection-status'],
 238             'connected')
 239         time.sleep(10)
 240
 241     def test_09_xpdr_portmapping_info(self):
 242         url = ("{}/config/transportpce-portmapping:network/"
 243                "nodes/XPDRA/node-info"
 244                .format(self.restconf_baseurl))
 245         headers = {'content-type': 'application/json'}
 246         response = requests.request(
 247             "GET", url, headers=headers, auth=('admin', 'admin'))
 248         self.assertEqual(response.status_code, requests.codes.ok)
 249         res = response.json()
 250         self.assertEqual(
 251             {u'node-info': {u'node-type': u'xpdr',
 252              u'node-ip-address': u'127.0.0.10',
 253              u'node-clli': u'NodeA',
 254              u'openroadm-version': u'1.2.1', u'node-vendor': u'vendorA',
 255              u'node-model': u'1'}},
 256             res)
 257         time.sleep(3)
 258
 259     def test_10_xpdr_portmapping_NETWORK1(self):
 260         url = ("{}/config/transportpce-portmapping:network/"
 261                "nodes/XPDRA/mapping/XPDR1-NETWORK1"
 262                .format(self.restconf_baseurl))
 263         headers = {'content-type': 'application/json'}
 264         response = requests.request(
 265             "GET", url, headers=headers, auth=('admin', 'admin'))
 266         self.assertEqual(response.status_code, requests.codes.ok)
 267         res = response.json()
 268         self.assertIn(
 269             {'supporting-port': '1', 'supporting-circuit-pack-name': '1/0/1-PLUG-NET',
 270              'logical-connection-point': 'XPDR1-NETWORK1', 'port-direction': 'bidirectional',
 271              'associated-lcp': 'XPDR1-CLIENT1', 'port-qual': 'xpdr-network'},
 272             res['mapping'])
 273
 274     def test_11_xpdr_portmapping_NETWORK2(self):
 275         url = ("{}/config/transportpce-portmapping:network/"
 276                "nodes/XPDRA/mapping/XPDR1-NETWORK2"
 277                .format(self.restconf_baseurl))
 278         headers = {'content-type': 'application/json'}
 279         response = requests.request(
 280             "GET", url, headers=headers, auth=('admin', 'admin'))
 281         self.assertEqual(response.status_code, requests.codes.ok)
 282         res = response.json()
 283         self.assertIn(
 284             {'supporting-port': '1', 'supporting-circuit-pack-name': '1/0/2-PLUG-NET',
 285              'logical-connection-point': 'XPDR1-NETWORK2', 'port-direction': 'bidirectional',
 286              'associated-lcp': 'XPDR1-CLIENT3', 'port-qual': 'xpdr-network'},
 287             res['mapping'])
 288
 289     def test_12_xpdr_portmapping_CLIENT1(self):
 290         url = ("{}/config/transportpce-portmapping:network/"
 291                "nodes/XPDRA/mapping/XPDR1-CLIENT1"
 292                .format(self.restconf_baseurl))
 293         headers = {'content-type': 'application/json'}
 294         response = requests.request(
 295             "GET", url, headers=headers, auth=('admin', 'admin'))
 296         self.assertEqual(response.status_code, requests.codes.ok)
 297         res = response.json()
 298         self.assertIn(
 299             {'supporting-port': 'C1',
 300              'supporting-circuit-pack-name': '1/0/C1-PLUG-CLIENT',
 301              'logical-connection-point': 'XPDR1-CLIENT1', 'port-direction': 'bidirectional',
 302              'associated-lcp': 'XPDR1-NETWORK1', 'port-qual': 'xpdr-client'},
 303             res['mapping'])
 304
 305     def test_13_xpdr_portmapping_CLIENT2(self):
 306         url = ("{}/config/transportpce-portmapping:network/"
 307                "nodes/XPDRA/mapping/XPDR1-CLIENT2"
 308                .format(self.restconf_baseurl))
 309         headers = {'content-type': 'application/json'}
 310         response = requests.request(
 311             "GET", url, headers=headers, auth=('admin', 'admin'))
 312         self.assertEqual(response.status_code, requests.codes.ok)
 313         res = response.json()
 314         self.assertIn(
 315             {'supporting-port': 'C2',
 316              'supporting-circuit-pack-name': '1/0/C2-PLUG-CLIENT',
 317              'logical-connection-point': 'XPDR1-CLIENT2', 'port-direction': 'bidirectional',
 318              'port-qual': 'xpdr-client'},
 319             res['mapping'])
 320
 321     def test_14_xpdr_portmapping_CLIENT3(self):
 322         url = ("{}/config/transportpce-portmapping:network/"
 323                "nodes/XPDRA/mapping/XPDR1-CLIENT3"
 324                .format(self.restconf_baseurl))
 325         headers = {'content-type': 'application/json'}
 326         response = requests.request(
 327             "GET", url, headers=headers, auth=('admin', 'admin'))
 328         self.assertEqual(response.status_code, requests.codes.ok)
 329         res = response.json()
 330         self.assertIn(
 331             {'supporting-port': 'C3',
 332              'supporting-circuit-pack-name': '1/0/C3-PLUG-CLIENT',
 333              'logical-connection-point': 'XPDR1-CLIENT3',
 334              'associated-lcp': 'XPDR1-NETWORK2', 'port-direction': 'bidirectional',
 335              'port-qual': 'xpdr-client'},
 336             res['mapping'])
 337
 338     def test_15_xpdr_portmapping_CLIENT4(self):
 339         url = ("{}/config/transportpce-portmapping:network/"
 340                "nodes/XPDRA/mapping/XPDR1-CLIENT4"
 341                .format(self.restconf_baseurl))
 342         headers = {'content-type': 'application/json'}
 343         response = requests.request(
 344             "GET", url, headers=headers, auth=('admin', 'admin'))
 345         self.assertEqual(response.status_code, requests.codes.ok)
 346         res = response.json()
 347         self.assertIn(
 348             {'supporting-port': 'C4',
 349              'supporting-circuit-pack-name': '1/0/C4-PLUG-CLIENT',
 350              'logical-connection-point': 'XPDR1-CLIENT4', 'port-direction': 'bidirectional',
 351              'port-qual': 'xpdr-client'},
 352             res['mapping'])
 353
 354     def test_16_xpdr_device_disconnected(self):
 355         url = ("{}/config/network-topology:"
 356                 "network-topology/topology/topology-netconf/node/XPDRA"
 357                .format(self.restconf_baseurl))
 358         headers = {'content-type': 'application/json'}
 359         response = requests.request(
 360              "DELETE", url, headers=headers,
 361              auth=('admin', 'admin'))
 362         self.assertEqual(response.status_code, requests.codes.ok)
 363         time.sleep(20)
 364
 365     def test_17_xpdr_device_disconnected(self):
 366         url = ("{}/operational/network-topology:network-topology/topology/"
 367                "topology-netconf/node/XPDRA".format(self.restconf_baseurl))
 368         headers = {'content-type': 'application/json'}
 369         response = requests.request(
 370             "GET", url, headers=headers, auth=('admin', 'admin'))
 371         self.assertEqual(response.status_code, requests.codes.not_found)
 372         res = response.json()
 373         self.assertIn(
 374             {"error-type":"application", "error-tag":"data-missing",
 375              "error-message":"Request could not be completed because the relevant data model content does not exist"},
 376             res['errors']['error'])
 377
 378     def test_18_xpdr_device_disconnected(self):
 379         url = ("{}/config/transportpce-portmapping:network/nodes/XPDRA".format(self.restconf_baseurl))
 380         headers = {'content-type': 'application/json'}
 381         response = requests.request(
 382             "GET", url, headers=headers, auth=('admin', 'admin'))
 383         self.assertEqual(response.status_code, requests.codes.not_found)
 384         res = response.json()
 385         self.assertIn(
 386             {"error-type":"application", "error-tag":"data-missing",
 387              "error-message":"Request could not be completed because the relevant data model content does not exist"},
 388             res['errors']['error'])
 389
 390     def test_19_rdm_device_disconnected(self):
 391         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/node/ROADMA"
 392                .format(self.restconf_baseurl))
 393         headers = {'content-type': 'application/json'}
 394         response = requests.request(
 395              "DELETE", url, headers=headers,
 396              auth=('admin', 'admin'))
 397         self.assertEqual(response.status_code, requests.codes.ok)
 398         time.sleep(20)
 399
 400     def test_20_rdm_device_disconnected(self):
 401         url = ("{}/operational/network-topology:network-topology/topology/topology-netconf/node/ROADMA"
 402                .format(self.restconf_baseurl))
 403         headers = {'content-type': 'application/json'}
 404         response = requests.request(
 405             "GET", url, headers=headers, auth=('admin', 'admin'))
 406         self.assertEqual(response.status_code, requests.codes.not_found)
 407         res = response.json()
 408         self.assertIn(
 409             {"error-type":"application", "error-tag":"data-missing",
 410              "error-message":"Request could not be completed because the relevant data model content does not exist"},
 411             res['errors']['error'])
 412
 413     def test_21_rdm_device_disconnected(self):
 414         url = ("{}/config/transportpce-portmapping:network/nodes/ROADMA".format(self.restconf_baseurl))
 415         headers = {'content-type': 'application/json'}
 416         response = requests.request(
 417             "GET", url, headers=headers, auth=('admin', 'admin'))
 418         self.assertEqual(response.status_code, requests.codes.not_found)
 419         res = response.json()
 420         self.assertIn(
 421             {"error-type":"application", "error-tag":"data-missing",
 422              "error-message":"Request could not be completed because the relevant data model content does not exist"},
 423             res['errors']['error'])
 424
 425
 426 if __name__ == "__main__":
 427     unittest.main(verbosity=2)