tests/transportpce_tests/2.2.1/test04_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 # pylint: disable=wrong-import-order
  20 import sys
  21 sys.path.append('transportpce_tests/common/')
  22 # pylint: disable=wrong-import-position
  23 # pylint: disable=import-error
  24 import test_utils  # nopep8
  25
  26
  27 class TransportPCEtesting(unittest.TestCase):
  28
  29     processes = None
  30     NODE_VERSION = '2.2.1'
  31
  32     @classmethod
  33     def setUpClass(cls):
  34         cls.processes = test_utils.start_tpce()
  35         cls.processes = test_utils.start_sims([('spdra', cls.NODE_VERSION)])
  36
  37     @classmethod
  38     def tearDownClass(cls):
  39         # pylint: disable=not-an-iterable
  40         for process in cls.processes:
  41             test_utils.shutdown_process(process)
  42         print("all processes killed")
  43
  44     def setUp(self):
  45         time.sleep(5)
  46
  47     def test_01_connect_SPDR_SA1(self):
  48         response = test_utils.mount_device("SPDR-SA1", ('spdra', self.NODE_VERSION))
  49         self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
  50         time.sleep(10)
  51
  52         response = test_utils.check_device_connection("SPDR-SA1")
  53         self.assertEqual(response['status_code'], requests.codes.ok)
  54         self.assertEqual(response['connection-status'], 'connected')
  55
  56     def test_02_getClliNetwork(self):
  57         response = test_utils.get_ietf_network_request('clli-network', 'config')
  58         self.assertEqual(response['status_code'], requests.codes.ok)
  59         logging.info(response)
  60         self.assertEqual(response['network'][0]['node'][0]['node-id'], 'NodeSA')
  61         self.assertEqual(response['network'][0]['node'][0]['org-openroadm-clli-network:clli'], 'NodeSA')
  62
  63     def test_03_getOpenRoadmNetwork(self):
  64         response = test_utils.get_ietf_network_request('openroadm-network', 'config')
  65         self.assertEqual(response['status_code'], requests.codes.ok)
  66         self.assertEqual(response['network'][0]['node'][0]['node-id'], 'SPDR-SA1')
  67         self.assertEqual(response['network'][0]['node'][0]['supporting-node'][0]['network-ref'], 'clli-network')
  68         self.assertEqual(response['network'][0]['node'][0]['supporting-node'][0]['node-ref'], 'NodeSA')
  69         self.assertEqual(response['network'][0]['node'][0]['org-openroadm-common-network:node-type'], 'XPONDER')
  70         self.assertEqual(response['network'][0]['node'][0]['org-openroadm-network:model'], 'universal-switchponder')
  71         self.assertEqual(response['network'][0]['node'][0]['org-openroadm-network:vendor'], 'vendorA')
  72         self.assertEqual(response['network'][0]['node'][0]['org-openroadm-network:ip'], '1.2.3.4')
  73
  74     def test_04_getLinks_OpenroadmTopology(self):
  75         response = test_utils.get_ietf_network_request('openroadm-topology', 'config')
  76         self.assertEqual(response['status_code'], requests.codes.ok)
  77         self.assertNotIn('ietf-network-topology:link', response['network'][0])
  78
  79     def test_05_getNodes_OpenRoadmTopology(self):
  80         # pylint: disable=redundant-unittest-assert
  81         response = test_utils.get_ietf_network_request('openroadm-topology', 'config')
  82         self.assertEqual(response['status_code'], requests.codes.ok)
  83         self.assertIn('node', response['network'][0])
  84         self.assertEqual(len(response['network'][0]['node']), 3)
  85         listNode = ['SPDR-SA1-XPDR1', 'SPDR-SA1-XPDR2', 'SPDR-SA1-XPDR3']
  86         for node in response['network'][0]['node']:
  87             self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'SPDR-SA1'}, node['supporting-node'])
  88             self.assertIn({'network-ref': 'clli-network', 'node-ref': 'NodeSA'}, node['supporting-node'])
  89             nodeType = node['org-openroadm-common-network:node-type']
  90             nodeId = node['node-id']
  91             if nodeId not in listNode:
  92                 self.assertFalse(True)
  93                 continue
  94             self.assertEqual(nodeType, 'XPONDER')
  95             client = 0
  96             network = 0
  97             for tp in node['ietf-network-topology:termination-point']:
  98                 tpType = tp['org-openroadm-common-network:tp-type']
  99                 if tpType == 'XPONDER-CLIENT':
 100                     client += 1
 101                 elif tpType == 'XPONDER-NETWORK':
 102                     network += 1
 103             self.assertTrue(client == 0)
 104             if nodeId in ('SPDR-SA1-XPDR1', 'SPDR-SA1-XPDR3'):
 105                 self.assertTrue(network == 1)
 106             else:
 107                 # elif nodeId == 'SPDR-SA1-XPDR2':
 108                 self.assertTrue(network == 4)
 109             listNode.remove(nodeId)
 110         self.assertEqual(len(listNode), 0)
 111
 112     def test_06_getLinks_OtnTopology(self):
 113         response = test_utils.get_ietf_network_request('otn-topology', 'config')
 114         self.assertEqual(response['status_code'], requests.codes.ok)
 115         self.assertNotIn('ietf-network-topology:link', response['network'][0])
 116
 117     def test_07_getNodes_OtnTopology(self):
 118         # pylint: disable=redundant-unittest-assert
 119         response = test_utils.get_ietf_network_request('otn-topology', 'config')
 120         self.assertEqual(response['status_code'], requests.codes.ok)
 121         self.assertEqual(len(response['network'][0]['node']), 3)
 122         listNode = ['SPDR-SA1-XPDR1', 'SPDR-SA1-XPDR2', 'SPDR-SA1-XPDR3']
 123         CHECK_LIST = {
 124             'SPDR-SA1-XPDR1': {
 125                 'node-type': 'MUXPDR',
 126                 'xpdr-number': 1,
 127                 'port-types': ['org-openroadm-port-types:if-10GE-ODU2', 'org-openroadm-port-types:if-10GE-ODU2e',
 128                                'org-openroadm-port-types:if-10GE'],
 129                 'otn-common-types': 'org-openroadm-otn-common-types:ODU2e',
 130                 'network_nb': 1,
 131                 'nbl_nb': 4,
 132                 'tp-checklist': ['XPDR1-NETWORK1', 'XPDR1-CLIENT1'],
 133                 'tp-unchecklist': ['XPDR1-CLIENT2']
 134             },
 135             'SPDR-SA1-XPDR2': {
 136                 'node-type': 'SWITCH',
 137                 'xpdr-number': 2,
 138                 'port-types': 'org-openroadm-port-types:if-100GE-ODU4',
 139                 'otn-common-types': 'org-openroadm-otn-common-types:ODU4',
 140                 'network_nb': 4,
 141                 'nbl_nb': 1,
 142                 'tp-checklist': ['XPDR2-NETWORK4', 'XPDR2-CLIENT1', 'XPDR2-NETWORK3', 'XPDR2-CLIENT4',
 143                                  'XPDR2-CLIENT2', 'XPDR2-NETWORK2', 'XPDR2-CLIENT3', 'XPDR2-NETWORK1'],
 144                 'tp-unchecklist': []
 145             }
 146         }
 147         for node in response['network'][0]['node']:
 148             nodeId = node['node-id']
 149             if nodeId in CHECK_LIST:
 150                 self.assertEqual(node['org-openroadm-common-network:node-type'], CHECK_LIST[nodeId]['node-type'])
 151                 self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'SPDR-SA1'},
 152                               node['supporting-node'])
 153                 self.assertIn({'network-ref': 'openroadm-topology', 'node-ref': nodeId},
 154                               node['supporting-node'])
 155                 self.assertIn({'network-ref': 'clli-network', 'node-ref': 'NodeSA'},
 156                               node['supporting-node'])
 157                 self.assertEqual(node['org-openroadm-otn-network-topology:xpdr-attributes']['xpdr-number'],
 158                                  CHECK_LIST[nodeId]['xpdr-number'])
 159                 client = 0
 160                 network = 0
 161                 for tp in node['ietf-network-topology:termination-point']:
 162                     tpType = tp['org-openroadm-common-network:tp-type']
 163                     tpId = tp['tp-id']
 164                     if tpType == 'XPONDER-CLIENT':
 165                         client += 1
 166                         # pylint: disable=consider-using-f-string
 167                         print("tpId = {}".format(tpId))
 168                         print("tp= {}".format(tp))
 169                         # pylint: disable=line-too-long
 170                         for capa in tp['org-openroadm-otn-network-topology:tp-supported-interfaces']['supported-interface-capability']:
 171                             self.assertIn((capa['if-cap-type']), CHECK_LIST[nodeId]['port-types'])
 172                     elif tpType == 'XPONDER-NETWORK':
 173                         network += 1
 174                         self.assertEqual((tp['org-openroadm-otn-network-topology:tp-supported-interfaces']
 175                                           ['supported-interface-capability'][0]['if-cap-type']),
 176                                          'org-openroadm-port-types:if-OCH-OTU4-ODU4')
 177                         self.assertEqual((tp['org-openroadm-otn-network-topology:xpdr-tp-port-connection-attributes']
 178                                           ['rate']),
 179                                          'org-openroadm-otn-common-types:ODU4')
 180                         self.assertEqual((tp['supporting-termination-point'][0]['network-ref']), 'openroadm-topology')
 181                         self.assertEqual((tp['supporting-termination-point'][0]['node-ref']), nodeId)
 182                         self.assertEqual((tp['supporting-termination-point'][0]['tp-ref']), tpId)
 183                 self.assertTrue(client == 4)
 184                 self.assertTrue(network == CHECK_LIST[nodeId]['network_nb'])
 185                 listNode.remove(nodeId)
 186                 self.assertEqual(
 187                     len(node['org-openroadm-otn-network-topology:switching-pools']
 188                         ['odu-switching-pools'][0]['non-blocking-list']),
 189                     CHECK_LIST[nodeId]['nbl_nb'])
 190                 # pylint: disable=line-too-long
 191                 for nbl in node['org-openroadm-otn-network-topology:switching-pools']['odu-switching-pools'][0]['non-blocking-list']:
 192                     if nbl['nbl-number'] == 1:
 193                         if nodeId == 'SPDR-SA1-XPDR1':
 194                             self.assertEqual(nbl['available-interconnect-bandwidth'], 10)
 195                             self.assertEqual(nbl['interconnect-bandwidth-unit'], 1000000000)
 196                         for tp in CHECK_LIST[nodeId]['tp-checklist']:
 197                             self.assertIn(tp, nbl['tp-list'])
 198                         for tp in CHECK_LIST[nodeId]['tp-unchecklist']:
 199                             self.assertNotIn(tp, nbl['tp-list'])
 200             else:
 201                 self.assertEqual('SPDR-SA1-XPDR3', nodeId)
 202                 listNode.remove(nodeId)
 203         self.assertEqual(len(listNode), 0)
 204
 205     def test_08_disconnect_SPDR_SA1(self):
 206         response = test_utils.unmount_device("SPDR-SA1")
 207         self.assertIn(response.status_code, (requests.codes.ok, requests.codes.no_content))
 208
 209     def test_09_getClliNetwork(self):
 210         response = test_utils.get_ietf_network_request('clli-network', 'config')
 211         self.assertEqual(response['status_code'], requests.codes.ok)
 212         self.assertEqual(len(response['network'][0]['node']), 1)
 213         self.assertEqual(response['network'][0]['node'][0]['org-openroadm-clli-network:clli'], 'NodeSA')
 214
 215     def test_10_getOpenRoadmNetwork(self):
 216         response = test_utils.get_ietf_network_request('openroadm-network', 'config')
 217         self.assertEqual(response['status_code'], requests.codes.ok)
 218         self.assertNotIn('node', response['network'][0])
 219
 220     def test_11_getNodes_OpenRoadmTopology(self):
 221         response = test_utils.get_ietf_network_request('openroadm-topology', 'config')
 222         self.assertNotIn('node', response['network'][0])
 223
 224     def test_12_getNodes_OtnTopology(self):
 225         response = test_utils.get_ietf_network_request('otn-topology', 'config')
 226         self.assertNotIn('node', response['network'][0])
 227
 228
 229 if __name__ == "__main__":
 230     unittest.main(verbosity=2)