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