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