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