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