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