tests/transportpce_tests/1.2.1/test_gnpy.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
  23 class TransportGNPYtesting(unittest.TestCase):
  24
  25     gnpy_process = None
  26     odl_process = None
  27     restconf_baseurl = "http://localhost:8181/restconf"
  28
  29     @classmethod
  30     def __init_logfile(cls):
  31         if os.path.isfile("./transportpce_tests/gnpy.log"):
  32             os.remove("transportpce_tests/gnpy.log")
  33
  34     @classmethod
  35     def __start_odl(cls):
  36         executable = "../karaf/target/assembly/bin/karaf"
  37         with open('transportpce_tests/odl.log', 'w') as outfile:
  38             cls.odl_process = subprocess.Popen(
  39                 ["bash", executable, "server"], stdout=outfile,
  40                 stdin=open(os.devnull))
  41
  42     @classmethod
  43     def setUpClass(cls):
  44         cls.__start_odl()
  45         time.sleep(30)
  46
  47     @classmethod
  48     def tearDownClass(cls):
  49         for child in psutil.Process(cls.odl_process.pid).children():
  50             child.send_signal(signal.SIGINT)
  51             child.wait()
  52         cls.odl_process.send_signal(signal.SIGINT)
  53         cls.odl_process.wait()
  54
  55     def setUp(self):
  56         time.sleep(2)
  57
  58     #Mount the different topologies
  59     def test_01_connect_clliNetwork(self):
  60         url = ("{}/config/ietf-network:networks/network/clli-network"
  61                .format(self.restconf_baseurl))
  62         topo_clliNet_file = "sample_configs/gnpy/clliNetwork.json"
  63         if os.path.isfile(topo_clliNet_file):
  64             with open(topo_clliNet_file, 'r') as clli_net:
  65                 body = clli_net.read()
  66         headers = {'content-type': 'application/json'}
  67         response = requests.request(
  68              "PUT", url, data=body, headers=headers,
  69              auth=('admin', 'admin'))
  70         self.assertEqual(response.status_code, requests.codes.ok)
  71         time.sleep(3)
  72
  73     def test_02_connect_openroadmNetwork(self):
  74         url = ("{}/config/ietf-network:networks/network/openroadm-network"
  75                .format(self.restconf_baseurl))
  76         topo_ordNet_file = "sample_configs/gnpy/openroadmNetwork.json"
  77         if os.path.isfile(topo_ordNet_file):
  78             with open(topo_ordNet_file, 'r') as ord_net:
  79                 body = ord_net.read()
  80         headers = {'content-type': 'application/json'}
  81         response = requests.request(
  82              "PUT", url, data=body, headers=headers,
  83              auth=('admin', 'admin'))
  84         self.assertEqual(response.status_code, requests.codes.ok)
  85         time.sleep(3)
  86
  87     def test_03_connect_openroadmTopology(self):
  88         url = ("{}/config/ietf-network:networks/network/openroadm-topology"
  89                .format(self.restconf_baseurl))
  90         topo_ordTopo_file = "sample_configs/gnpy/openroadmTopology.json"
  91         if os.path.isfile(topo_ordTopo_file):
  92             with open(topo_ordTopo_file, 'r') as ord_topo:
  93                 body = ord_topo.read()
  94         headers = {'content-type': 'application/json'}
  95         response = requests.request(
  96              "PUT", url, data=body, headers=headers,
  97              auth=('admin', 'admin'))
  98         self.assertEqual(response.status_code, requests.codes.ok)
  99         time.sleep(3)
 100
 101     #Test the gnpy
 102     def test_04_path_computation_xpdr_bi(self):
 103         url = ("{}/operations/transportpce-pce:path-computation-request"
 104               .format(self.restconf_baseurl))
 105         body = {"input": {
 106                 "service-name": "service-1",
 107                 "resource-reserve": "true",
 108                 "pce-metric": "hop-count",
 109                 "service-handler-header": {
 110                     "request-id": "request-1"
 111                 },
 112                 "service-a-end": {
 113                     "node-id": "XPDRA",
 114                     "service-rate": "100",
 115                     "clli": "nodeA"
 116                 },
 117                 "service-z-end": {
 118                     "node-id": "XPDRB",
 119                     "service-rate": "100",
 120                     "clli": "nodeB"
 121                 }
 122             }
 123         }
 124         headers = {'content-type': 'application/json',
 125         "Accept": "application/json"}
 126         response = requests.request(
 127             "POST", url, data=json.dumps(body), headers=headers,
 128             auth=('admin', 'admin'))
 129         self.assertEqual(response.status_code, requests.codes.ok)
 130         res = response.json()
 131         self.assertEqual(res['output']['gnpy-response'][0]['path-dir'], 'A-to-Z')
 132         self.assertEqual(res['output']['gnpy-response'][0]['feasibility'],True)
 133         self.assertEqual(res['output']['gnpy-response'][1]['path-dir'], 'Z-to-A')
 134         self.assertEqual(res['output']['gnpy-response'][1]['feasibility'],True)
 135         time.sleep(5)
 136
 137     #Disconnect the different topology
 138     def test_05_disconnect_openroadmTopology(self):
 139         url = ("{}/config/ietf-network:networks/network/openroadm-topology"
 140                .format(self.restconf_baseurl))
 141         data = {}
 142         headers = {'content-type': 'application/json'}
 143         response = requests.request(
 144              "DELETE", url, data=json.dumps(data), headers=headers,
 145              auth=('admin', 'admin'))
 146         self.assertEqual(response.status_code, requests.codes.ok)
 147         time.sleep(3)
 148
 149     def test_06_disconnect_openroadmNetwork(self):
 150         #Config ROADMA
 151         url = ("{}/config/ietf-network:networks/network/openroadm-network"
 152                .format(self.restconf_baseurl))
 153         data = {}
 154         headers = {'content-type': 'application/json'}
 155         response = requests.request(
 156              "DELETE", url, data=json.dumps(data), headers=headers,
 157              auth=('admin', 'admin'))
 158         self.assertEqual(response.status_code, requests.codes.ok)
 159         time.sleep(3)
 160
 161     def test_07_disconnect_clliNetwork(self):
 162         url = ("{}/config/ietf-network:networks/network/clli-network"
 163                .format(self.restconf_baseurl))
 164         data = {}
 165         headers = {'content-type': 'application/json'}
 166         response = requests.request(
 167              "DELETE", url, data=json.dumps(data), headers=headers,
 168              auth=('admin', 'admin'))
 169         self.assertEqual(response.status_code, requests.codes.ok)
 170         time.sleep(3)
 171
 172 if __name__ == "__main__":
 173     #logging.basicConfig(filename='./transportpce_tests/log/response.log',filemode='w',level=logging.DEBUG)
 174     unittest.main(verbosity=2)