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-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-network-topology: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.assertEqual(nodeType, 'XPONDER')
 147                     nbTps = len(res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
 148                     client = 0
 149                     network = 0
 150                     for j in range(0, nbTps):
 151                         tpType = res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]['org-openroadm-network-topology:tp-type']
 152                         tpId = res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]['tp-id']
 153                         if (tpType == 'XPONDER-CLIENT'):
 154                             client += 1
 155                         elif (tpType == 'XPONDER-NETWORK'):
 156                             network += 1
 157                     self.assertTrue(client == 4)
 158                     self.assertTrue(network == 1)
 159                     listNode.remove(nodeId)
 160                 if(nodeId == 'SPDR-SA1-XPDR2'):
 161                     self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'SPDR-SA1'},
 162                                   res['network'][0]['node'][i]['supporting-node'])
 163                     self.assertEqual(nodeType, 'XPONDER')
 164                     nbTps = len(res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
 165                     client = 0
 166                     network = 0
 167                     for j in range(0, nbTps):
 168                         tpType = res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]['org-openroadm-network-topology:tp-type']
 169                         tpId = res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]['tp-id']
 170                         if (tpType == 'XPONDER-CLIENT'):
 171                             client += 1
 172                         elif (tpType == 'XPONDER-NETWORK'):
 173                             network += 1
 174                     self.assertTrue(client == 4)
 175                     self.assertTrue(network == 4)
 176                     listNode.remove(nodeId)
 177                 else:
 178                     self.assertFalse(True)
 179             self.assertEqual(len(listNode), 0)
 180
 181     def test_06_getLinks_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         self.assertEqual(response.status_code, requests.codes.ok)
 188         res = response.json()
 189         self.assertNotIn('ietf-network-topology:link', res['network'][0])
 190
 191     def test_07_getNodes_OtnTopology(self):
 192         url = ("{}/config/ietf-network:networks/network/otn-topology"
 193               .format(self.restconf_baseurl))
 194         headers = {'content-type': 'application/json'}
 195         response = requests.request(
 196             "GET", url, headers=headers, auth=('admin', 'admin'))
 197         res = response.json()
 198         self.assertEqual(response.status_code, requests.codes.ok)
 199         nbNode = len(res['network'][0]['node'])
 200         self.assertEqual(nbNode, 2)
 201         listNode = ['SPDR-SA1-XPDR1', 'SPDR-SA1-XPDR2']
 202         for i in range(0, nbNode):
 203             nodeId = res['network'][0]['node'][i]['node-id']
 204             if(nodeId == 'SPDR-SA1-XPDR1'):
 205                 nodeType = res['network'][0]['node'][i]['org-openroadm-otn-network-topology:node-type']
 206                 self.assertEqual(nodeType, 'MUXPDR')
 207                 self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'SPDR-SA1'},
 208                               res['network'][0]['node'][i]['supporting-node'])
 209                 self.assertIn({'network-ref': 'openroadm-topology', 'node-ref': 'SPDR-SA1-XPDR1'},
 210                               res['network'][0]['node'][i]['supporting-node'])
 211                 self.assertEqual(res['network'][0]['node'][i]['org-openroadm-otn-network-topology:xpdr-attributes']['xpdr-number'], 1)
 212                 nbTps = len(res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
 213                 client = 0
 214                 network = 0
 215                 for j in range(0, nbTps):
 216                     tpType = res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]['org-openroadm-otn-network-topology:tp-type']
 217                     tpId = res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]['tp-id']
 218                     if (tpType == 'XPONDER-CLIENT'):
 219                         client += 1
 220                         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')
 221                         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')
 222                     elif (tpType == 'XPONDER-NETWORK'):
 223                         network += 1
 224                         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')
 225                         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')
 226                         self.assertEqual(res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]['supporting-termination-point'][0]['network-ref'], 'openroadm-topology')
 227                         self.assertEqual(res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]['supporting-termination-point'][0]['node-ref'], 'SPDR-SA1-XPDR1')
 228                         self.assertEqual(res['network'][0]['node'][i]['ietf-network-topology:termination-point'][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']['odu-switching-pools'][0]['non-blocking-list'])
 233                 self.assertEqual(nbNbl, 4)
 234                 for k in range(0, nbNbl):
 235                     nbl = res['network'][0]['node'][i]['org-openroadm-otn-network-topology:switching-pools']['odu-switching-pools'][0]['non-blocking-list'][k]
 236                     if (nbl['nbl-number'] == 1):
 237                         self.assertEqual(nbl['available-interconnect-bandwidth'], 10)
 238                         self.assertEqual(nbl['interconnect-bandwidth-unit'], 1000000000)
 239                         self.assertIn('XPDR1-NETWORK1', nbl['tp-list'])
 240                         self.assertIn('XPDR1-CLIENT1', nbl['tp-list'])
 241                         self.assertNotIn('XPDR1-CLIENT2', nbl['tp-list'])
 242             elif(nodeId == 'SPDR-SA1-XPDR2'):
 243                 nodeType = res['network'][0]['node'][i]['org-openroadm-otn-network-topology:node-type']
 244                 self.assertEqual(nodeType, 'SWITCH')
 245                 self.assertIn({'network-ref': 'openroadm-network', 'node-ref': 'SPDR-SA1'},
 246                               res['network'][0]['node'][i]['supporting-node'])
 247                 self.assertIn({'network-ref': 'openroadm-topology', 'node-ref': 'SPDR-SA1-XPDR2'},
 248                               res['network'][0]['node'][i]['supporting-node'])
 249                 self.assertEqual(res['network'][0]['node'][i]['org-openroadm-otn-network-topology:xpdr-attributes']['xpdr-number'], 2)
 250                 nbTps = len(res['network'][0]['node'][i]['ietf-network-topology:termination-point'])
 251                 client = 0
 252                 network = 0
 253                 for j in range(0, nbTps):
 254                     tpType = res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]['org-openroadm-otn-network-topology:tp-type']
 255                     tpId = res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]['tp-id']
 256                     if (tpType == 'XPONDER-CLIENT'):
 257                         client += 1
 258                         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')
 259                         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')
 260                     elif (tpType == 'XPONDER-NETWORK'):
 261                         network += 1
 262                         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')
 263                         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')
 264                         self.assertEqual(res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]['supporting-termination-point'][0]['network-ref'], 'openroadm-topology')
 265                         self.assertEqual(res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]['supporting-termination-point'][0]['node-ref'], 'SPDR-SA1-XPDR2')
 266                         self.assertEqual(res['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]['supporting-termination-point'][0]['tp-ref'], tpId)
 267                 self.assertTrue(client == 4)
 268                 self.assertTrue(network == 4)
 269                 listNode.remove(nodeId)
 270                 nbNbl = len(res['network'][0]['node'][i]['org-openroadm-otn-network-topology:switching-pools']['odu-switching-pools'][0]['non-blocking-list'])
 271                 self.assertEqual(nbNbl, 1)
 272                 nbl = res['network'][0]['node'][i]['org-openroadm-otn-network-topology:switching-pools']['odu-switching-pools'][0]['non-blocking-list'][0]
 273                 self.assertIn('XPDR2-NETWORK1', nbl['tp-list'])
 274                 self.assertIn('XPDR2-CLIENT1', nbl['tp-list'])
 275                 self.assertIn('XPDR2-NETWORK2', nbl['tp-list'])
 276                 self.assertIn('XPDR2-CLIENT2', nbl['tp-list'])
 277                 self.assertIn('XPDR2-NETWORK3', nbl['tp-list'])
 278                 self.assertIn('XPDR2-CLIENT3', nbl['tp-list'])
 279                 self.assertIn('XPDR2-NETWORK4', nbl['tp-list'])
 280                 self.assertIn('XPDR2-CLIENT4', nbl['tp-list'])
 281             else:
 282                 self.assertFalse(True)
 283         self.assertEqual(len(listNode), 0)
 284
 285     def test_08_disconnect_SPDR_SA1(self):
 286         url = ("{}/config/network-topology:"
 287                "network-topology/topology/topology-netconf/node/SPDR-SA1"
 288               .format(self.restconf_baseurl))
 289         data = {}
 290         headers = {'content-type': 'application/json'}
 291         response = requests.request(
 292             "DELETE", url, data=json.dumps(data), headers=headers,
 293             auth=('admin', 'admin'))
 294         self.assertEqual(response.status_code, requests.codes.ok)
 295
 296     def test_09_getClliNetwork(self):
 297         url = ("{}/config/ietf-network:networks/network/clli-network"
 298               .format(self.restconf_baseurl))
 299         headers = {'content-type': 'application/json'}
 300         response = requests.request(
 301             "GET", url, headers=headers, auth=('admin', 'admin'))
 302         self.assertEqual(response.status_code, requests.codes.ok)
 303         res = response.json()
 304         nbNode = len(res['network'][0]['node'])
 305         self.assertEqual(nbNode, 1)
 306         self.assertEqual(res['network'][0]['node'][0]['org-openroadm-clli-network:clli'], 'NodeSA')
 307
 308     def test_10_getOpenRoadmNetwork(self):
 309         url = ("{}/config/ietf-network:networks/network/openroadm-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         self.assertNotIn('node', res['network'][0])
 317
 318     def test_11_getNodes_OpenRoadmTopology(self):
 319         url = ("{}/config/ietf-network:networks/network/openroadm-topology"
 320               .format(self.restconf_baseurl))
 321         headers = {'content-type': 'application/json'}
 322         response = requests.request(
 323             "GET", url, headers=headers, auth=('admin', 'admin'))
 324         res = response.json()
 325         self.assertNotIn('node', res['network'][0])
 326
 327     def test_12_getNodes_OtnTopology(self):
 328         url = ("{}/config/ietf-network:networks/network/otn-topology"
 329               .format(self.restconf_baseurl))
 330         headers = {'content-type': 'application/json'}
 331         response = requests.request(
 332             "GET", url, headers=headers, auth=('admin', 'admin'))
 333         res = response.json()
 334         self.assertNotIn('node', res['network'][0])
 335
 336
 337 if __name__ == "__main__":
 338     unittest.main(verbosity=2)