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