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