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