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 import test_utils
  22
  23
  24 class TransportPCEPortMappingTesting(unittest.TestCase):
  25
  26     honeynode_process1 = None
  27     honeynode_process2 = None
  28     odl_process = None
  29     restconf_baseurl = "http://localhost:8181/restconf"
  30
  31 #START_IGNORE_XTESTING
  32
  33     @classmethod
  34     def setUpClass(cls):
  35         print ("starting honeynode1...")
  36         cls.honeynode_process1 = test_utils.start_xpdra_honeynode()
  37         time.sleep(20)
  38
  39         print ("starting honeynode2...")
  40         cls.honeynode_process2 = test_utils.start_roadma_honeynode()
  41         time.sleep(20)
  42
  43         print ("starting opendaylight...")
  44         cls.odl_process = test_utils.start_tpce()
  45         time.sleep(60)
  46         print ("opendaylight started")
  47
  48     @classmethod
  49     def tearDownClass(cls):
  50         for child in psutil.Process(cls.odl_process.pid).children():
  51             child.send_signal(signal.SIGINT)
  52             child.wait()
  53         cls.odl_process.send_signal(signal.SIGINT)
  54         cls.odl_process.wait()
  55         for child in psutil.Process(cls.honeynode_process1.pid).children():
  56             child.send_signal(signal.SIGINT)
  57             child.wait()
  58         cls.honeynode_process1.send_signal(signal.SIGINT)
  59         cls.honeynode_process1.wait()
  60         for child in psutil.Process(cls.honeynode_process2.pid).children():
  61             child.send_signal(signal.SIGINT)
  62             child.wait()
  63         cls.honeynode_process2.send_signal(signal.SIGINT)
  64         cls.honeynode_process2.wait()
  65
  66     def setUp(self):
  67         print ("execution of {}".format(self.id().split(".")[-1]))
  68         time.sleep(10)
  69
  70 #END_IGNORE_XTESTING
  71
  72 #    def test_01_restconfAPI(self):
  73 #        url = ("{}/operational/network-topology:network-topology/topology/"
  74 #        "topology-netconf/node/controller-config".format(self.restconf_baseurl))
  75 #        headers = {'content-type': 'application/json'}
  76 #        response = requests.request("GET", url, headers=headers, auth=('admin', 'admin'))
  77 #        self.assertEqual(response.status_code, requests.codes.ok)
  78 #        res = response.json()
  79 #        self.assertEqual(res['node'] [0] ['netconf-node-topology:connection-status'],
  80 #                         'connected')
  81
  82 #     def test_02_restconfAPI(self):
  83 #         url = ("{}/config/transportpce-portmapping:network/nodes/controller-config".format(self.restconf_baseurl))
  84 #         headers = {'content-type': 'application/json'}
  85 #         response = requests.request(
  86 #             "GET", url, headers=headers, auth=('admin', 'admin'))
  87 #         self.assertEqual(response.status_code, requests.codes.not_found)
  88 #         res = response.json()
  89 #         self.assertIn(
  90 #             {"error-type":"application", "error-tag":"data-missing",
  91 #              "error-message":"Request could not be completed because the relevant data model content does not exist "},
  92 #             res['errors']['error'])
  93
  94     def test_01_rdm_device_connected(self):
  95         url = ("{}/config/network-topology:"
  96                "network-topology/topology/topology-netconf/node/ROADMA01"
  97               .format(self.restconf_baseurl))
  98         data = {"node": [{
  99             "node-id": "ROADMA01",
 100             "netconf-node-topology:username": "admin",
 101             "netconf-node-topology:password": "admin",
 102             "netconf-node-topology:host": "127.0.0.1",
 103             "netconf-node-topology:port": "17831",
 104             "netconf-node-topology:tcp-only": "false",
 105             "netconf-node-topology:pass-through": {}}]}
 106         headers = {'content-type': 'application/json'}
 107         response = requests.request(
 108             "PUT", url, data=json.dumps(data), headers=headers,
 109             auth=('admin', 'admin'))
 110         self.assertEqual(response.status_code, requests.codes.created)
 111         time.sleep(20)
 112
 113     def test_02_rdm_device_connected(self):
 114         url = ("{}/operational/network-topology:"
 115                "network-topology/topology/topology-netconf/node/ROADMA01"
 116                .format(self.restconf_baseurl))
 117         headers = {'content-type': 'application/json'}
 118         response = requests.request(
 119             "GET", url, headers=headers, auth=('admin', 'admin'))
 120         self.assertEqual(response.status_code, requests.codes.ok)
 121         res = response.json()
 122         self.assertEqual(
 123             res['node'][0]['netconf-node-topology:connection-status'],
 124             'connected')
 125         time.sleep(10)
 126
 127     def test_03_rdm_portmapping_info(self):
 128         url = ("{}/config/transportpce-portmapping:network/"
 129                "nodes/ROADMA01/node-info"
 130                .format(self.restconf_baseurl))
 131         headers = {'content-type': 'application/json'}
 132         response = requests.request(
 133             "GET", url, headers=headers, auth=('admin', 'admin'))
 134         self.assertEqual(response.status_code, requests.codes.ok)
 135         res = response.json()
 136         self.assertEqual(
 137             {u'node-info': {u'node-type': u'rdm',
 138              u'node-ip-address': u'127.0.0.12',
 139              u'node-clli': u'NodeA',
 140              u'openroadm-version': u'1.2.1', u'node-vendor': u'vendorA',
 141              u'node-model': u'2'}},
 142             res)
 143         time.sleep(3)
 144
 145     def test_04_rdm_portmapping_DEG1_TTP_TXRX(self):
 146         url = ("{}/config/transportpce-portmapping:network/"
 147                "nodes/ROADMA01/mapping/DEG1-TTP-TXRX"
 148                .format(self.restconf_baseurl))
 149         headers = {'content-type': 'application/json'}
 150         response = requests.request(
 151             "GET", url, headers=headers, auth=('admin', 'admin'))
 152         self.assertEqual(response.status_code, requests.codes.ok)
 153         res = response.json()
 154         self.assertIn(
 155             {'supporting-port': 'L1', 'supporting-circuit-pack-name': '2/0',
 156              'logical-connection-point': 'DEG1-TTP-TXRX', 'port-direction': 'bidirectional'},
 157             res['mapping'])
 158
 159     def test_05_rdm_portmapping_SRG1_PP7_TXRX(self):
 160         url = ("{}/config/transportpce-portmapping:network/"
 161                "nodes/ROADMA01/mapping/SRG1-PP7-TXRX"
 162                .format(self.restconf_baseurl))
 163         headers = {'content-type': 'application/json'}
 164         response = requests.request(
 165             "GET", url, headers=headers, auth=('admin', 'admin'))
 166         self.assertEqual(response.status_code, requests.codes.ok)
 167         res = response.json()
 168         self.assertIn(
 169             {'supporting-port': 'C7', 'supporting-circuit-pack-name': '4/0',
 170              'logical-connection-point': 'SRG1-PP7-TXRX', 'port-direction': 'bidirectional'},
 171             res['mapping'])
 172
 173     def test_06_rdm_portmapping_SRG3_PP1_TXRX(self):
 174         url = ("{}/config/transportpce-portmapping:network/"
 175                "nodes/ROADMA01/mapping/SRG3-PP1-TXRX"
 176                .format(self.restconf_baseurl))
 177         headers = {'content-type': 'application/json'}
 178         response = requests.request(
 179             "GET", url, headers=headers, auth=('admin', 'admin'))
 180         self.assertEqual(response.status_code, requests.codes.ok)
 181         res = response.json()
 182         self.assertIn(
 183             {'supporting-port': 'C1', 'supporting-circuit-pack-name': '5/0',
 184              'logical-connection-point': 'SRG3-PP1-TXRX', 'port-direction': 'bidirectional'},
 185             res['mapping'])
 186
 187     def test_07_xpdr_device_connected(self):
 188         url = ("{}/config/network-topology:"
 189                "network-topology/topology/topology-netconf/node/XPDRA01"
 190               .format(self.restconf_baseurl))
 191         data = {"node": [{
 192             "node-id": "XPDRA01",
 193             "netconf-node-topology:username": "admin",
 194             "netconf-node-topology:password": "admin",
 195             "netconf-node-topology:host": "127.0.0.1",
 196             "netconf-node-topology:port": "17830",
 197             "netconf-node-topology:tcp-only": "false",
 198             "netconf-node-topology:pass-through": {}}]}
 199         headers = {'content-type': 'application/json'}
 200         response = requests.request(
 201             "PUT", url, data=json.dumps(data), headers=headers,
 202             auth=('admin', 'admin'))
 203         self.assertEqual(response.status_code, requests.codes.created)
 204         time.sleep(20)
 205
 206     def test_08_xpdr_device_connected(self):
 207         url = ("{}/operational/network-topology:"
 208                "network-topology/topology/topology-netconf/node/XPDRA01"
 209                .format(self.restconf_baseurl))
 210         headers = {'content-type': 'application/json'}
 211         response = requests.request(
 212             "GET", url, headers=headers, auth=('admin', 'admin'))
 213         self.assertEqual(response.status_code, requests.codes.ok)
 214         res = response.json()
 215         self.assertEqual(
 216             res['node'][0]['netconf-node-topology:connection-status'],
 217             'connected')
 218         time.sleep(10)
 219
 220     def test_09_xpdr_portmapping_info(self):
 221         url = ("{}/config/transportpce-portmapping:network/"
 222                "nodes/XPDRA01/node-info"
 223                .format(self.restconf_baseurl))
 224         headers = {'content-type': 'application/json'}
 225         response = requests.request(
 226             "GET", url, headers=headers, auth=('admin', 'admin'))
 227         self.assertEqual(response.status_code, requests.codes.ok)
 228         res = response.json()
 229         self.assertEqual(
 230             {u'node-info': {u'node-type': u'xpdr',
 231              u'node-ip-address': u'127.0.0.10',
 232              u'node-clli': u'NodeA',
 233              u'openroadm-version': u'1.2.1', u'node-vendor': u'vendorA',
 234              u'node-model': u'1'}},
 235             res)
 236         time.sleep(3)
 237
 238     def test_10_xpdr_portmapping_NETWORK1(self):
 239         url = ("{}/config/transportpce-portmapping:network/"
 240                "nodes/XPDRA01/mapping/XPDR1-NETWORK1"
 241                .format(self.restconf_baseurl))
 242         headers = {'content-type': 'application/json'}
 243         response = requests.request(
 244             "GET", url, headers=headers, auth=('admin', 'admin'))
 245         self.assertEqual(response.status_code, requests.codes.ok)
 246         res = response.json()
 247         self.assertIn(
 248             {'supporting-port': '1', 'supporting-circuit-pack-name': '1/0/1-PLUG-NET',
 249              'logical-connection-point': 'XPDR1-NETWORK1', 'port-direction': 'bidirectional',
 250              'connection-map-lcp': 'XPDR1-CLIENT1', 'port-qual': 'xpdr-network'},
 251             res['mapping'])
 252
 253     def test_11_xpdr_portmapping_NETWORK2(self):
 254         url = ("{}/config/transportpce-portmapping:network/"
 255                "nodes/XPDRA01/mapping/XPDR1-NETWORK2"
 256                .format(self.restconf_baseurl))
 257         headers = {'content-type': 'application/json'}
 258         response = requests.request(
 259             "GET", url, headers=headers, auth=('admin', 'admin'))
 260         self.assertEqual(response.status_code, requests.codes.ok)
 261         res = response.json()
 262         self.assertIn(
 263             {'supporting-port': '1', 'supporting-circuit-pack-name': '1/0/2-PLUG-NET',
 264              'logical-connection-point': 'XPDR1-NETWORK2', 'port-direction': 'bidirectional',
 265              'connection-map-lcp': 'XPDR1-CLIENT3', 'port-qual': 'xpdr-network'},
 266             res['mapping'])
 267
 268     def test_12_xpdr_portmapping_CLIENT1(self):
 269         url = ("{}/config/transportpce-portmapping:network/"
 270                "nodes/XPDRA01/mapping/XPDR1-CLIENT1"
 271                .format(self.restconf_baseurl))
 272         headers = {'content-type': 'application/json'}
 273         response = requests.request(
 274             "GET", url, headers=headers, auth=('admin', 'admin'))
 275         self.assertEqual(response.status_code, requests.codes.ok)
 276         res = response.json()
 277         self.assertIn(
 278             {'supporting-port': 'C1',
 279              'supporting-circuit-pack-name': '1/0/C1-PLUG-CLIENT',
 280              'logical-connection-point': 'XPDR1-CLIENT1', 'port-direction': 'bidirectional',
 281              'connection-map-lcp': 'XPDR1-NETWORK1', 'port-qual': 'xpdr-client'},
 282             res['mapping'])
 283
 284     def test_13_xpdr_portmapping_CLIENT2(self):
 285         url = ("{}/config/transportpce-portmapping:network/"
 286                "nodes/XPDRA01/mapping/XPDR1-CLIENT2"
 287                .format(self.restconf_baseurl))
 288         headers = {'content-type': 'application/json'}
 289         response = requests.request(
 290             "GET", url, headers=headers, auth=('admin', 'admin'))
 291         self.assertEqual(response.status_code, requests.codes.ok)
 292         res = response.json()
 293         self.assertIn(
 294             {'supporting-port': 'C2',
 295              'supporting-circuit-pack-name': '1/0/C2-PLUG-CLIENT',
 296              'logical-connection-point': 'XPDR1-CLIENT2', 'port-direction': 'bidirectional',
 297              'port-qual': 'xpdr-client'},
 298             res['mapping'])
 299
 300     def test_14_xpdr_portmapping_CLIENT3(self):
 301         url = ("{}/config/transportpce-portmapping:network/"
 302                "nodes/XPDRA01/mapping/XPDR1-CLIENT3"
 303                .format(self.restconf_baseurl))
 304         headers = {'content-type': 'application/json'}
 305         response = requests.request(
 306             "GET", url, headers=headers, auth=('admin', 'admin'))
 307         self.assertEqual(response.status_code, requests.codes.ok)
 308         res = response.json()
 309         self.assertIn(
 310             {'supporting-port': 'C3',
 311              'supporting-circuit-pack-name': '1/0/C3-PLUG-CLIENT',
 312              'logical-connection-point': 'XPDR1-CLIENT3',
 313              'connection-map-lcp': 'XPDR1-NETWORK2', 'port-direction': 'bidirectional',
 314              'port-qual': 'xpdr-client'},
 315             res['mapping'])
 316
 317     def test_15_xpdr_portmapping_CLIENT4(self):
 318         url = ("{}/config/transportpce-portmapping:network/"
 319                "nodes/XPDRA01/mapping/XPDR1-CLIENT4"
 320                .format(self.restconf_baseurl))
 321         headers = {'content-type': 'application/json'}
 322         response = requests.request(
 323             "GET", url, headers=headers, auth=('admin', 'admin'))
 324         self.assertEqual(response.status_code, requests.codes.ok)
 325         res = response.json()
 326         self.assertIn(
 327             {'supporting-port': 'C4',
 328              'supporting-circuit-pack-name': '1/0/C4-PLUG-CLIENT',
 329              'logical-connection-point': 'XPDR1-CLIENT4', 'port-direction': 'bidirectional',
 330              'port-qual': 'xpdr-client'},
 331             res['mapping'])
 332
 333     def test_16_xpdr_device_disconnected(self):
 334         url = ("{}/config/network-topology:"
 335                 "network-topology/topology/topology-netconf/node/XPDRA01"
 336                .format(self.restconf_baseurl))
 337         headers = {'content-type': 'application/json'}
 338         response = requests.request(
 339              "DELETE", url, headers=headers,
 340              auth=('admin', 'admin'))
 341         self.assertEqual(response.status_code, requests.codes.ok)
 342         time.sleep(20)
 343
 344     def test_17_xpdr_device_disconnected(self):
 345         url = ("{}/operational/network-topology:network-topology/topology/"
 346                "topology-netconf/node/XPDRA01".format(self.restconf_baseurl))
 347         headers = {'content-type': 'application/json'}
 348         response = requests.request(
 349             "GET", url, headers=headers, auth=('admin', 'admin'))
 350         self.assertEqual(response.status_code, requests.codes.not_found)
 351         res = response.json()
 352         self.assertIn(
 353             {"error-type":"application", "error-tag":"data-missing",
 354              "error-message":"Request could not be completed because the relevant data model content does not exist"},
 355             res['errors']['error'])
 356
 357     def test_18_xpdr_device_disconnected(self):
 358         url = ("{}/config/transportpce-portmapping:network/nodes/XPDRA01".format(self.restconf_baseurl))
 359         headers = {'content-type': 'application/json'}
 360         response = requests.request(
 361             "GET", url, headers=headers, auth=('admin', 'admin'))
 362         self.assertEqual(response.status_code, requests.codes.not_found)
 363         res = response.json()
 364         self.assertIn(
 365             {"error-type":"application", "error-tag":"data-missing",
 366              "error-message":"Request could not be completed because the relevant data model content does not exist"},
 367             res['errors']['error'])
 368
 369     def test_19_rdm_device_disconnected(self):
 370         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/node/ROADMA01"
 371                .format(self.restconf_baseurl))
 372         headers = {'content-type': 'application/json'}
 373         response = requests.request(
 374              "DELETE", url, headers=headers,
 375              auth=('admin', 'admin'))
 376         self.assertEqual(response.status_code, requests.codes.ok)
 377         time.sleep(20)
 378
 379     def test_20_rdm_device_disconnected(self):
 380         url = ("{}/operational/network-topology:network-topology/topology/topology-netconf/node/ROADMA01"
 381                .format(self.restconf_baseurl))
 382         headers = {'content-type': 'application/json'}
 383         response = requests.request(
 384             "GET", url, headers=headers, auth=('admin', 'admin'))
 385         self.assertEqual(response.status_code, requests.codes.not_found)
 386         res = response.json()
 387         self.assertIn(
 388             {"error-type":"application", "error-tag":"data-missing",
 389              "error-message":"Request could not be completed because the relevant data model content does not exist"},
 390             res['errors']['error'])
 391
 392     def test_21_rdm_device_disconnected(self):
 393         url = ("{}/config/transportpce-portmapping:network/nodes/ROADMA01".format(self.restconf_baseurl))
 394         headers = {'content-type': 'application/json'}
 395         response = requests.request(
 396             "GET", url, headers=headers, auth=('admin', 'admin'))
 397         self.assertEqual(response.status_code, requests.codes.not_found)
 398         res = response.json()
 399         self.assertIn(
 400             {"error-type":"application", "error-tag":"data-missing",
 401              "error-message":"Request could not be completed because the relevant data model content does not exist"},
 402             res['errors']['error'])
 403
 404
 405 if __name__ == "__main__":
 406     unittest.main(verbosity=2)