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 from common import test_utils
  23
  24
  25 class TransportPCEtesting(unittest.TestCase):
  26
  27     processes = None
  28
  29     @classmethod
  30     def setUpClass(cls):
  31         cls.processes = test_utils.start_tpce()
  32         cls.processes = test_utils.start_sims(['xpdra', 'roadma'])
  33
  34     @classmethod
  35     def tearDownClass(cls):
  36         for process in cls.processes:
  37             test_utils.shutdown_process(process)
  38         print("all processes killed")
  39
  40     def setUp(self):
  41         time.sleep(10)
  42
  43     # Connect the ROADMA
  44     def test_01_connect_rdm(self):
  45         response = test_utils.mount_device("ROADM-A1", 'roadma')
  46         self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
  47
  48     # Verify the termination points of the ROADMA
  49     def test_02_compareOpenroadmTopologyPortMapping_rdm(self):
  50         urlTopo = ("{}/config/ietf-network:networks/network/openroadm-topology"
  51                    .format(test_utils.RESTCONF_BASE_URL))
  52         responseTopo = requests.request(
  53             "GET", urlTopo, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
  54         resTopo = responseTopo.json()
  55         nbNode = len(resTopo['network'][0]['node'])
  56         nbMapCumul = 0
  57         nbMappings = 0
  58         for i in range(0, nbNode):
  59             nodeId = resTopo['network'][0]['node'][i]['node-id']
  60             print("nodeId={}".format(nodeId))
  61             nodeMapId = nodeId.split("-")[0] + "-" + nodeId.split("-")[1]
  62             print("nodeMapId={}".format(nodeMapId))
  63             urlMapList = "{}/config/transportpce-portmapping:network/nodes/" + nodeMapId
  64             urlMapListFull = urlMapList.format(test_utils.RESTCONF_BASE_URL)
  65             responseMapList = requests.request(
  66                 "GET", urlMapListFull, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
  67             resMapList = responseMapList.json()
  68
  69             nbMappings = len(resMapList['nodes'][0]['mapping']) - nbMapCumul
  70             nbTp = len(resTopo['network'][0]['node'][i]['ietf-network-topology:termination-point'])
  71             nbMapCurrent = 0
  72             for j in range(0, nbTp):
  73                 tpId = resTopo['network'][0]['node'][i]['ietf-network-topology:termination-point'][j]['tp-id']
  74                 if((not "CP" in tpId) and (not "CTP" in tpId)):
  75                     urlMap = "{}/config/transportpce-portmapping:network/nodes/" + nodeMapId + "/mapping/" + tpId
  76                     urlMapFull = urlMap.format(test_utils.RESTCONF_BASE_URL)
  77                     responseMap = requests.request(
  78                         "GET", urlMapFull, headers=test_utils.TYPE_APPLICATION_JSON, auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
  79                     self.assertEqual(responseMap.status_code, requests.codes.ok)
  80                     if(responseMap.status_code == requests.codes.ok):
  81                         nbMapCurrent += 1
  82             nbMapCumul += nbMapCurrent
  83         nbMappings -= nbMapCurrent
  84         self.assertEqual(nbMappings, 0)
  85
  86     # Disconnect the ROADMA
  87     def test_03_disconnect_rdm(self):
  88         url = ("{}/config/network-topology:"
  89                "network-topology/topology/topology-netconf/node/ROADM-A1"
  90                .format(test_utils.RESTCONF_BASE_URL))
  91         data = {}
  92         response = requests.request(
  93             "DELETE", url, data=json.dumps(data), headers=test_utils.TYPE_APPLICATION_JSON,
  94             auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
  95         self.assertEqual(response.status_code, requests.codes.ok)
  96
  97 #     #Connect the XPDRA
  98     def test_04_connect_xpdr(self):
  99         response = test_utils.mount_device("XPDR-A1", 'xpdra')
 100         self.assertEqual(response.status_code, requests.codes.created, test_utils.CODE_SHOULD_BE_201)
 101
 102 #     #Verify the termination points related to XPDR
 103     def test_05_compareOpenroadmTopologyPortMapping_xpdr(self):
 104         self.test_02_compareOpenroadmTopologyPortMapping_rdm()
 105
 106     # Disconnect the XPDRA
 107     def test_06_disconnect_device(self):
 108         url = ("{}/config/network-topology:"
 109                "network-topology/topology/topology-netconf/node/XPDR-A1"
 110                .format(test_utils.RESTCONF_BASE_URL))
 111         data = {}
 112         response = requests.request(
 113             "DELETE", url, data=json.dumps(data), headers=test_utils.TYPE_APPLICATION_JSON,
 114             auth=(test_utils.ODL_LOGIN, test_utils.ODL_PWD))
 115         self.assertEqual(response.status_code, requests.codes.ok)
 116
 117
 118 if __name__ == "__main__":
 119     # logging.basicConfig(filename='./transportpce_tests/log/response.log',filemode='w',level=logging.DEBUG)
 120     #logging.debug('I am there')
 121     unittest.main(verbosity=2)