tests/transportpce_tests/2.2.1/test_otn_topology.py

   1 #!/usr/bin/env python
   2
   3 ##############################################################################
   4 # Copyright (c) 2019 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 logging
  22 import test_utils
  23
  24
  25 class TransportPCEtesting(unittest.TestCase):
  26
  27     honeynode_process1 = None
  28     odl_process = None
  29     restconf_baseurl = "http://localhost:8181/restconf"
  30
  31     @classmethod
  32     def setUpClass(cls):
  33         print("starting honeynode1...")
  34         cls.honeynode_process1 = test_utils.start_spdra_honeynode()
  35         time.sleep(20)
  36
  37         print("starting opendaylight...")
  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.honeynode_process1.pid).children():
  50             child.send_signal(signal.SIGINT)
  51             child.wait()
  52         cls.honeynode_process1.send_signal(signal.SIGINT)
  53         cls.honeynode_process1.wait()
  54
  55     def setUp(self):
  56         time.sleep(5)
  57
  58     def test_01_connect_SPDR_SA1(self):
  59         url = ("{}/config/network-topology:"
  60                "network-topology/topology/topology-netconf/node/SPDR-SA1"
  61                .format(self.restconf_baseurl))
  62         data = {"node": [{
  63             "node-id": "SPDR-SA1",
  64             "netconf-node-topology:username": "admin",
  65             "netconf-node-topology:password": "admin",
  66             "netconf-node-topology:host": "127.0.0.1",
  67             "netconf-node-topology:port": "17845",
  68             "netconf-node-topology:tcp-only": "false",
  69             "netconf-node-topology:pass-through": {}}]}
  70         headers = {'content-type': 'application/json'}
  71         response = requests.request(
  72             "PUT", url, data=json.dumps(data), headers=headers,
  73             auth=('admin', 'admin'))
  74         self.assertEqual(response.status_code, requests.codes.created)
  75         time.sleep(10)
  76         url = ("{}/operational/network-topology:"
  77                "network-topology/topology/topology-netconf/node/SPDR-SA1"
  78                .format(self.restconf_baseurl))
  79         response = requests.request(
  80             "GET", url, headers=headers, auth=('admin', 'admin'))
  81         self.assertEqual(response.status_code, requests.codes.ok)
  82         res = response.json()
  83         self.assertEqual(
  84             res['node'][0]['netconf-node-topology:connection-status'],
  85             'connected')
  86
  87     def test_02_getClliNetwork(self):
  88         url = ("{}/config/ietf-network:networks/network/clli-network"
  89                .format(self.restconf_baseurl))
  90         headers = {'content-type': 'application/json'}
  91         response = requests.request(
  92             "GET", url, headers=headers, auth=('admin', 'admin'))
  93         self.assertEqual(response.status_code, requests.codes.ok)
  94         res = response.json()
  95         logging.info(res)
  96         self.assertEqual(res['network'][0]['node'][0]['node-id'], 'NodeSA')
  97         self.assertEqual(res['network'][0]['node'][0]['org-openroadm-clli-network:clli'], 'NodeSA')
  98
  99     def test_03_getOpenRoadmNetwork(self):
 100         url = ("{}/config/ietf-network:networks/network/openroadm-network"
 101                .format(self.restconf_baseurl))
 102         headers = {'content-type': 'application/json'}
 103         response = requests.request(
 104             "GET", url, headers=headers, auth=('admin', 'admin'))
 105         self.assertEqual(response.status_code, requests.codes.ok)
 106         res = response.json()
 107         self.assertEqual(res['network'][0]['node'][0]['node-id'], 'SPDR-SA1')
 108         self.assertEqual(res['network'][0]['node'][0]['supporting-node'][0]['network-ref'], 'clli-network')
 109         self.assertEqual(res['network'][0]['node'][0]['supporting-node'][0]['node-ref'], 'NodeSA')
 110         self.assertEqual(res['network'][0]['node'][0]['org-openroadm-common-network:node-type'], 'XPONDER')
 111         self.assertEqual(res['network'][0]['node'][0]['org-openroadm-network:model'], 'universal-switchponder')
 112         self.assertEqual(res['network'][0]['node'][0]['org-openroadm-network:vendor'], 'vendorA')
 113         self.assertEqual(res['network'][0]['node'][0]['org-openroadm-network:ip'], '1.2.3.4')
 114
 115     def test_04_getLinks_OpenroadmTopology(self):
 116         url = ("{}/config/ietf-network:networks/network/openroadm-topology"
 117                .format(self.restconf_baseurl))
 118         headers = {'content-type': 'application/json'}
 119         response = requests.request(
 120             "GET", url, headers=headers, auth=('admin', 'admin'))
 121         self.assertEqual(response.status_code, requests.codes.ok)
 122         res = response.json()
 123         # Tests related to links
 124         self.assertNotIn('ietf-network-topology:link', res['network'][0])
 125
 126     def test_05_getNodes_OpenRoadmTopology(self):
 127         url = ("{}/config/ietf-network:networks/network/openroadm-topology"
 128                .format(self.restconf_baseurl))
 129         headers = {'content-type': 'application/json'}
 130         response = requests.request(
 131             "GET", url, headers=headers, auth=('admin', 'admin'))
 132         res = response.json()
 133         # Tests related to nodes
 134         self.assertEqual(response.status_code, requests.codes.ok)
 135         self.assertIn('node', res['network'][0])
 136         if ('node' in res['network'][0]):
 137             nbNode = len(res['network'][0]['node'])
 138             self.assertEqual(nbNode, 2)
 139             listNode = ['SPDR-SA1-XPDR1', 'SPDR-SA1-XPDR2']
 140             for i in range(0, nbNode):
 141                 nodeType = res['network'][0]['node'][i]['org-openroadm-common-network:node-type']
 142                 nodeId = res['network'][0]['node'][i]['node-id']
 143                 if (nodeId == 'SPDR-SA1-XPDR1'):
 144                     self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'SPDR-SA1'},
 145                                   res['network'][0]['node'][i]['supporting-node'])
 146                     self.assertIn({'network-ref': 'clli-network', 'node-ref': 'NodeSA'},
 147                                   res['network'][0]['node'][i]['supporting-node'])
 148                     self.assertEqual(nodeType, 'XPONDER')
 149                     nbTps = len(res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
 150                     client = 0
 151                     network = 0
 152                     for j in range(0, nbTps):
 153                         tpType = res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]['org-openroadm-common-network:tp-type']
 154                         tpId = res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]['tp-id']
 155                         if (tpType == 'XPONDER-CLIENT'):
 156                             client += 1
 157                         elif (tpType == 'XPONDER-NETWORK'):
 158                             network += 1
 159                     self.assertTrue(client == 0)
 160                     self.assertTrue(network == 1)
 161                     listNode.remove(nodeId)
 162                 elif (nodeId == 'SPDR-SA1-XPDR2'):
 163                     self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'SPDR-SA1'},
 164                                   res['network'][0]['node'][i]['supporting-node'])
 165                     self.assertIn({'network-ref': 'clli-network', 'node-ref': 'NodeSA'},
 166                                   res['network'][0]['node'][i]['supporting-node'])
 167                     self.assertEqual(nodeType, 'XPONDER')
 168                     nbTps = len(res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
 169                     client = 0
 170                     network = 0
 171                     for j in range(0, nbTps):
 172                         tpType = res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]['org-openroadm-common-network:tp-type']
 173                         tpId = res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]['tp-id']
 174                         if (tpType == 'XPONDER-CLIENT'):
 175                             client += 1
 176                         elif (tpType == 'XPONDER-NETWORK'):
 177                             network += 1
 178                     self.assertTrue(client == 0)
 179                     self.assertTrue(network == 4)
 180                     listNode.remove(nodeId)
 181                 else:
 182                     self.assertFalse(True)
 183             self.assertEqual(len(listNode), 0)
 184
 185     def test_06_getLinks_OtnTopology(self):
 186         url = ("{}/config/ietf-network:networks/network/otn-topology"
 187                .format(self.restconf_baseurl))
 188         headers = {'content-type': 'application/json'}
 189         response = requests.request(
 190             "GET", url, headers=headers, auth=('admin', 'admin'))
 191         self.assertEqual(response.status_code, requests.codes.ok)
 192         res = response.json()
 193         self.assertNotIn('ietf-network-topology:link', res['network'][0])
 194
 195     def test_07_getNodes_OtnTopology(self):
 196         url = ("{}/config/ietf-network:networks/network/otn-topology"
 197                .format(self.restconf_baseurl))
 198         headers = {'content-type': 'application/json'}
 199         response = requests.request(
 200             "GET", url, headers=headers, auth=('admin', 'admin'))
 201         res = response.json()
 202         self.assertEqual(response.status_code, requests.codes.ok)
 203         nbNode = len(res['network'][0]['node'])
 204         self.assertEqual(nbNode, 2)
 205         listNode = ['SPDR-SA1-XPDR1', 'SPDR-SA1-XPDR2']
 206         for i in range(0, nbNode):
 207             nodeId = res['network'][0]['node'][i]['node-id']
 208             if(nodeId == 'SPDR-SA1-XPDR1'):
 209                 nodeType = res['network'][0]['node'][i]['org-openroadm-common-network:node-type']
 210                 self.assertEqual(nodeType, 'MUXPDR')
 211                 self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'SPDR-SA1'},
 212                               res['network'][0]['node'][i]['supporting-node'])
 213                 self.assertIn({'network-ref': 'openroadm-topology', 'node-ref': 'SPDR-SA1-XPDR1'},
 214                               res['network'][0]['node'][i]['supporting-node'])
 215                 self.assertIn({'network-ref': 'clli-network', 'node-ref': 'NodeSA'},
 216                               res['network'][0]['node'][i]['supporting-node'])
 217                 self.assertEqual(res['network'][0]['node'][i]
 218                                  ['org-openroadm-otn-network-topology:xpdr-attributes']['xpdr-number'], 1)
 219                 nbTps = len(res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
 220                 client = 0
 221                 network = 0
 222                 for j in range(0, nbTps):
 223                     tpType = res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]['org-openroadm-common-network:tp-type']
 224                     tpId = res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]['tp-id']
 225                     if (tpType == 'XPONDER-CLIENT'):
 226                         client += 1
 227                         self.assertEqual(res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]['org-openroadm-otn-network-topology:tp-supported-interfaces']
 228                                          ['supported-interface-capability'][0]['if-cap-type'], 'org-openroadm-port-types:if-10GE-ODU2e')
 229                         self.assertEqual(res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]
 230                                          ['org-openroadm-otn-network-topology:xpdr-tp-port-connection-attributes']['rate'], 'org-openroadm-otn-common-types:ODU2e')
 231                     elif (tpType == 'XPONDER-NETWORK'):
 232                         network += 1
 233                         self.assertEqual(res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]['org-openroadm-otn-network-topology:tp-supported-interfaces']
 234                                          ['supported-interface-capability'][0]['if-cap-type'], 'org-openroadm-port-types:if-OCH-OTU4-ODU4')
 235                         self.assertEqual(res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]
 236                                          ['org-openroadm-otn-network-topology:xpdr-tp-port-connection-attributes']['rate'], 'org-openroadm-otn-common-types:ODU4')
 237                         self.assertEqual(res['network'][0]['node'][i]['ietf-network-topology:termination-point']
 238                                          [j]['supporting-termination-point'][0]['network-ref'], 'openroadm-topology')
 239                         self.assertEqual(res['network'][0]['node'][i]['ietf-network-topology:termination-point']
 240                                          [j]['supporting-termination-point'][0]['node-ref'], 'SPDR-SA1-XPDR1')
 241                         self.assertEqual(res['network'][0]['node'][i]['ietf-network-topology:termination-point']
 242                                          [j]['supporting-termination-point'][0]['tp-ref'], 'XPDR1-NETWORK1')
 243                 self.assertTrue(client == 4)
 244                 self.assertTrue(network == 1)
 245                 listNode.remove(nodeId)
 246                 nbNbl = len(res['network'][0]['node'][i]['org-openroadm-otn-network-topology:switching-pools']
 247                             ['odu-switching-pools'][0]['non-blocking-list'])
 248                 self.assertEqual(nbNbl, 4)
 249                 for k in range(0, nbNbl):
 250                     nbl = res['network'][0]['node'][i]['org-openroadm-otn-network-topology:switching-pools']['odu-switching-pools'][0]['non-blocking-list'][k]
 251                     if (nbl['nbl-number'] == 1):
 252                         self.assertEqual(nbl['available-interconnect-bandwidth'], 10)
 253                         self.assertEqual(nbl['interconnect-bandwidth-unit'], 1000000000)
 254                         self.assertIn('XPDR1-NETWORK1', nbl['tp-list'])
 255                         self.assertIn('XPDR1-CLIENT1', nbl['tp-list'])
 256                         self.assertNotIn('XPDR1-CLIENT2', nbl['tp-list'])
 257             elif(nodeId == 'SPDR-SA1-XPDR2'):
 258                 nodeType = res['network'][0]['node'][i]['org-openroadm-common-network:node-type']
 259                 self.assertEqual(nodeType, 'SWITCH')
 260                 self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'SPDR-SA1'},
 261                               res['network'][0]['node'][i]['supporting-node'])
 262                 self.assertIn({'network-ref': 'openroadm-topology', 'node-ref': 'SPDR-SA1-XPDR2'},
 263                               res['network'][0]['node'][i]['supporting-node'])
 264                 self.assertIn({'network-ref': 'clli-network', 'node-ref': 'NodeSA'},
 265                               res['network'][0]['node'][i]['supporting-node'])
 266                 self.assertEqual(res['network'][0]['node'][i]
 267                                  ['org-openroadm-otn-network-topology:xpdr-attributes']['xpdr-number'], 2)
 268                 nbTps = len(res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
 269                 client = 0
 270                 network = 0
 271                 for j in range(0, nbTps):
 272                     tpType = res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]['org-openroadm-common-network:tp-type']
 273                     tpId = res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]['tp-id']
 274                     if (tpType == 'XPONDER-CLIENT'):
 275                         client += 1
 276                         self.assertEqual(res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]['org-openroadm-otn-network-topology:tp-supported-interfaces']
 277                                          ['supported-interface-capability'][0]['if-cap-type'], 'org-openroadm-port-types:if-100GE-ODU4')
 278                         self.assertEqual(res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]
 279                                          ['org-openroadm-otn-network-topology:xpdr-tp-port-connection-attributes']['rate'], 'org-openroadm-otn-common-types:ODU4')
 280                     elif (tpType == 'XPONDER-NETWORK'):
 281                         network += 1
 282                         self.assertEqual(res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]['org-openroadm-otn-network-topology:tp-supported-interfaces']
 283                                          ['supported-interface-capability'][0]['if-cap-type'], 'org-openroadm-port-types:if-OCH-OTU4-ODU4')
 284                         self.assertEqual(res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]
 285                                          ['org-openroadm-otn-network-topology:xpdr-tp-port-connection-attributes']['rate'], 'org-openroadm-otn-common-types:ODU4')
 286                         self.assertEqual(res['network'][0]['node'][i]['ietf-network-topology:termination-point']
 287                                          [j]['supporting-termination-point'][0]['network-ref'], 'openroadm-topology')
 288                         self.assertEqual(res['network'][0]['node'][i]['ietf-network-topology:termination-point']
 289                                          [j]['supporting-termination-point'][0]['node-ref'], 'SPDR-SA1-XPDR2')
 290                         self.assertEqual(res['network'][0]['node'][i]['ietf-network-topology:termination-point']
 291                                          [j]['supporting-termination-point'][0]['tp-ref'], tpId)
 292                 self.assertTrue(client == 4)
 293                 self.assertTrue(network == 4)
 294                 listNode.remove(nodeId)
 295                 nbNbl = len(res['network'][0]['node'][i]['org-openroadm-otn-network-topology:switching-pools']
 296                             ['odu-switching-pools'][0]['non-blocking-list'])
 297                 self.assertEqual(nbNbl, 1)
 298                 nbl = res['network'][0]['node'][i]['org-openroadm-otn-network-topology:switching-pools']['odu-switching-pools'][0]['non-blocking-list'][0]
 299                 self.assertIn('XPDR2-NETWORK1', nbl['tp-list'])
 300                 self.assertIn('XPDR2-CLIENT1', nbl['tp-list'])
 301                 self.assertIn('XPDR2-NETWORK2', nbl['tp-list'])
 302                 self.assertIn('XPDR2-CLIENT2', nbl['tp-list'])
 303                 self.assertIn('XPDR2-NETWORK3', nbl['tp-list'])
 304                 self.assertIn('XPDR2-CLIENT3', nbl['tp-list'])
 305                 self.assertIn('XPDR2-NETWORK4', nbl['tp-list'])
 306                 self.assertIn('XPDR2-CLIENT4', nbl['tp-list'])
 307             else:
 308                 self.assertFalse(True)
 309         self.assertEqual(len(listNode), 0)
 310
 311     def test_08_disconnect_SPDR_SA1(self):
 312         url = ("{}/config/network-topology:"
 313                "network-topology/topology/topology-netconf/node/SPDR-SA1"
 314                .format(self.restconf_baseurl))
 315         data = {}
 316         headers = {'content-type': 'application/json'}
 317         response = requests.request(
 318             "DELETE", url, data=json.dumps(data), headers=headers,
 319             auth=('admin', 'admin'))
 320         self.assertEqual(response.status_code, requests.codes.ok)
 321
 322     def test_09_getClliNetwork(self):
 323         url = ("{}/config/ietf-network:networks/network/clli-network"
 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         nbNode = len(res['network'][0]['node'])
 331         self.assertEqual(nbNode, 1)
 332         self.assertEqual(res['network'][0]['node'][0]['org-openroadm-clli-network:clli'], 'NodeSA')
 333
 334     def test_10_getOpenRoadmNetwork(self):
 335         url = ("{}/config/ietf-network:networks/network/openroadm-network"
 336                .format(self.restconf_baseurl))
 337         headers = {'content-type': 'application/json'}
 338         response = requests.request(
 339             "GET", url, headers=headers, auth=('admin', 'admin'))
 340         self.assertEqual(response.status_code, requests.codes.ok)
 341         res = response.json()
 342         self.assertNotIn('node', res['network'][0])
 343
 344     def test_11_getNodes_OpenRoadmTopology(self):
 345         url = ("{}/config/ietf-network:networks/network/openroadm-topology"
 346                .format(self.restconf_baseurl))
 347         headers = {'content-type': 'application/json'}
 348         response = requests.request(
 349             "GET", url, headers=headers, auth=('admin', 'admin'))
 350         res = response.json()
 351         self.assertNotIn('node', res['network'][0])
 352
 353     def test_12_getNodes_OtnTopology(self):
 354         url = ("{}/config/ietf-network:networks/network/otn-topology"
 355                .format(self.restconf_baseurl))
 356         headers = {'content-type': 'application/json'}
 357         response = requests.request(
 358             "GET", url, headers=headers, auth=('admin', 'admin'))
 359         res = response.json()
 360         self.assertNotIn('node', res['network'][0])
 361
 362
 363 if __name__ == "__main__":
 364     unittest.main(verbosity=2)