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 # pylint: disable=missing-docstring
  11
  12 """ Utility module of vrouter testcase """
  13
  14 import json
  15 import logging
  16 import os
  17 import pkg_resources
  18 import requests
  19 import yaml
  20
  21 from git import Repo
  22
  23 from functest.utils import config
  24
  25 RESULT_SPRIT_INDEX = {
  26     "transfer": 8,
  27     "bandwidth": 6,
  28     "jitter": 4,
  29     "los_total": 2,
  30     "pkt_loss": 1
  31 }
  32
  33 BIT_PER_BYTE = 8
  34
  35 NOVA_CLIENT_API_VERSION = '2'
  36 NOVA_CILENT_NETWORK_INFO_INDEX = 0
  37 CFY_INFO_OUTPUT_FILE = "output.txt"
  38
  39 CIDR_NETWORK_SEGMENT_INFO_INDEX = 0
  40 PACKET_LOST_INFO_INDEX = 0
  41 PACKET_TOTAL_INFO_INDEX = 1
  42
  43 NUMBER_OF_DIGITS_FOR_AVG_TRANSFER = 0
  44 NUMBER_OF_DIGITS_FOR_AVG_BANDWIDTH = 0
  45 NUMBER_OF_DIGITS_FOR_AVG_JITTER = 3
  46 NUMBER_OF_DIGITS_FOR_AVG_PKT_LOSS = 1
  47
  48
  49 class Utilvnf(object):  # pylint: disable=too-many-instance-attributes
  50     """ Utility class of vrouter testcase """
  51
  52     logger = logging.getLogger(__name__)
  53
  54     def __init__(self):
  55         self.vnf_data_dir = getattr(config.CONF, 'dir_router_data')
  56         self.opnfv_vnf_data_dir = "opnfv-vnf-data/"
  57         self.command_template_dir = "command_template/"
  58         self.test_scenario_yaml = "test_scenario.yaml"
  59         test_env_config_yaml_file = "test_env_config.yaml"
  60         self.test_cmd_map_yaml_file = "test_cmd_map.yaml"
  61         self.test_env_config_yaml = os.path.join(
  62             self.vnf_data_dir,
  63             self.opnfv_vnf_data_dir,
  64             test_env_config_yaml_file)
  65
  66         self.blueprint_dir = "opnfv-vnf-vyos-blueprint/"
  67         self.blueprint_file_name = "function-test-openstack-blueprint.yaml"
  68
  69         if not os.path.exists(self.vnf_data_dir):
  70             os.makedirs(self.vnf_data_dir)
  71
  72         case_dir = pkg_resources.resource_filename(
  73             'functest', 'opnfv_tests/vnf/router')
  74
  75         config_file_name = getattr(
  76             config.CONF, 'vnf_{}_config'.format("vyos_vrouter"))
  77
  78         config_file = os.path.join(case_dir, config_file_name)
  79
  80         with open(config_file) as file_fd:
  81             vrouter_config_yaml = yaml.safe_load(file_fd)
  82         file_fd.close()
  83
  84         test_data = vrouter_config_yaml.get("test_data")
  85
  86         self.logger.debug("Downloading the test data.")
  87         vrouter_data_path = self.vnf_data_dir + self.opnfv_vnf_data_dir
  88
  89         if not os.path.exists(vrouter_data_path):
  90             Repo.clone_from(test_data['url'],
  91                             vrouter_data_path,
  92                             branch=test_data['branch'])
  93
  94         with open(self.test_env_config_yaml) as file_fd:
  95             test_env_config_yaml = yaml.safe_load(file_fd)
  96         file_fd.close()
  97
  98         self.image = test_env_config_yaml.get(
  99             "general").get("images").get("vyos")
 100         self.tester_image = test_env_config_yaml.get(
 101             "general").get("images").get("tester_vm_os")
 102
 103         self.test_result_json_file = "test_result.json"
 104         if os.path.isfile(self.test_result_json_file):
 105             os.remove(self.test_result_json_file)
 106             self.logger.debug("removed %s", self.test_result_json_file)
 107
 108         self.cloud = None
 109
 110     def set_credentials(self, cloud):
 111         self.cloud = cloud
 112
 113     def get_address(self, server_name, network_name):
 114         server = self.cloud.get_server(server_name)
 115         address = server.addresses[
 116             network_name][NOVA_CILENT_NETWORK_INFO_INDEX]["addr"]
 117
 118         return address
 119
 120     def get_mac_address(self, server_name, network_name):
 121         server = self.cloud.get_server(server_name)
 122         mac_address = server.addresses[network_name][
 123             NOVA_CILENT_NETWORK_INFO_INDEX]["OS-EXT-IPS-MAC:mac_addr"]
 124
 125         return mac_address
 126
 127     def get_blueprint_outputs(self, cfy_manager_ip, deployment_name):
 128         url = "http://%s/deployments/%s/outputs" % (
 129             cfy_manager_ip, deployment_name)
 130
 131         response = requests.get(
 132             url,
 133             auth=requests.auth.HTTPBasicAuth('admin', 'admin'),
 134             headers={'Tenant': 'default_tenant'})
 135
 136         resp_data = response.json()
 137         self.logger.debug(resp_data)
 138         data = resp_data["outputs"]
 139         return data
 140
 141     def get_blueprint_outputs_vnfs(self, cfy_manager_ip, deployment_name):
 142         outputs = self.get_blueprint_outputs(cfy_manager_ip,
 143                                              deployment_name)
 144         vnfs = outputs["vnfs"]
 145         vnf_list = []
 146         for vnf_name in vnfs:
 147             vnf_list.append(vnfs[vnf_name])
 148         return vnf_list
 149
 150     def get_blueprint_outputs_networks(self, cfy_manager_ip, deployment_name):
 151         outputs = self.get_blueprint_outputs(cfy_manager_ip,
 152                                              deployment_name)
 153         networks = outputs["networks"]
 154         network_list = []
 155         for network_name in networks:
 156             network_list.append(networks[network_name])
 157         return network_list
 158
 159     def request_vm_delete(self, vnf_info_list):
 160         for vnf in vnf_info_list:
 161             self.logger.debug("delete the %s", vnf["vnf_name"])
 162             self.cloud.delete_server(vnf["vnf_name"])
 163
 164     def get_vnf_info_list(self, cfy_manager_ip, topology_deploy_name,
 165                           target_vnf_name):
 166         network_list = self.get_blueprint_outputs_networks(
 167             cfy_manager_ip,
 168             topology_deploy_name)
 169         vnf_info_list = self.get_blueprint_outputs_vnfs(cfy_manager_ip,
 170                                                         topology_deploy_name)
 171         for vnf in vnf_info_list:
 172             vnf_name = vnf["vnf_name"]
 173             vnf["os_type"] = self.image["os_type"]
 174             vnf["user"] = self.image["user"]
 175             vnf["pass"] = self.image["pass"]
 176
 177             vnf["target_vnf_flag"] = bool(vnf_name == target_vnf_name)
 178
 179             self.logger.debug("vnf name : %s", vnf_name)
 180             self.logger.debug(vnf_name + " floating ip address : " +
 181                               vnf["floating_ip"])
 182
 183             for network in network_list:
 184                 network_name = network["network_name"]
 185                 ip_address = self.get_address(vnf["vnf_name"],
 186                                               network["network_name"])
 187                 vnf[network_name + "_ip"] = ip_address
 188                 mac = self.get_mac_address(vnf["vnf_name"],
 189                                            network["network_name"])
 190                 vnf[network_name + "_mac"] = mac
 191
 192                 self.logger.debug(network_name + "_ip of " + vnf["vnf_name"] +
 193                                   " : " + vnf[network_name + "_ip"])
 194                 self.logger.debug(network_name + "_mac of " + vnf["vnf_name"] +
 195                                   " : " + vnf[network_name + "_mac"])
 196
 197         return vnf_info_list
 198
 199     @staticmethod
 200     def get_target_vnf(vnf_info_list):
 201         for vnf in vnf_info_list:
 202             if vnf["target_vnf_flag"]:
 203                 return vnf
 204
 205         return None
 206
 207     @staticmethod
 208     def get_reference_vnf_list(vnf_info_list):
 209         reference_vnf_list = []
 210         for vnf in vnf_info_list:
 211             if not vnf["target_vnf_flag"]:
 212                 reference_vnf_list.append(vnf)
 213
 214         return reference_vnf_list
 215
 216     @staticmethod
 217     def get_vnf_info(vnf_info_list, vnf_name):
 218         for vnf in vnf_info_list:
 219             if vnf["vnf_name"] == vnf_name:
 220                 return vnf
 221
 222         return None
 223
 224     @staticmethod
 225     def convert_functional_test_result(result_data_list):
 226         result = {}
 227         for result_data in result_data_list:
 228             test_kind = result_data["test_kind"]
 229             protocol = result_data["protocol"]
 230             test_result_data = result_data["result"]
 231
 232             if test_kind not in result:
 233                 result[test_kind] = []
 234
 235             result[test_kind].append({protocol: test_result_data})
 236
 237         return {"Functional_test": result}
 238
 239     def write_result_data(self, result_data):
 240         test_result = []
 241         if not os.path.isfile(self.test_result_json_file):
 242             file_fd = open(self.test_result_json_file, "w")
 243             file_fd.close()
 244         else:
 245             file_fd = open(self.test_result_json_file, "r")
 246             test_result = json.load(file_fd)
 247             file_fd.close()
 248
 249         test_result.append(result_data)
 250
 251         file_fd = open(self.test_result_json_file, "w")
 252         json.dump(test_result, file_fd)
 253         file_fd.close()
 254
 255     def output_test_result_json(self):
 256         if os.path.isfile(self.test_result_json_file):
 257             file_fd = open(self.test_result_json_file, "r")
 258             test_result = json.load(file_fd)
 259             file_fd.close()
 260             output_json_data = json.dumps(test_result,
 261                                           sort_keys=True,
 262                                           indent=4)
 263             self.logger.debug("test_result %s", output_json_data)
 264         else:
 265             self.logger.debug("Not found %s", self.test_result_json_file)
 266
 267     @staticmethod
 268     def get_test_scenario(file_path):
 269         test_scenario_file = open(file_path,
 270                                   'r')
 271         test_scenario_yaml = yaml.safe_load(test_scenario_file)
 272         test_scenario_file.close()
 273         return test_scenario_yaml["test_scenario_list"]