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 functest.utils import config
  22 from git import Repo
  23 from snaps.openstack.utils import nova_utils
  24
  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):  # pylint: disable=too-many-instance-attributes
  51     """ Utility class of vrouter testcase """
  52
  53     logger = logging.getLogger(__name__)
  54
  55     def __init__(self):
  56         self.snaps_creds = ""
  57         self.vnf_data_dir = getattr(config.CONF, 'dir_router_data')
  58         self.opnfv_vnf_data_dir = "opnfv-vnf-data/"
  59         self.command_template_dir = "command_template/"
  60         self.test_scenario_yaml = "test_scenario.yaml"
  61         test_env_config_yaml_file = "test_env_config.yaml"
  62         self.test_cmd_map_yaml_file = "test_cmd_map.yaml"
  63         self.test_env_config_yaml = os.path.join(
  64             self.vnf_data_dir,
  65             self.opnfv_vnf_data_dir,
  66             test_env_config_yaml_file)
  67
  68         self.blueprint_dir = "opnfv-vnf-vyos-blueprint/"
  69         self.blueprint_file_name = "function-test-openstack-blueprint.yaml"
  70
  71         if not os.path.exists(self.vnf_data_dir):
  72             os.makedirs(self.vnf_data_dir)
  73
  74         case_dir = pkg_resources.resource_filename(
  75             'functest', 'opnfv_tests/vnf/router')
  76
  77         config_file_name = getattr(
  78             config.CONF, 'vnf_{}_config'.format("vyos_vrouter"))
  79
  80         config_file = os.path.join(case_dir, config_file_name)
  81
  82         with open(config_file) as file_fd:
  83             vrouter_config_yaml = yaml.safe_load(file_fd)
  84         file_fd.close()
  85
  86         test_data = vrouter_config_yaml.get("test_data")
  87
  88         self.logger.debug("Downloading the test data.")
  89         vrouter_data_path = self.vnf_data_dir + self.opnfv_vnf_data_dir
  90
  91         if not os.path.exists(vrouter_data_path):
  92             Repo.clone_from(test_data['url'],
  93                             vrouter_data_path,
  94                             branch=test_data['branch'])
  95
  96         with open(self.test_env_config_yaml) as file_fd:
  97             test_env_config_yaml = yaml.safe_load(file_fd)
  98         file_fd.close()
  99
 100         self.image = test_env_config_yaml.get(
 101             "general").get("images").get("vyos")
 102         self.tester_image = test_env_config_yaml.get(
 103             "general").get("images").get("tester_vm_os")
 104
 105         self.test_result_json_file = "test_result.json"
 106         if os.path.isfile(self.test_result_json_file):
 107             os.remove(self.test_result_json_file)
 108             self.logger.debug("removed %s", self.test_result_json_file)
 109
 110     def get_nova_client(self):
 111         nova_client = nova_utils.nova_client(self.snaps_creds)
 112
 113         return nova_client
 114
 115     def set_credentials(self, snaps_creds):
 116         self.snaps_creds = snaps_creds
 117
 118     def get_address(self, server_name, network_name):
 119         nova_client = self.get_nova_client()
 120         servers_list = nova_client.servers.list()
 121         server = None
 122
 123         for server in servers_list:
 124             if server.name == server_name:
 125                 break
 126
 127         address = server.addresses[
 128             network_name][NOVA_CILENT_NETWORK_INFO_INDEX]["addr"]
 129
 130         return address
 131
 132     def get_mac_address(self, server_name, network_name):
 133         nova_client = self.get_nova_client()
 134         servers_list = nova_client.servers.list()
 135         server = None
 136
 137         for server in servers_list:
 138             if server.name == server_name:
 139                 break
 140
 141         mac_address = server.addresses[network_name][
 142             NOVA_CILENT_NETWORK_INFO_INDEX]["OS-EXT-IPS-MAC:mac_addr"]
 143
 144         return mac_address
 145
 146     def reboot_vm(self, server_name):
 147         nova_client = self.get_nova_client()
 148         servers_list = nova_client.servers.list()
 149         server = None
 150
 151         for server in servers_list:
 152             if server.name == server_name:
 153                 break
 154
 155         server.reboot()
 156
 157         return
 158
 159     def delete_vm(self, server_name):
 160         nova_client = self.get_nova_client()
 161         servers_list = nova_client.servers.list()
 162         server = None
 163
 164         for server in servers_list:
 165             if server.name == server_name:
 166                 nova_client.servers.delete(server)
 167                 break
 168
 169         return
 170
 171     def get_blueprint_outputs(self, cfy_manager_ip, deployment_name):
 172         url = "http://%s/deployments/%s/outputs" % (
 173             cfy_manager_ip, deployment_name)
 174
 175         response = requests.get(
 176             url,
 177             auth=requests.auth.HTTPBasicAuth('admin', 'admin'),
 178             headers={'Tenant': 'default_tenant'})
 179
 180         resp_data = response.json()
 181         self.logger.debug(resp_data)
 182         data = resp_data["outputs"]
 183         return data
 184
 185     def get_blueprint_outputs_vnfs(self, cfy_manager_ip, deployment_name):
 186         outputs = self.get_blueprint_outputs(cfy_manager_ip,
 187                                              deployment_name)
 188         vnfs = outputs["vnfs"]
 189         vnf_list = []
 190         for vnf_name in vnfs:
 191             vnf_list.append(vnfs[vnf_name])
 192         return vnf_list
 193
 194     def get_blueprint_outputs_networks(self, cfy_manager_ip, deployment_name):
 195         outputs = self.get_blueprint_outputs(cfy_manager_ip,
 196                                              deployment_name)
 197         networks = outputs["networks"]
 198         network_list = []
 199         for network_name in networks:
 200             network_list.append(networks[network_name])
 201         return network_list
 202
 203     def request_vnf_reboot(self, vnf_info_list):
 204         for vnf in vnf_info_list:
 205             self.logger.debug("reboot the " + vnf["vnf_name"])
 206             self.reboot_vm(vnf["vnf_name"])
 207
 208     def request_vm_delete(self, vnf_info_list):
 209         for vnf in vnf_info_list:
 210             self.logger.debug("delete the " + vnf["vnf_name"])
 211             self.delete_vm(vnf["vnf_name"])
 212
 213     def get_vnf_info_list(self, cfy_manager_ip, topology_deploy_name,
 214                           target_vnf_name):
 215         network_list = self.get_blueprint_outputs_networks(
 216             cfy_manager_ip,
 217             topology_deploy_name)
 218         vnf_info_list = self.get_blueprint_outputs_vnfs(cfy_manager_ip,
 219                                                         topology_deploy_name)
 220         for vnf in vnf_info_list:
 221             vnf_name = vnf["vnf_name"]
 222             vnf["os_type"] = self.image["os_type"]
 223             vnf["user"] = self.image["user"]
 224             vnf["pass"] = self.image["pass"]
 225
 226             vnf["target_vnf_flag"] = bool(vnf_name == target_vnf_name)
 227
 228             self.logger.debug("vnf name : " + vnf_name)
 229             self.logger.debug(vnf_name + " floating ip address : " +
 230                               vnf["floating_ip"])
 231
 232             for network in network_list:
 233                 network_name = network["network_name"]
 234                 ip_address = self.get_address(vnf["vnf_name"],
 235                                               network["network_name"])
 236                 vnf[network_name + "_ip"] = ip_address
 237                 mac = self.get_mac_address(vnf["vnf_name"],
 238                                            network["network_name"])
 239                 vnf[network_name + "_mac"] = mac
 240
 241                 self.logger.debug(network_name + "_ip of " + vnf["vnf_name"] +
 242                                   " : " + vnf[network_name + "_ip"])
 243                 self.logger.debug(network_name + "_mac of " + vnf["vnf_name"] +
 244                                   " : " + vnf[network_name + "_mac"])
 245
 246         return vnf_info_list
 247
 248     @staticmethod
 249     def get_target_vnf(vnf_info_list):
 250         for vnf in vnf_info_list:
 251             if vnf["target_vnf_flag"]:
 252                 return vnf
 253
 254         return None
 255
 256     @staticmethod
 257     def get_reference_vnf_list(vnf_info_list):
 258         reference_vnf_list = []
 259         for vnf in vnf_info_list:
 260             if not vnf["target_vnf_flag"]:
 261                 reference_vnf_list.append(vnf)
 262
 263         return reference_vnf_list
 264
 265     @staticmethod
 266     def get_vnf_info(vnf_info_list, vnf_name):
 267         for vnf in vnf_info_list:
 268             if vnf["vnf_name"] == vnf_name:
 269                 return vnf
 270
 271         return None
 272
 273     @staticmethod
 274     def convert_functional_test_result(result_data_list):
 275         result = {}
 276         for result_data in result_data_list:
 277             test_kind = result_data["test_kind"]
 278             protocol = result_data["protocol"]
 279             test_result_data = result_data["result"]
 280
 281             if test_kind not in result:
 282                 result[test_kind] = []
 283
 284             result[test_kind].append({protocol: test_result_data})
 285
 286         return {"Functional_test": result}
 287
 288     def write_result_data(self, result_data):
 289         test_result = []
 290         if not os.path.isfile(self.test_result_json_file):
 291             file_fd = open(self.test_result_json_file, "w")
 292             file_fd.close()
 293         else:
 294             file_fd = open(self.test_result_json_file, "r")
 295             test_result = json.load(file_fd)
 296             file_fd.close()
 297
 298         test_result.append(result_data)
 299
 300         file_fd = open(self.test_result_json_file, "w")
 301         json.dump(test_result, file_fd)
 302         file_fd.close()
 303
 304     def output_test_result_json(self):
 305         if os.path.isfile(self.test_result_json_file):
 306             file_fd = open(self.test_result_json_file, "r")
 307             test_result = json.load(file_fd)
 308             file_fd.close()
 309             output_json_data = json.dumps(test_result,
 310                                           sort_keys=True,
 311                                           indent=4)
 312             self.logger.debug("test_result %s", output_json_data)
 313         else:
 314             self.logger.debug("Not found %s", self.test_result_json_file)
 315
 316     @staticmethod
 317     def get_test_scenario(file_path):
 318         test_scenario_file = open(file_path,
 319                                   'r')
 320         test_scenario_yaml = yaml.safe_load(test_scenario_file)
 321         test_scenario_file.close()
 322         return test_scenario_yaml["test_scenario_list"]