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