functest/opnfv_tests/vnf/router/utilvnf.py

   1 #!/usr/bin/env python
   2
   3 # Copyright (c) 2017 Okinawa Open Laboratory and others.
   4 #
   5 # All rights reserved. This program and the accompanying materials
   6 # are made available under the terms of the Apache License, Version 2.0
   7 # which accompanies this distribution, and is available at
   8 # http://www.apache.org/licenses/LICENSE-2.0
   9
  10 """ Utility module of vrouter testcase """
  11
  12 import json
  13 import logging
  14 import os
  15 import pkg_resources
  16 import requests
  17 import yaml
  18
  19 from functest.utils.constants import CONST
  20 from git import Repo
  21 from novaclient import client as novaclient
  22 from keystoneauth1 import session
  23 from keystoneauth1 import loading
  24 from requests.auth import HTTPBasicAuth
  25
  26 RESULT_SPRIT_INDEX = {
  27     "transfer": 8,
  28     "bandwidth": 6,
  29     "jitter": 4,
  30     "los_total": 2,
  31     "pkt_loss": 1
  32 }
  33
  34 BIT_PER_BYTE = 8
  35
  36 NOVA_CLIENT_API_VERSION = '2'
  37 NOVA_CILENT_NETWORK_INFO_INDEX = 0
  38 CFY_INFO_OUTPUT_FILE = "output.txt"
  39
  40 CIDR_NETWORK_SEGMENT_INFO_INDEX = 0
  41 PACKET_LOST_INFO_INDEX = 0
  42 PACKET_TOTAL_INFO_INDEX = 1
  43
  44 NUMBER_OF_DIGITS_FOR_AVG_TRANSFER = 0
  45 NUMBER_OF_DIGITS_FOR_AVG_BANDWIDTH = 0
  46 NUMBER_OF_DIGITS_FOR_AVG_JITTER = 3
  47 NUMBER_OF_DIGITS_FOR_AVG_PKT_LOSS = 1
  48
  49
  50 class Utilvnf(object):
  51     """ Utility class of vrouter testcase """
  52
  53     logger = logging.getLogger(__name__)
  54
  55     def __init__(self):
  56         self.username = ""
  57         self.password = ""
  58         self.auth_url = ""
  59         self.tenant_name = ""
  60         self.region_name = ""
  61
  62         data_dir = data_dir = CONST.__getattribute__('dir_router_data')
  63
  64         self.vnf_data_dir = data_dir
  65         self.opnfv_vnf_data_dir = "opnfv-vnf-data/"
  66         self.command_template_dir = "command_template/"
  67         self.test_scenario_yaml = "test_scenario.yaml"
  68         test_env_config_yaml_file = "test_env_config.yaml"
  69         self.test_cmd_map_yaml_file = "test_cmd_map.yaml"
  70         self.test_env_config_yaml = os.path.join(
  71             self.vnf_data_dir,
  72             self.opnfv_vnf_data_dir,
  73             test_env_config_yaml_file)
  74
  75         self.blueprint_dir = "opnfv-vnf-vyos-blueprint/"
  76         self.blueprint_file_name = "function-test-openstack-blueprint.yaml"
  77
  78         if not os.path.exists(self.vnf_data_dir):
  79             os.makedirs(self.vnf_data_dir)
  80
  81         case_dir = pkg_resources.resource_filename(
  82             'functest', 'opnfv_tests/vnf/router')
  83
  84         config_file_name = CONST.__getattribute__(
  85             'vnf_{}_config'.format("vyos_vrouter"))
  86
  87         config_file = os.path.join(case_dir, config_file_name)
  88
  89         with open(config_file) as file_fd:
  90             vrouter_config_yaml = yaml.safe_load(file_fd)
  91         file_fd.close()
  92
  93         test_data = vrouter_config_yaml.get("test_data")
  94
  95         self.logger.debug("Downloading the test data.")
  96         vrouter_data_path = self.vnf_data_dir + self.opnfv_vnf_data_dir
  97
  98         if not os.path.exists(vrouter_data_path):
  99             Repo.clone_from(test_data['url'],
 100                             vrouter_data_path,
 101                             branch=test_data['branch'])
 102
 103         with open(self.test_env_config_yaml) as file_fd:
 104             test_env_config_yaml = yaml.safe_load(file_fd)
 105         file_fd.close()
 106
 107         self.image = test_env_config_yaml.get(
 108             "general").get("images").get("vyos")
 109         self.tester_image = test_env_config_yaml.get(
 110             "general").get("images").get("tester_vm_os")
 111
 112         self.test_result_json_file = "test_result.json"
 113         if os.path.isfile(self.test_result_json_file):
 114             os.remove(self.test_result_json_file)
 115             self.logger.debug("removed %s" % self.test_result_json_file)
 116
 117     def get_nova_client(self):
 118         creds = self.get_nova_credentials()
 119         loader = loading.get_plugin_loader('password')
 120         auth = loader.load_from_options(**creds)
 121         sess = session.Session(auth=auth)
 122         nova_client = novaclient.Client(NOVA_CLIENT_API_VERSION, session=sess)
 123
 124         return nova_client
 125
 126     def set_credentials(self, username, password, auth_url,
 127                         tenant_name, region_name="RegionOne"):
 128         self.username = username
 129         self.password = password
 130         self.auth_url = auth_url
 131         self.tenant_name = tenant_name
 132         self.region_name = region_name
 133
 134     def get_nova_credentials(self):
 135         creds = {}
 136         creds['username'] = self.username
 137         creds['password'] = self.password
 138         creds['auth_url'] = self.auth_url
 139         creds['tenant_name'] = self.tenant_name
 140         return creds
 141
 142     def get_address(self, server_name, network_name):
 143         nova_client = self.get_nova_client()
 144         servers_list = nova_client.servers.list()
 145         server = None
 146
 147         for server in servers_list:
 148             if server.name == server_name:
 149                 break
 150
 151         address = server.addresses[
 152                       network_name][NOVA_CILENT_NETWORK_INFO_INDEX]["addr"]
 153
 154         return address
 155
 156     def get_mac_address(self, server_name, network_name):
 157         nova_client = self.get_nova_client()
 158         servers_list = nova_client.servers.list()
 159         server = None
 160
 161         for server in servers_list:
 162             if server.name == server_name:
 163                 break
 164
 165         mac_address = server.addresses[network_name][
 166                           NOVA_CILENT_NETWORK_INFO_INDEX][
 167                           "OS-EXT-IPS-MAC:mac_addr"]
 168
 169         return mac_address
 170
 171     def reboot_vm(self, server_name):
 172         nova_client = self.get_nova_client()
 173         servers_list = nova_client.servers.list()
 174         server = None
 175
 176         for server in servers_list:
 177             if server.name == server_name:
 178                 break
 179
 180         server.reboot()
 181
 182         return
 183
 184     def delete_vm(self, server_name):
 185         nova_client = self.get_nova_client()
 186         servers_list = nova_client.servers.list()
 187         server = None
 188
 189         for server in servers_list:
 190             if server.name == server_name:
 191                 nova_client.servers.delete(server)
 192                 break
 193
 194         return
 195
 196     def get_blueprint_outputs(self, cfy_manager_ip, deployment_name):
 197         url = "http://%s/deployments/%s/outputs" % (
 198             cfy_manager_ip, deployment_name)
 199
 200         response = requests.get(
 201             url,
 202             auth=HTTPBasicAuth('admin', 'admin'),
 203             headers={'Tenant': 'default_tenant'})
 204
 205         resp_data = response.json()
 206         self.logger.debug(resp_data)
 207         data = resp_data["outputs"]
 208         return data
 209
 210     def get_blueprint_outputs_vnfs(self, cfy_manager_ip, deployment_name):
 211         outputs = self.get_blueprint_outputs(cfy_manager_ip,
 212                                              deployment_name)
 213         vnfs = outputs["vnfs"]
 214         vnf_list = []
 215         for vnf_name in vnfs:
 216             vnf_list.append(vnfs[vnf_name])
 217         return vnf_list
 218
 219     def get_blueprint_outputs_networks(self, cfy_manager_ip, deployment_name):
 220         outputs = self.get_blueprint_outputs(cfy_manager_ip,
 221                                              deployment_name)
 222         networks = outputs["networks"]
 223         network_list = []
 224         for network_name in networks:
 225             network_list.append(networks[network_name])
 226         return network_list
 227
 228     def request_vnf_reboot(self, vnf_info_list):
 229         for vnf in vnf_info_list:
 230             self.logger.debug("reboot the " + vnf["vnf_name"])
 231             self.reboot_vm(vnf["vnf_name"])
 232
 233     def request_vm_delete(self, vnf_info_list):
 234         for vnf in vnf_info_list:
 235             self.logger.debug("delete the " + vnf["vnf_name"])
 236             self.delete_vm(vnf["vnf_name"])
 237
 238     def get_vnf_info_list(self, cfy_manager_ip, topology_deploy_name,
 239                           target_vnf_name):
 240         network_list = self.get_blueprint_outputs_networks(
 241             cfy_manager_ip,
 242             topology_deploy_name)
 243         vnf_info_list = self.get_blueprint_outputs_vnfs(cfy_manager_ip,
 244                                                         topology_deploy_name)
 245         for vnf in vnf_info_list:
 246             vnf_name = vnf["vnf_name"]
 247             vnf["os_type"] = self.image["os_type"]
 248             vnf["user"] = self.image["user"]
 249             vnf["pass"] = self.image["pass"]
 250
 251             if vnf_name == target_vnf_name:
 252                 vnf["target_vnf_flag"] = True
 253             else:
 254                 vnf["target_vnf_flag"] = False
 255
 256             self.logger.debug("vnf name : " + vnf_name)
 257             self.logger.debug(vnf_name + " floating ip address : " +
 258                               vnf["floating_ip"])
 259
 260             for network in network_list:
 261                 network_name = network["network_name"]
 262                 ip_address = self.get_address(vnf["vnf_name"],
 263                                               network["network_name"])
 264                 vnf[network_name + "_ip"] = ip_address
 265                 mac = self.get_mac_address(vnf["vnf_name"],
 266                                            network["network_name"])
 267                 vnf[network_name + "_mac"] = mac
 268
 269                 self.logger.debug(network_name + "_ip of " + vnf["vnf_name"] +
 270                                   " : " + vnf[network_name + "_ip"])
 271                 self.logger.debug(network_name + "_mac of " + vnf["vnf_name"] +
 272                                   " : " + vnf[network_name + "_mac"])
 273
 274         return vnf_info_list
 275
 276     def get_target_vnf(self, vnf_info_list):
 277         for vnf in vnf_info_list:
 278             if vnf["target_vnf_flag"]:
 279                 return vnf
 280
 281         return None
 282
 283     def get_reference_vnf_list(self, vnf_info_list):
 284         reference_vnf_list = []
 285         for vnf in vnf_info_list:
 286             if not vnf["target_vnf_flag"]:
 287                 reference_vnf_list.append(vnf)
 288
 289         return reference_vnf_list
 290
 291     def get_vnf_info(self, vnf_info_list, vnf_name):
 292         for vnf in vnf_info_list:
 293             if vnf["vnf_name"] == vnf_name:
 294                 return vnf
 295
 296         return None
 297
 298     def convert_functional_test_result(self, result_data_list):
 299         result = {}
 300         for result_data in result_data_list:
 301             test_kind = result_data["test_kind"]
 302             protocol = result_data["protocol"]
 303             test_result_data = result_data["result"]
 304
 305             if test_kind not in result:
 306                 result[test_kind] = []
 307
 308             result[test_kind].append({protocol: test_result_data})
 309
 310         return {"Functional_test": result}
 311
 312     def write_result_data(self, result_data):
 313         test_result = []
 314         if not os.path.isfile(self.test_result_json_file):
 315             file_fd = open(self.test_result_json_file, "w")
 316             file_fd.close()
 317         else:
 318             file_fd = open(self.test_result_json_file, "r")
 319             test_result = json.load(file_fd)
 320             file_fd.close()
 321
 322         test_result.append(result_data)
 323
 324         file_fd = open(self.test_result_json_file, "w")
 325         json.dump(test_result, file_fd)
 326         file_fd.close()
 327
 328     def output_test_result_json(self):
 329         if os.path.isfile(self.test_result_json_file):
 330             file_fd = open(self.test_result_json_file, "r")
 331             test_result = json.load(file_fd)
 332             file_fd.close()
 333             output_json_data = json.dumps(test_result,
 334                                           sort_keys=True,
 335                                           indent=4)
 336             self.logger.debug("test_result %s" % output_json_data)
 337         else:
 338             self.logger.debug("Not found %s" % self.test_result_json_file)
 339
 340     def get_test_scenario(self, file_path):
 341         test_scenario_file = open(file_path,
 342                                   'r')
 343         test_scenario_yaml = yaml.safe_load(test_scenario_file)
 344         test_scenario_file.close()
 345         return test_scenario_yaml["test_scenario_list"]