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     def test_05_compareOpenroadmTopologyPortMapping(self):
 226         nbXPDR=1
 227         for p in(1,nbXPDR+1):
 228             if(p > nbXPDR):
 229                 break;
 230             url_topo = "{}/config/ietf-network:networks/network/openroadm-topology/node/XPDRA-XPDR"+`p`
 231             with open('./transportpce_tests/log/topoPortMap.log', 'a') as outfile1:
 232                 outfile1.write('Config: '+`p`+' : '+url_topo+'\n')
 233             url = (url_topo.format(self.restconf_baseurl))
 234             headers = {'content-type': 'application/json'}
 235             response_topo = requests.request(
 236                 "GET", url, headers=headers, auth=('admin', 'admin'))
 237             self.assertEqual(response_topo.status_code, requests.codes.ok)
 238             res_topo = response_topo.json()
 239             nbTP=len(res_topo['node'][0]['ietf-network-topology:termination-point'])
 240             for i in range(0,nbTP):
 241                 tp_id=res_topo['node'][0]['ietf-network-topology:termination-point'][i]['tp-id']
 242                 url_map = "{}/config/transportpce-portmapping:network/nodes/XPDRA/mapping/"+tp_id
 243                 with open('./transportpce_tests/log/topoPortMap.log', 'a') as outfile1:
 244                     outfile1.write('Config: '+`i`+'/'+ `nbTP`+' : '+url_map+'\n')
 245                 url = url_map.format(self.restconf_baseurl)
 246                 headers = {'content-type': 'application/json'}
 247                 response_portMap = requests.request(
 248                     "GET", url, headers=headers, auth=('admin', 'admin'))
 249                 self.assertEqual(response_portMap.status_code, requests.codes.ok)
 250                 if("CLIENT" in tp_id):
 251                     #Verify the tail equipment id of the client
 252                     xpdr_client=res_topo['node'][0]['ietf-network-topology:termination-point'][i]["org-openroadm-network-topology:xpdr-client-attributes"]["tail-equipment-id"]
 253                     url_map = "{}/config/transportpce-portmapping:network/nodes/XPDRA/mapping/"+xpdr_client
 254                     with open('./transportpce_tests/log/topoPortMap.log', 'a') as outfile1:
 255                         outfile1.write('Config: '+`i`+'/'+ `nbTP`+' : '+xpdr_client+'\n')
 256                     url = url_map.format(self.restconf_baseurl)
 257                     headers = {'content-type': 'application/json'}
 258                     response_xpdrClient = requests.request(
 259                         "GET", url, headers=headers, auth=('admin', 'admin'))
 260                     self.assertEqual(response_xpdrClient.status_code, requests.codes.ok)
 261                 if("NETWORK" in tp_id):
 262                     #Verify the tail equipment id of the network
 263                     xpdr_network=res_topo['node'][0]['ietf-network-topology:termination-point'][i]["org-openroadm-network-topology:xpdr-network-attributes"]["tail-equipment-id"]
 264                     url_map = "{}/config/transportpce-portmapping:network/nodes/XPDRA/mapping/"+xpdr_network
 265                     with open('./transportpce_tests/log/topoPortMap.log', 'a') as outfile1:
 266                         outfile1.write('Config: '+`i`+'/'+ `nbTP`+' : '+xpdr_network+'\n')
 267                     url = url_map.format(self.restconf_baseurl)
 268                     headers = {'content-type': 'application/json'}
 269                     response_xpdrNetwork = requests.request(
 270                         "GET", url, headers=headers, auth=('admin', 'admin'))
 271                     self.assertEqual(response_xpdrNetwork.status_code, requests.codes.ok)
 272
 273     #Disconnect the XPDRA
 274     def test_06_disconnect_device(self):
 275         url = ("{}/config/network-topology:"
 276                "network-topology/topology/topology-netconf/node/XPDRA"
 277               .format(self.restconf_baseurl))
 278         data = {}
 279         headers = {'content-type': 'application/json'}
 280         response = requests.request(
 281             "DELETE", url, data=json.dumps(data), headers=headers,
 282             auth=('admin', 'admin'))
 283         self.assertEqual(response.status_code, requests.codes.ok)
 284         #Delete in the openroadm-network
 285 #         url = ("{}/config/ietf-network:networks/network/openroadm-network/node/XPDRA"
 286 #                .format(self.restconf_baseurl))
 287 #         data = {}
 288 #         headers = {'content-type': 'application/json'}
 289 #         response = requests.request(
 290 #              "DELETE", url, data=json.dumps(data), headers=headers,
 291 #              auth=('admin', 'admin'))
 292 #         self.assertEqual(response.status_code, requests.codes.ok)
 293
 294
 295 if __name__ == "__main__":
 296     #logging.basicConfig(filename='./transportpce_tests/log/response.log',filemode='w',level=logging.DEBUG)
 297     #logging.debug('I am there')
 298     unittest.main(verbosity=2)