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