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              'lcp-hash-val': '3b3ab304d2a6eb3c3623e52746dbb7aa'},
 249             res['mapping'])
 250
 251     def test_11_xpdr_portmapping_NETWORK2(self):
 252         url = ("{}/config/transportpce-portmapping:network/"
 253                "nodes/XPDRA01/mapping/XPDR1-NETWORK2"
 254                .format(self.restconf_baseurl))
 255         headers = {'content-type': 'application/json'}
 256         response = requests.request(
 257             "GET", url, headers=headers, auth=('admin', 'admin'))
 258         self.assertEqual(response.status_code, requests.codes.ok)
 259         res = response.json()
 260         self.assertIn(
 261             {'supporting-port': '1', 'supporting-circuit-pack-name': '1/0/2-PLUG-NET',
 262              'logical-connection-point': 'XPDR1-NETWORK2', 'port-direction': 'bidirectional',
 263              'connection-map-lcp': 'XPDR1-CLIENT3', 'port-qual': 'xpdr-network',
 264              'lcp-hash-val': '3b3ab304d2a6eb3c3623e52746dbb7a9'},
 265             res['mapping'])
 266
 267     def test_12_xpdr_portmapping_CLIENT1(self):
 268         url = ("{}/config/transportpce-portmapping:network/"
 269                "nodes/XPDRA01/mapping/XPDR1-CLIENT1"
 270                .format(self.restconf_baseurl))
 271         headers = {'content-type': 'application/json'}
 272         response = requests.request(
 273             "GET", url, headers=headers, auth=('admin', 'admin'))
 274         self.assertEqual(response.status_code, requests.codes.ok)
 275         res = response.json()
 276         self.assertIn(
 277             {'supporting-port': 'C1',
 278              'supporting-circuit-pack-name': '1/0/C1-PLUG-CLIENT',
 279              'logical-connection-point': 'XPDR1-CLIENT1', 'port-direction': 'bidirectional',
 280              'connection-map-lcp': 'XPDR1-NETWORK1', 'port-qual': 'xpdr-client',
 281              'lcp-hash-val': '64b8effe7ba72211420bf267d0ca1ae5'},
 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              'lcp-hash-val': '64b8effe7ba72211420bf267d0ca1ae6'},
 299             res['mapping'])
 300
 301     def test_14_xpdr_portmapping_CLIENT3(self):
 302         url = ("{}/config/transportpce-portmapping:network/"
 303                "nodes/XPDRA01/mapping/XPDR1-CLIENT3"
 304                .format(self.restconf_baseurl))
 305         headers = {'content-type': 'application/json'}
 306         response = requests.request(
 307             "GET", url, headers=headers, auth=('admin', 'admin'))
 308         self.assertEqual(response.status_code, requests.codes.ok)
 309         res = response.json()
 310         self.assertIn(
 311             {'supporting-port': 'C3',
 312              'supporting-circuit-pack-name': '1/0/C3-PLUG-CLIENT',
 313              'logical-connection-point': 'XPDR1-CLIENT3',
 314              'connection-map-lcp': 'XPDR1-NETWORK2', 'port-direction': 'bidirectional',
 315              'port-qual': 'xpdr-client', 'lcp-hash-val': '64b8effe7ba72211420bf267d0ca1ae7'},
 316             res['mapping'])
 317
 318     def test_15_xpdr_portmapping_CLIENT4(self):
 319         url = ("{}/config/transportpce-portmapping:network/"
 320                "nodes/XPDRA01/mapping/XPDR1-CLIENT4"
 321                .format(self.restconf_baseurl))
 322         headers = {'content-type': 'application/json'}
 323         response = requests.request(
 324             "GET", url, headers=headers, auth=('admin', 'admin'))
 325         self.assertEqual(response.status_code, requests.codes.ok)
 326         res = response.json()
 327         self.assertIn(
 328             {'supporting-port': 'C4',
 329              'supporting-circuit-pack-name': '1/0/C4-PLUG-CLIENT',
 330              'logical-connection-point': 'XPDR1-CLIENT4', 'port-direction': 'bidirectional',
 331              'port-qual': 'xpdr-client', 'lcp-hash-val': '64b8effe7ba72211420bf267d0ca1ae0'},
 332             res['mapping'])
 333
 334     def test_16_xpdr_device_disconnected(self):
 335         url = ("{}/config/network-topology:"
 336                "network-topology/topology/topology-netconf/node/XPDRA01"
 337                .format(self.restconf_baseurl))
 338         headers = {'content-type': 'application/json'}
 339         response = requests.request(
 340             "DELETE", url, headers=headers,
 341             auth=('admin', 'admin'))
 342         self.assertEqual(response.status_code, requests.codes.ok)
 343         time.sleep(20)
 344
 345     def test_17_xpdr_device_disconnected(self):
 346         url = ("{}/operational/network-topology:network-topology/topology/"
 347                "topology-netconf/node/XPDRA01".format(self.restconf_baseurl))
 348         headers = {'content-type': 'application/json'}
 349         response = requests.request(
 350             "GET", url, headers=headers, auth=('admin', 'admin'))
 351         self.assertEqual(response.status_code, requests.codes.not_found)
 352         res = response.json()
 353         self.assertIn(
 354             {"error-type": "application", "error-tag": "data-missing",
 355              "error-message": "Request could not be completed because the relevant data model content does not exist"},
 356             res['errors']['error'])
 357
 358     def test_18_xpdr_device_disconnected(self):
 359         url = ("{}/config/transportpce-portmapping:network/nodes/XPDRA01".format(self.restconf_baseurl))
 360         headers = {'content-type': 'application/json'}
 361         response = requests.request(
 362             "GET", url, headers=headers, auth=('admin', 'admin'))
 363         self.assertEqual(response.status_code, requests.codes.not_found)
 364         res = response.json()
 365         self.assertIn(
 366             {"error-type": "application", "error-tag": "data-missing",
 367              "error-message": "Request could not be completed because the relevant data model content does not exist"},
 368             res['errors']['error'])
 369
 370     def test_19_rdm_device_disconnected(self):
 371         url = ("{}/config/network-topology:network-topology/topology/topology-netconf/node/ROADMA01"
 372                .format(self.restconf_baseurl))
 373         headers = {'content-type': 'application/json'}
 374         response = requests.request(
 375             "DELETE", url, headers=headers,
 376             auth=('admin', 'admin'))
 377         self.assertEqual(response.status_code, requests.codes.ok)
 378         time.sleep(20)
 379
 380     def test_20_rdm_device_disconnected(self):
 381         url = ("{}/operational/network-topology:network-topology/topology/topology-netconf/node/ROADMA01"
 382                .format(self.restconf_baseurl))
 383         headers = {'content-type': 'application/json'}
 384         response = requests.request(
 385             "GET", url, headers=headers, auth=('admin', 'admin'))
 386         self.assertEqual(response.status_code, requests.codes.not_found)
 387         res = response.json()
 388         self.assertIn(
 389             {"error-type": "application", "error-tag": "data-missing",
 390              "error-message": "Request could not be completed because the relevant data model content does not exist"},
 391             res['errors']['error'])
 392
 393     def test_21_rdm_device_disconnected(self):
 394         url = ("{}/config/transportpce-portmapping:network/nodes/ROADMA01".format(self.restconf_baseurl))
 395         headers = {'content-type': 'application/json'}
 396         response = requests.request(
 397             "GET", url, headers=headers, auth=('admin', 'admin'))
 398         self.assertEqual(response.status_code, requests.codes.not_found)
 399         res = response.json()
 400         self.assertIn(
 401             {"error-type": "application", "error-tag": "data-missing",
 402              "error-message": "Request could not be completed because the relevant data model content does not exist"},
 403             res['errors']['error'])
 404
 405
 406 if __name__ == "__main__":
 407     unittest.main(verbosity=2)