diff options
Diffstat (limited to 'functest/opnfv_tests/vnf/router/utilvnf.py')
| -rw-r--r-- | functest/opnfv_tests/vnf/router/utilvnf.py | 176 |
1 files changed, 50 insertions, 126 deletions
diff --git a/functest/opnfv_tests/vnf/router/utilvnf.py b/functest/opnfv_tests/vnf/router/utilvnf.py index 9d196836e..111f20c1a 100644 --- a/functest/opnfv_tests/vnf/router/utilvnf.py +++ b/functest/opnfv_tests/vnf/router/utilvnf.py @@ -7,20 +7,17 @@ # which accompanies this distribution, and is available at # http://www.apache.org/licenses/LICENSE-2.0 +# pylint: disable=missing-docstring + """ 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 requests.auth import HTTPBasicAuth -from snaps.openstack.utils import nova_utils - +from functest.utils import config RESULT_SPRIT_INDEX = { "transfer": 8, @@ -46,25 +43,19 @@ NUMBER_OF_DIGITS_FOR_AVG_JITTER = 3 NUMBER_OF_DIGITS_FOR_AVG_PKT_LOSS = 1 -class Utilvnf(object): +class Utilvnf(): # pylint: disable=too-many-instance-attributes """ Utility class of vrouter testcase """ logger = logging.getLogger(__name__) def __init__(self): - self.snaps_creds = "" - - data_dir = data_dir = CONST.__getattribute__('dir_router_data') - - self.vnf_data_dir = data_dir - self.opnfv_vnf_data_dir = "opnfv-vnf-data/" + self.vnf_data_dir = getattr(config.CONF, 'dir_router_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/" @@ -73,29 +64,7 @@ class Utilvnf(object): 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: + with open(self.test_env_config_yaml, encoding='utf-8') as file_fd: test_env_config_yaml = yaml.safe_load(file_fd) file_fd.close() @@ -107,77 +76,32 @@ class Utilvnf(object): 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): - nova_client = nova_utils.nova_client(self.snaps_creds) + self.logger.debug("removed %s", self.test_result_json_file) - return nova_client + self.cloud = None - def set_credentials(self, snaps_creds): - self.snaps_creds = snaps_creds + def set_credentials(self, cloud): + self.cloud = cloud 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 - + server = self.cloud.get_server(server_name) address = server.addresses[ - network_name][NOVA_CILENT_NETWORK_INFO_INDEX]["addr"] + 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 - + server = self.cloud.get_server(server_name) mac_address = server.addresses[network_name][ - NOVA_CILENT_NETWORK_INFO_INDEX][ - "OS-EXT-IPS-MAC:mac_addr"] + 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) - + url = f"http://{cfy_manager_ip}/deployments/{deployment_name}/outputs" response = requests.get( url, - auth=HTTPBasicAuth('admin', 'admin'), + auth=requests.auth.HTTPBasicAuth('admin', 'admin'), headers={'Tenant': 'default_tenant'}) resp_data = response.json() @@ -203,15 +127,10 @@ class Utilvnf(object): 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"]) + self.logger.debug("delete the %s", vnf["vnf_name"]) + self.cloud.delete_server(vnf["vnf_name"]) def get_vnf_info_list(self, cfy_manager_ip, topology_deploy_name, target_vnf_name): @@ -226,12 +145,9 @@ class Utilvnf(object): 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 + vnf["target_vnf_flag"] = bool(vnf_name == target_vnf_name) - self.logger.debug("vnf name : " + vnf_name) + self.logger.debug("vnf name : %s", vnf_name) self.logger.debug(vnf_name + " floating ip address : " + vnf["floating_ip"]) @@ -251,14 +167,16 @@ class Utilvnf(object): return vnf_info_list - def get_target_vnf(self, vnf_info_list): + @staticmethod + def get_target_vnf(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): + @staticmethod + def get_reference_vnf_list(vnf_info_list): reference_vnf_list = [] for vnf in vnf_info_list: if not vnf["target_vnf_flag"]: @@ -266,14 +184,16 @@ class Utilvnf(object): return reference_vnf_list - def get_vnf_info(self, vnf_info_list, vnf_name): + @staticmethod + def get_vnf_info(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): + @staticmethod + def convert_functional_test_result(result_data_list): result = {} for result_data in result_data_list: test_kind = result_data["test_kind"] @@ -290,34 +210,38 @@ class Utilvnf(object): 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() + with open( + self.test_result_json_file, "w", + encoding="utf-8") as file_fd: + pass else: - file_fd = open(self.test_result_json_file, "r") - test_result = json.load(file_fd) - file_fd.close() + with open( + self.test_result_json_file, "r", + encoding="utf-8") as file_fd: + test_result = json.load(file_fd) test_result.append(result_data) - file_fd = open(self.test_result_json_file, "w") - json.dump(test_result, file_fd) - file_fd.close() + with open( + self.test_result_json_file, "w", + encoding="utf-8") as file_fd: + json.dump(test_result, file_fd) 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() + with open( + self.test_result_json_file, "r", + encoding="utf-8") as file_fd: + test_result = json.load(file_fd) output_json_data = json.dumps(test_result, sort_keys=True, indent=4) - self.logger.debug("test_result %s" % output_json_data) + self.logger.debug("test_result %s", output_json_data) else: - self.logger.debug("Not found %s" % self.test_result_json_file) + 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() + @staticmethod + def get_test_scenario(file_path): + with open(file_path, "r", encoding="utf-8") as test_scenario_file: + test_scenario_yaml = yaml.safe_load(test_scenario_file) return test_scenario_yaml["test_scenario_list"] |