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