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