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