tests/transportpce_tests/2.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 import test_utils
  23
  24 class TransportPCEtesting(unittest.TestCase):
  25
  26     honeynode_process1 = None
  27     honeynode_process2 = None
  28     odl_process = None
  29     restconf_baseurl = "http://localhost:8181/restconf"
  30
  31 #START_IGNORE_XTESTING
  32
  33     @classmethod
  34     def setUpClass(cls):
  35         print ("starting honeynode1...")
  36         cls.honeynode_process1 = test_utils.start_xpdra_honeynode()
  37         time.sleep(20)
  38
  39         print ("starting honeynode2...")
  40         cls.honeynode_process2 = test_utils.start_roadma_honeynode()
  41         time.sleep(20)
  42         print ("all honeynodes started")
  43
  44         print ("starting opendaylight...")
  45         cls.odl_process = test_utils.start_tpce()
  46         time.sleep(60)
  47         print ("opendaylight started")
  48
  49     @classmethod
  50     def tearDownClass(cls):
  51         for child in psutil.Process(cls.odl_process.pid).children():
  52             child.send_signal(signal.SIGINT)
  53             child.wait()
  54         cls.odl_process.send_signal(signal.SIGINT)
  55         cls.odl_process.wait()
  56         for child in psutil.Process(cls.honeynode_process1.pid).children():
  57             child.send_signal(signal.SIGINT)
  58             child.wait()
  59         cls.honeynode_process1.send_signal(signal.SIGINT)
  60         cls.honeynode_process1.wait()
  61         for child in psutil.Process(cls.honeynode_process2.pid).children():
  62             child.send_signal(signal.SIGINT)
  63             child.wait()
  64         cls.honeynode_process2.send_signal(signal.SIGINT)
  65         cls.honeynode_process2.wait()
  66
  67     def setUp(self):
  68         time.sleep(10)
  69
  70 #END_IGNORE_XTESTING
  71
  72     #Connect the ROADMA
  73     def test_01_connect_rdm(self):
  74         #Config ROADMA
  75         url = ("{}/config/network-topology:"
  76                 "network-topology/topology/topology-netconf/node/ROADM-A1"
  77                .format(self.restconf_baseurl))
  78         data = {"node": [{
  79              "node-id": "ROADM-A1",
  80              "netconf-node-topology:username": "admin",
  81              "netconf-node-topology:password": "admin",
  82              "netconf-node-topology:host": "127.0.0.1",
  83              "netconf-node-topology:port": "17841",
  84              "netconf-node-topology:tcp-only": "false",
  85              "netconf-node-topology:pass-through": {}}]}
  86         headers = {'content-type': 'application/json'}
  87         response = requests.request(
  88              "PUT", url, data=json.dumps(data), headers=headers,
  89              auth=('admin', 'admin'))
  90         self.assertEqual(response.status_code, requests.codes.created)
  91         time.sleep(20)
  92
  93     #Verify the termination points of the ROADMA
  94     def test_02_compareOpenroadmTopologyPortMapping_rdm(self):
  95         urlTopo = ("{}/config/ietf-network:networks/network/openroadm-topology"
  96             .format(self.restconf_baseurl))
  97         headers = {'content-type': 'application/json'}
  98         responseTopo = requests.request(
  99             "GET", urlTopo, headers=headers, auth=('admin', 'admin'))
 100         resTopo = responseTopo.json()
 101         nbNode = len(resTopo['network'][0]['node'])
 102         nbMapCumul = 0
 103         nbMappings = 0
 104         for i in range(0, nbNode):
 105             nodeId = resTopo['network'][0]['node'][i]['node-id']
 106             print("nodeId={}".format(nodeId))
 107             nodeMapId = nodeId.split("-")[0] + "-" + nodeId.split("-")[1]
 108             print("nodeMapId={}".format(nodeMapId))
 109             urlMapList = "{}/config/transportpce-portmapping:network/nodes/" + nodeMapId
 110             urlMapListFull = urlMapList.format(self.restconf_baseurl)
 111             responseMapList = requests.request(
 112                         "GET", urlMapListFull, headers=headers, auth=('admin', 'admin'))
 113             resMapList = responseMapList.json()
 114
 115             nbMappings = len(resMapList['nodes'][0]['mapping']) - nbMapCumul
 116             nbTp = len(resTopo['network'][0]['node'][i]['ietf-network-topology:termination-point'])
 117             nbMapCurrent = 0
 118             for j in range(0, nbTp):
 119                 tpId = resTopo['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]['tp-id']
 120                 if((not "CP" in tpId) and (not "CTP" in tpId)):
 121                      urlMap = "{}/config/transportpce-portmapping:network/nodes/" + nodeMapId + "/mapping/" + tpId
 122                      urlMapFull = urlMap.format(self.restconf_baseurl)
 123                      responseMap = requests.request(
 124                         "GET", urlMapFull, headers=headers, auth=('admin', 'admin'))
 125                      self.assertEqual(responseMap.status_code, requests.codes.ok)
 126                      if(responseMap.status_code == requests.codes.ok):
 127                         nbMapCurrent += 1
 128             nbMapCumul += nbMapCurrent
 129         nbMappings -= nbMapCurrent
 130         self.assertEqual(nbMappings, 0)
 131
 132     #Disconnect the ROADMA
 133     def test_03_disconnect_rdm(self):
 134         url = ("{}/config/network-topology:"
 135                 "network-topology/topology/topology-netconf/node/ROADM-A1"
 136                .format(self.restconf_baseurl))
 137         data = {}
 138         headers = {'content-type': 'application/json'}
 139         response = requests.request(
 140              "DELETE", url, data=json.dumps(data), headers=headers,
 141              auth=('admin', 'admin'))
 142         self.assertEqual(response.status_code, requests.codes.ok)
 143
 144 #     #Connect the XPDRA
 145     def test_04_connect_xpdr(self):
 146          #Config XPDRA
 147          url = ("{}/config/network-topology:"
 148                  "network-topology/topology/topology-netconf/node/XPDR-A1"
 149                 .format(self.restconf_baseurl))
 150          data = {"node": [{
 151               "node-id": "XPDR-A1",
 152               "netconf-node-topology:username": "admin",
 153               "netconf-node-topology:password": "admin",
 154               "netconf-node-topology:host": "127.0.0.1",
 155               "netconf-node-topology:port": "17840",
 156               "netconf-node-topology:tcp-only": "false",
 157               "netconf-node-topology:pass-through": {}}]}
 158          headers = {'content-type': 'application/json'}
 159          response = requests.request(
 160               "PUT", url, data=json.dumps(data), headers=headers,
 161               auth=('admin', 'admin'))
 162          self.assertEqual(response.status_code, requests.codes.created)
 163          time.sleep(20)
 164
 165 #     #Verify the termination points related to XPDR
 166     def test_05_compareOpenroadmTopologyPortMapping_xpdr(self):
 167         self.test_02_compareOpenroadmTopologyPortMapping_rdm()
 168
 169     #Disconnect the XPDRA
 170     def test_06_disconnect_device(self):
 171         url = ("{}/config/network-topology:"
 172                "network-topology/topology/topology-netconf/node/XPDR-A1"
 173               .format(self.restconf_baseurl))
 174         data = {}
 175         headers = {'content-type': 'application/json'}
 176         response = requests.request(
 177             "DELETE", url, data=json.dumps(data), headers=headers,
 178             auth=('admin', 'admin'))
 179         self.assertEqual(response.status_code, requests.codes.ok)
 180
 181 if __name__ == "__main__":
 182     #logging.basicConfig(filename='./transportpce_tests/log/response.log',filemode='w',level=logging.DEBUG)
 183     #logging.debug('I am there')
 184     unittest.main(verbosity=2)