diff options
Diffstat (limited to 'functest/opnfv_tests/vnf/router/utilvnf.py')
| -rw-r--r-- | functest/opnfv_tests/vnf/router/utilvnf.py | 345 |
1 files changed, 345 insertions, 0 deletions
diff --git a/functest/opnfv_tests/vnf/router/utilvnf.py b/functest/opnfv_tests/vnf/router/utilvnf.py new file mode 100644 index 00000000..084af331 --- /dev/null +++ b/functest/opnfv_tests/vnf/router/utilvnf.py @@ -0,0 +1,345 @@ +#!/usr/bin/env python + +# Copyright (c) 2017 Okinawa Open Laboratory and others. +# +# All rights reserved. This program and the accompanying materials +# are made available under the terms of the Apache License, Version 2.0 +# which accompanies this distribution, and is available at +# http://www.apache.org/licenses/LICENSE-2.0 + +""" Utility module of vrouter testcase """ + +import json +import logging +import os +import pkg_resources +import requests +import yaml + +from functest.utils.constants import CONST +from git import Repo +from novaclient import client as novaclient +from keystoneauth1 import session +from keystoneauth1 import loading +from requests.auth import HTTPBasicAuth + +RESULT_SPRIT_INDEX = { + "transfer": 8, + "bandwidth": 6, + "jitter": 4, + "los_total": 2, + "pkt_loss": 1 +} + +BIT_PER_BYTE = 8 + +NOVA_CLIENT_API_VERSION = '2' +NOVA_CILENT_NETWORK_INFO_INDEX = 0 +CFY_INFO_OUTPUT_FILE = "output.txt" + +CIDR_NETWORK_SEGMENT_INFO_INDEX = 0 +PACKET_LOST_INFO_INDEX = 0 +PACKET_TOTAL_INFO_INDEX = 1 + +NUMBER_OF_DIGITS_FOR_AVG_TRANSFER = 0 +NUMBER_OF_DIGITS_FOR_AVG_BANDWIDTH = 0 +NUMBER_OF_DIGITS_FOR_AVG_JITTER = 3 +NUMBER_OF_DIGITS_FOR_AVG_PKT_LOSS = 1 + + +class Utilvnf(object): + """ Utility class of vrouter testcase """ + + logger = logging.getLogger(__name__) + + def __init__(self): + self.username = "" + self.password = "" + self.auth_url = "" + self.tenant_name = "" + self.region_name = "" + + data_dir = data_dir = CONST.__getattribute__('dir_router_data') + + self.vnf_data_dir = data_dir + self.opnfv_vnf_data_dir = "opnfv-vnf-data/" + self.command_template_dir = "command_template/" + self.test_scenario_yaml = "test_scenario.yaml" + test_env_config_yaml_file = "test_env_config.yaml" + self.test_cmd_map_yaml_file = "test_cmd_map.yaml" + self.test_env_config_yaml = os.path.join( + self.vnf_data_dir, + self.opnfv_vnf_data_dir, + test_env_config_yaml_file) + + self.blueprint_dir = "opnfv-vnf-vyos-blueprint/" + self.blueprint_file_name = "function-test-openstack-blueprint.yaml" + + if not os.path.exists(self.vnf_data_dir): + os.makedirs(self.vnf_data_dir) + + case_dir = pkg_resources.resource_filename( + 'functest', 'opnfv_tests/vnf/router') + + config_file_name = CONST.__getattribute__( + 'vnf_{}_config'.format("vyos_vrouter")) + + config_file = os.path.join(case_dir, config_file_name) + + with open(config_file) as file_fd: + vrouter_config_yaml = yaml.safe_load(file_fd) + file_fd.close() + + test_data = vrouter_config_yaml.get("test_data") + + self.logger.debug("Downloading the test data.") + vrouter_data_path = self.vnf_data_dir + self.opnfv_vnf_data_dir + + if not os.path.exists(vrouter_data_path): + Repo.clone_from(test_data['url'], + vrouter_data_path, + branch=test_data['branch']) + + with open(self.test_env_config_yaml) as file_fd: + test_env_config_yaml = yaml.safe_load(file_fd) + file_fd.close() + + self.image = test_env_config_yaml.get( + "general").get("images").get("vyos") + self.tester_image = test_env_config_yaml.get( + "general").get("images").get("tester_vm_os") + + self.test_result_json_file = "test_result.json" + if os.path.isfile(self.test_result_json_file): + os.remove(self.test_result_json_file) + self.logger.debug("removed %s" % self.test_result_json_file) + + def get_nova_client(self): + creds = self.get_nova_credentials() + loader = loading.get_plugin_loader('password') + auth = loader.load_from_options(**creds) + sess = session.Session(auth=auth) + nova_client = novaclient.Client(NOVA_CLIENT_API_VERSION, session=sess) + + return nova_client + + def set_credentials(self, username, password, auth_url, + tenant_name, region_name="RegionOne"): + self.username = username + self.password = password + self.auth_url = auth_url + self.tenant_name = tenant_name + self.region_name = region_name + + def get_nova_credentials(self): + creds = {} + creds['username'] = self.username + creds['password'] = self.password + creds['auth_url'] = self.auth_url + creds['tenant_name'] = self.tenant_name + return creds + + def get_address(self, server_name, network_name): + nova_client = self.get_nova_client() + servers_list = nova_client.servers.list() + server = None + + for server in servers_list: + if server.name == server_name: + break + + address = server.addresses[ + network_name][NOVA_CILENT_NETWORK_INFO_INDEX]["addr"] + + return address + + def get_mac_address(self, server_name, network_name): + nova_client = self.get_nova_client() + servers_list = nova_client.servers.list() + server = None + + for server in servers_list: + if server.name == server_name: + break + + mac_address = server.addresses[network_name][ + NOVA_CILENT_NETWORK_INFO_INDEX][ + "OS-EXT-IPS-MAC:mac_addr"] + + return mac_address + + def reboot_vm(self, server_name): + nova_client = self.get_nova_client() + servers_list = nova_client.servers.list() + server = None + + for server in servers_list: + if server.name == server_name: + break + + server.reboot() + + return + + def delete_vm(self, server_name): + nova_client = self.get_nova_client() + servers_list = nova_client.servers.list() + server = None + + for server in servers_list: + if server.name == server_name: + nova_client.servers.delete(server) + break + + return + + def get_blueprint_outputs(self, cfy_manager_ip, deployment_name): + url = "http://%s/deployments/%s/outputs" % ( + cfy_manager_ip, deployment_name) + + response = requests.get( + url, + auth=HTTPBasicAuth('admin', 'admin'), + headers={'Tenant': 'default_tenant'}) + + resp_data = response.json() + self.logger.debug(resp_data) + data = resp_data["outputs"] + return data + + def get_blueprint_outputs_vnfs(self, cfy_manager_ip, deployment_name): + outputs = self.get_blueprint_outputs(cfy_manager_ip, + deployment_name) + vnfs = outputs["vnfs"] + vnf_list = [] + for vnf_name in vnfs: + vnf_list.append(vnfs[vnf_name]) + return vnf_list + + def get_blueprint_outputs_networks(self, cfy_manager_ip, deployment_name): + outputs = self.get_blueprint_outputs(cfy_manager_ip, + deployment_name) + networks = outputs["networks"] + network_list = [] + for network_name in networks: + network_list.append(networks[network_name]) + return network_list + + def request_vnf_reboot(self, vnf_info_list): + for vnf in vnf_info_list: + self.logger.debug("reboot the " + vnf["vnf_name"]) + self.reboot_vm(vnf["vnf_name"]) + + def request_vm_delete(self, vnf_info_list): + for vnf in vnf_info_list: + self.logger.debug("delete the " + vnf["vnf_name"]) + self.delete_vm(vnf["vnf_name"]) + + def get_vnf_info_list(self, cfy_manager_ip, topology_deploy_name, + target_vnf_name): + network_list = self.get_blueprint_outputs_networks( + cfy_manager_ip, + topology_deploy_name) + vnf_info_list = self.get_blueprint_outputs_vnfs(cfy_manager_ip, + topology_deploy_name) + for vnf in vnf_info_list: + vnf_name = vnf["vnf_name"] + vnf["os_type"] = self.image["os_type"] + vnf["user"] = self.image["user"] + vnf["pass"] = self.image["pass"] + + if vnf_name == target_vnf_name: + vnf["target_vnf_flag"] = True + else: + vnf["target_vnf_flag"] = False + + self.logger.debug("vnf name : " + vnf_name) + self.logger.debug(vnf_name + " floating ip address : " + + vnf["floating_ip"]) + + for network in network_list: + network_name = network["network_name"] + ip_address = self.get_address(vnf["vnf_name"], + network["network_name"]) + vnf[network_name + "_ip"] = ip_address + mac = self.get_mac_address(vnf["vnf_name"], + network["network_name"]) + vnf[network_name + "_mac"] = mac + + self.logger.debug(network_name + "_ip of " + vnf["vnf_name"] + + " : " + vnf[network_name + "_ip"]) + self.logger.debug(network_name + "_mac of " + vnf["vnf_name"] + + " : " + vnf[network_name + "_mac"]) + + return vnf_info_list + + def get_target_vnf(self, vnf_info_list): + for vnf in vnf_info_list: + if vnf["target_vnf_flag"]: + return vnf + + return None + + def get_reference_vnf_list(self, vnf_info_list): + reference_vnf_list = [] + for vnf in vnf_info_list: + if not vnf["target_vnf_flag"]: + reference_vnf_list.append(vnf) + + return reference_vnf_list + + def get_vnf_info(self, vnf_info_list, vnf_name): + for vnf in vnf_info_list: + if vnf["vnf_name"] == vnf_name: + return vnf + + return None + + def convert_functional_test_result(self, result_data_list): + result = {} + for result_data in result_data_list: + test_kind = result_data["test_kind"] + protocol = result_data["protocol"] + test_result_data = result_data["result"] + + if test_kind not in result: + result[test_kind] = [] + + result[test_kind].append({protocol: test_result_data}) + + return {"Functional_test": result} + + def write_result_data(self, result_data): + test_result = [] + if not os.path.isfile(self.test_result_json_file): + file_fd = open(self.test_result_json_file, "w") + file_fd.close() + else: + file_fd = open(self.test_result_json_file, "r") + test_result = json.load(file_fd) + file_fd.close() + + test_result.append(result_data) + + file_fd = open(self.test_result_json_file, "w") + json.dump(test_result, file_fd) + file_fd.close() + + def output_test_result_json(self): + if os.path.isfile(self.test_result_json_file): + file_fd = open(self.test_result_json_file, "r") + test_result = json.load(file_fd) + file_fd.close() + output_json_data = json.dumps(test_result, + sort_keys=True, + indent=4) + self.logger.debug("test_result %s" % output_json_data) + else: + self.logger.debug("Not found %s" % self.test_result_json_file) + + def get_test_scenario(self, file_path): + test_scenario_file = open(file_path, + 'r') + test_scenario_yaml = yaml.safe_load(test_scenario_file) + test_scenario_file.close() + return test_scenario_yaml["test_scenario_list"] |