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 openroadm-network
 167         url = ("{}/config/ietf-network:network/openroadm-network/node/ROADMA"
 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         time.sleep(5)
 176
 177     #Connect the XPDRA
 178     def test_04_connect_xpdr(self):
 179          #Config XPDRA
 180          url = ("{}/config/network-topology:"
 181                  "network-topology/topology/topology-netconf/node/XPDRA"
 182                 .format(self.restconf_baseurl))
 183          data = {"node": [{
 184               "node-id": "XPDRA",
 185               "netconf-node-topology:username": "admin",
 186               "netconf-node-topology:password": "admin",
 187               "netconf-node-topology:host": "127.0.0.1",
 188               "netconf-node-topology:port": "17830",
 189               "netconf-node-topology:tcp-only": "false",
 190               "netconf-node-topology:pass-through": {}}]}
 191          headers = {'content-type': 'application/json'}
 192          response = requests.request(
 193               "PUT", url, data=json.dumps(data), headers=headers,
 194               auth=('admin', 'admin'))
 195          self.assertEqual(response.status_code, requests.codes.created)
 196          #seems sometimes to return 200 instead of 201
 197          #self.assertEqual(response.status_code, requests.codes.ok)
 198          time.sleep(15)
 199
 200     #Verify the termination points related to XPDR
 201     def test_05_compareOpenroadmTopologyPortMapping(self):
 202         nbXPDR=1
 203         for p in(1,nbXPDR+1):
 204             if(p > nbXPDR):
 205                 break;
 206             url_topo = "{}/config/ietf-network:network/openroadm-topology/node/XPDRA-XPDR"+`p`
 207             with open('./transportpce_tests/log/topoPortMap.log', 'a') as outfile1:
 208                 outfile1.write('Config: '+`p`+' : '+url_topo+'\n')
 209             url = (url_topo.format(self.restconf_baseurl))
 210             headers = {'content-type': 'application/json'}
 211             response_topo = requests.request(
 212                 "GET", url, headers=headers, auth=('admin', 'admin'))
 213             self.assertEqual(response_topo.status_code, requests.codes.ok)
 214             res_topo = response_topo.json()
 215             nbTP=len(res_topo['node'][0]['ietf-network-topology:termination-point'])
 216             for i in range(0,nbTP):
 217                 tp_id=res_topo['node'][0]['ietf-network-topology:termination-point'][i]['tp-id']
 218                 url_map = "{}/config/portmapping:network/nodes/XPDRA/mapping/"+tp_id
 219                 with open('./transportpce_tests/log/topoPortMap.log', 'a') as outfile1:
 220                     outfile1.write('Config: '+`i`+'/'+ `nbTP`+' : '+url_map+'\n')
 221                 url = url_map.format(self.restconf_baseurl)
 222                 headers = {'content-type': 'application/json'}
 223                 response_portMap = requests.request(
 224                     "GET", url, headers=headers, auth=('admin', 'admin'))
 225                 self.assertEqual(response_portMap.status_code, requests.codes.ok)
 226                 if("CLIENT" in tp_id):
 227                     #Verify the tail equipment id of the client
 228                     xpdr_client=res_topo['node'][0]['ietf-network-topology:termination-point'][i]["org-openroadm-network-topology:xpdr-client-attributes"]["tail-equipment-id"]
 229                     url_map = "{}/config/portmapping:network/nodes/XPDRA/mapping/"+xpdr_client
 230                     with open('./transportpce_tests/log/topoPortMap.log', 'a') as outfile1:
 231                         outfile1.write('Config: '+`i`+'/'+ `nbTP`+' : '+xpdr_client+'\n')
 232                     url = url_map.format(self.restconf_baseurl)
 233                     headers = {'content-type': 'application/json'}
 234                     response_xpdrClient = requests.request(
 235                         "GET", url, headers=headers, auth=('admin', 'admin'))
 236                     self.assertEqual(response_xpdrClient.status_code, requests.codes.ok)
 237                 if("NETWORK" in tp_id):
 238                     #Verify the tail equipment id of the network
 239                     xpdr_network=res_topo['node'][0]['ietf-network-topology:termination-point'][i]["org-openroadm-network-topology:xpdr-network-attributes"]["tail-equipment-id"]
 240                     url_map = "{}/config/portmapping:network/nodes/XPDRA/mapping/"+xpdr_network
 241                     with open('./transportpce_tests/log/topoPortMap.log', 'a') as outfile1:
 242                         outfile1.write('Config: '+`i`+'/'+ `nbTP`+' : '+xpdr_network+'\n')
 243                     url = url_map.format(self.restconf_baseurl)
 244                     headers = {'content-type': 'application/json'}
 245                     response_xpdrNetwork = requests.request(
 246                         "GET", url, headers=headers, auth=('admin', 'admin'))
 247                     self.assertEqual(response_xpdrNetwork.status_code, requests.codes.ok)
 248
 249     #Disconnect the XPDRA
 250     def test_06_disconnect_device(self):
 251         url = ("{}/config/network-topology:"
 252                "network-topology/topology/topology-netconf/node/XPDRA"
 253               .format(self.restconf_baseurl))
 254         data = {}
 255         headers = {'content-type': 'application/json'}
 256         response = requests.request(
 257             "DELETE", url, data=json.dumps(data), headers=headers,
 258             auth=('admin', 'admin'))
 259         self.assertEqual(response.status_code, requests.codes.ok)
 260         #Delete in the openroadm-network
 261         url = ("{}/config/ietf-network:network/openroadm-network/node/XPDRA"
 262                .format(self.restconf_baseurl))
 263         data = {}
 264         headers = {'content-type': 'application/json'}
 265         response = requests.request(
 266              "DELETE", url, data=json.dumps(data), headers=headers,
 267              auth=('admin', 'admin'))
 268         self.assertEqual(response.status_code, requests.codes.ok)
 269
 270
 271 if __name__ == "__main__":
 272     #logging.basicConfig(filename='./transportpce_tests/log/response.log',filemode='w',level=logging.DEBUG)
 273     #logging.debug('I am there')
 274     unittest.main(verbosity=2)