tests/transportpce_tests/test_topoPortMapping.py

   1 #!/usr/bin/env python
   2
   3 ##############################################################################
   4 #Copyright (c) 2017 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
  23 class TransportPCEtesting(unittest.TestCase):
  24
  25     testtools_process1 = None
  26     testtools_process2 = None
  27     odl_process = None
  28     restconf_baseurl = "http://localhost:8181/restconf"
  29
  30     @classmethod
  31     def __start_testtools(cls):
  32         executable = ("./netconf/netconf/tools/netconf-testtool/target/"
  33                       "netconf-testtool-1.5.0-executable.jar")
  34         if os.path.isfile(executable):
  35             if not os.path.exists("transportpce_tests/log"):
  36                 os.makedirs("transportpce_tests/log")
  37             if os.path.isfile("./transportpce_tests/log/topoPortMap.log"):
  38                 os.remove("transportpce_tests/log/topoPortMap.log")
  39             with open('transportpce_tests/log/testtools_ROADMA.log', 'w') as outfile1:
  40                 cls.testtools_process1 = subprocess.Popen(
  41                     ["java", "-jar", executable, "--schemas-dir", "schemas",
  42                      "--initial-config-xml", "sample_configs/nodes_config/sample-config-ROADMA.xml","--starting-port","17831"],
  43                     stdout=outfile1)
  44             with open('transportpce_tests/log/testtools_XPDRA.log', 'w') as outfile2:
  45                 cls.testtools_process2 = subprocess.Popen(
  46                     ["java", "-jar", executable, "--schemas-dir", "schemas",
  47                      "--initial-config-xml", "sample_configs/nodes_config/sample-config-XPDRA.xml","--starting-port","17830"],
  48                     stdout=outfile2)
  49
  50     @classmethod
  51     def __start_odl(cls):
  52         executable = "../karaf/target/assembly/bin/karaf"
  53         with open('transportpce_tests/log/odl.log', 'w') as outfile:
  54             cls.odl_process = subprocess.Popen(
  55                 ["bash", executable, "server"], stdout=outfile,
  56                 stdin=open(os.devnull))
  57
  58     @classmethod
  59     def setUpClass(cls):
  60         cls.__start_testtools()
  61         cls.__start_odl()
  62         time.sleep(40)
  63
  64     @classmethod
  65     def tearDownClass(cls):
  66         cls.testtools_process1.send_signal(signal.SIGINT)
  67         cls.testtools_process1.wait()
  68         cls.testtools_process2.send_signal(signal.SIGINT)
  69         cls.testtools_process2.wait()
  70         for child in psutil.Process(cls.odl_process.pid).children():
  71             child.send_signal(signal.SIGINT)
  72             child.wait()
  73         cls.odl_process.send_signal(signal.SIGINT)
  74         cls.odl_process.wait()
  75
  76     def setUp(self):
  77         time.sleep(10)
  78
  79     #Connect the ROADMA
  80     def test_01_connect_roadma(self):
  81         #Config ROADMA
  82         url = ("{}/config/network-topology:"
  83                 "network-topology/topology/topology-netconf/node/ROADMA"
  84                .format(self.restconf_baseurl))
  85         data = {"node": [{
  86              "node-id": "ROADMA",
  87              "netconf-node-topology:username": "admin",
  88              "netconf-node-topology:password": "admin",
  89              "netconf-node-topology:host": "127.0.0.1",
  90              "netconf-node-topology:port": "17831",
  91              "netconf-node-topology:tcp-only": "false",
  92              "netconf-node-topology:pass-through": {}}]}
  93         headers = {'content-type': 'application/json'}
  94         response = requests.request(
  95              "PUT", url, data=json.dumps(data), headers=headers,
  96              auth=('admin', 'admin'))
  97         self.assertEqual(response.status_code, requests.codes.created)
  98         #seems sometimes to return 200 instead of 201
  99         #self.assertEqual(response.status_code, requests.codes.ok)
 100         time.sleep(10)
 101
 102     #Verify the termination points of the ROADMA
 103     def test_02_compareOpenroadmTopologyPortMapping(self):
 104         #Verify the termination points related to the SRGs
 105         nbSrg=1
 106         for s in range(1,nbSrg+1):
 107             url_topo="{}/config/ietf-network:network/openroadm-topology/node/ROADMA-SRG"+`s`
 108             with open('./transportpce_tests/log/topoPortMap.log', 'a') as outfile1:
 109                 outfile1.write('Config: '+`s`+' : '+url_topo+'\n')
 110             url = (url_topo.format(self.restconf_baseurl))
 111             headers = {'content-type': 'application/json'}
 112             response_topo = requests.request(
 113                 "GET", url, headers=headers, auth=('admin', 'admin'))
 114             self.assertEqual(response_topo.status_code, requests.codes.ok)
 115             res_topo = response_topo.json()
 116             nbTP=len(res_topo['node'][0]['ietf-network-topology:termination-point'])
 117             for i in range(0,nbTP):
 118                 tp_id=res_topo['node'][0]['ietf-network-topology:termination-point'][i]['tp-id']
 119                 if(not "CP" in tp_id):
 120                     url_map="{}/config/portmapping:network/nodes/ROADMA/mapping/"+tp_id
 121                     with open('./transportpce_tests/log/topoPortMap.log', 'a') as outfile1:
 122                         outfile1.write('Config: '+`i`+'/'+ `nbTP`+' : '+url_map+'\n')
 123                     url = (url_map.format(self.restconf_baseurl))
 124                     headers = {'content-type': 'application/json'}
 125                     response_portMap = requests.request(
 126                         "GET", url, headers=headers, auth=('admin', 'admin'))
 127                     self.assertEqual(response_portMap.status_code, requests.codes.ok)
 128
 129         #Verify the termination points related to the degrees
 130         nbDeg=2
 131         for d in range(1,nbDeg+1):
 132             url_topo="{}/config/ietf-network:network/openroadm-topology/node/ROADMA-DEG"+`d`
 133             with open('./transportpce_tests/log/topoPortMap.log', 'a') as outfile1:
 134                 outfile1.write(url_topo+'\n')
 135             url = (url_topo.format(self.restconf_baseurl))
 136             headers = {'content-type': 'application/json'}
 137             response_topo = requests.request(
 138                 "GET", url, headers=headers, auth=('admin', 'admin'))
 139             self.assertEqual(response_topo.status_code, requests.codes.ok)
 140             res_topo = response_topo.json()
 141             nbTP=len(res_topo['node'][0]['ietf-network-topology:termination-point'])
 142             for i in range(0,nbTP):
 143                 tp_id=res_topo['node'][0]['ietf-network-topology:termination-point'][i]['tp-id']
 144                 if(not "CTP" in tp_id):
 145                     url_map ="{}/config/portmapping:network/nodes/ROADMA/mapping/"+tp_id
 146                     with open('./transportpce_tests/log/topoPortMap.log', 'a') as outfile1:
 147                         outfile1.write('Config: '+`i`+'/'+ `nbTP`+' : '+url_map+'\n')
 148                     url = (url_map.format(self.restconf_baseurl))
 149                     headers = {'content-type': 'application/json'}
 150                     response_portMap = requests.request(
 151                         "GET", url, headers=headers, auth=('admin', 'admin'))
 152                     self.assertEqual(response_portMap.status_code, requests.codes.ok)
 153         time.sleep(1)
 154
 155     #Disconnect the ROADMA
 156     def test_03_disconnect_device(self):
 157         url = ("{}/config/network-topology:"
 158                 "network-topology/topology/topology-netconf/node/ROADMA"
 159                .format(self.restconf_baseurl))
 160         data = {}
 161         headers = {'content-type': 'application/json'}
 162         response = requests.request(
 163              "DELETE", url, data=json.dumps(data), headers=headers,
 164              auth=('admin', 'admin'))
 165         self.assertEqual(response.status_code, requests.codes.ok)
 166         #Delete in the clli-network
 167         url = ("{}/config/ietf-network:network/clli-network/node/NodeA"
 168                .format(self.restconf_baseurl))
 169         data = {}
 170         headers = {'content-type': 'application/json'}
 171         response = requests.request(
 172              "DELETE", url, data=json.dumps(data), headers=headers,
 173              auth=('admin', 'admin'))
 174         self.assertEqual(response.status_code, requests.codes.ok)
 175         #Delete in the openroadm-network
 176         url = ("{}/config/ietf-network:network/openroadm-network/node/ROADMA"
 177                .format(self.restconf_baseurl))
 178         data = {}
 179         headers = {'content-type': 'application/json'}
 180         response = requests.request(
 181              "DELETE", url, data=json.dumps(data), headers=headers,
 182              auth=('admin', 'admin'))
 183         self.assertEqual(response.status_code, requests.codes.ok)
 184         time.sleep(5)
 185
 186     #Connect the XPDRA
 187     def test_04_connect_xpdr(self):
 188          #Config XPDRA
 189          url = ("{}/config/network-topology:"
 190                  "network-topology/topology/topology-netconf/node/XPDRA"
 191                 .format(self.restconf_baseurl))
 192          data = {"node": [{
 193               "node-id": "XPDRA",
 194               "netconf-node-topology:username": "admin",
 195               "netconf-node-topology:password": "admin",
 196               "netconf-node-topology:host": "127.0.0.1",
 197               "netconf-node-topology:port": "17830",
 198               "netconf-node-topology:tcp-only": "false",
 199               "netconf-node-topology:pass-through": {}}]}
 200          headers = {'content-type': 'application/json'}
 201          response = requests.request(
 202               "PUT", url, data=json.dumps(data), headers=headers,
 203               auth=('admin', 'admin'))
 204          self.assertEqual(response.status_code, requests.codes.created)
 205          #seems sometimes to return 200 instead of 201
 206          #self.assertEqual(response.status_code, requests.codes.ok)
 207          time.sleep(15)
 208
 209     #Verify the termination points related to XPDR
 210     def test_05_compareOpenroadmTopologyPortMapping(self):
 211         nbXPDR=1
 212         for p in(1,nbXPDR+1):
 213             url_topo = "{}/config/ietf-network:network/openroadm-topology/node/XPDRA-XPDR"+`p`
 214             with open('./transportpce_tests/log/topoPortMap.log', 'a') as outfile1:
 215                 outfile1.write('Config: '+`p`+' : '+url_topo+'\n')
 216             url = (url_topo.format(self.restconf_baseurl))
 217             headers = {'content-type': 'application/json'}
 218             response_topo = requests.request(
 219                 "GET", url, headers=headers, auth=('admin', 'admin'))
 220             self.assertEqual(response_topo.status_code, requests.codes.ok)
 221             res_topo = response_topo.json()
 222             nbTP=len(res_topo['node'][0]['ietf-network-topology:termination-point'])
 223             for i in range(0,nbTP):
 224                 tp_id=res_topo['node'][0]['ietf-network-topology:termination-point'][i]['tp-id']
 225                 url_map = "{}/config/portmapping:network/nodes/XPDRA/mapping/"+tp_id
 226                 with open('./transportpce_tests/log/topoPortMap.log', 'a') as outfile1:
 227                     outfile1.write('Config: '+`i`+'/'+ `nbTP`+' : '+url_map+'\n')
 228                 url = url_map.format(self.restconf_baseurl)
 229                 headers = {'content-type': 'application/json'}
 230                 response_portMap = requests.request(
 231                     "GET", url, headers=headers, auth=('admin', 'admin'))
 232                 self.assertEqual(response_portMap.status_code, requests.codes.ok)
 233                 #Verify the tail equipment id of the client
 234                 xpdr_client=res_topo['node'][0]['ietf-network-topology:termination-point'][i]["org-openroadm-network-topology:xpdr-client-attributes"]["tail-equipment-id"]
 235                 url_map = "{}/config/portmapping:network/nodes/XPDRA/mapping/"+xpdr_client
 236                 with open('./transportpce_tests/log/topoPortMap.log', 'a') as outfile1:
 237                     outfile1.write('Config: '+`i`+'/'+ `nbTP`+' : '+xpdr_client+'\n')
 238                 url = url_map.format(self.restconf_baseurl)
 239                 headers = {'content-type': 'application/json'}
 240                 response_xpdrClient = requests.request(
 241                     "GET", url, headers=headers, auth=('admin', 'admin'))
 242                 self.assertEqual(response_xpdrClient.status_code, requests.codes.ok)
 243                 #Verify the tail equipment id of the network
 244                 xpdr_network=res_topo['node'][0]['ietf-network-topology:termination-point'][i]["org-openroadm-network-topology:xpdr-network-attributes"]["tail-equipment-id"]
 245                 url_map = "{}/config/portmapping:network/nodes/XPDRA/mapping/"+xpdr_network
 246                 with open('./transportpce_tests/log/topoPortMap.log', 'a') as outfile1:
 247                     outfile1.write('Config: '+`i`+'/'+ `nbTP`+' : '+xpdr_network+'\n')
 248                 url = url_map.format(self.restconf_baseurl)
 249                 headers = {'content-type': 'application/json'}
 250                 response_xpdrNetwork = requests.request(
 251                     "GET", url, headers=headers, auth=('admin', 'admin'))
 252                 self.assertEqual(response_xpdrNetwork.status_code, requests.codes.ok)
 253
 254     #Disconnect the XPDRA
 255     def test_06_disconnect_device(self):
 256         url = ("{}/config/network-topology:"
 257                "network-topology/topology/topology-netconf/node/XPDRA"
 258               .format(self.restconf_baseurl))
 259         data = {}
 260         headers = {'content-type': 'application/json'}
 261         response = requests.request(
 262             "DELETE", url, data=json.dumps(data), headers=headers,
 263             auth=('admin', 'admin'))
 264         self.assertEqual(response.status_code, requests.codes.ok)
 265         #Delete in the openroadm-network
 266         url = ("{}/config/ietf-network:network/openroadm-network/node/XPDRA"
 267                .format(self.restconf_baseurl))
 268         data = {}
 269         headers = {'content-type': 'application/json'}
 270         response = requests.request(
 271              "DELETE", url, data=json.dumps(data), headers=headers,
 272              auth=('admin', 'admin'))
 273         self.assertEqual(response.status_code, requests.codes.ok)
 274
 275
 276 if __name__ == "__main__":
 277     #logging.basicConfig(filename='./transportpce_tests/log/response.log',filemode='w',level=logging.DEBUG)
 278     #logging.debug('I am there')
 279     unittest.main(verbosity=2)