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