X-Git-Url: https://gerrit.opnfv.org/gerrit/gitweb?a=blobdiff_plain;f=functest%2Fopnfv_tests%2Fvnf%2Frouter%2Futilvnf.py;h=a54f6cb0b012faf06ad432d7eb42b07d31ee7ea5;hb=2e7f66734394dc5e749ef9ac95e46a8ded9fb2c9;hp=084af3312be9cae6b184dbed3362ea8b9ae4b761;hpb=63cd8a0bdb80c5d2c885e5205612dcd4bb956981;p=functest.git diff --git a/functest/opnfv_tests/vnf/router/utilvnf.py b/functest/opnfv_tests/vnf/router/utilvnf.py index 084af3312..a54f6cb0b 100644 --- a/functest/opnfv_tests/vnf/router/utilvnf.py +++ b/functest/opnfv_tests/vnf/router/utilvnf.py @@ -7,21 +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 novaclient import client as novaclient -from keystoneauth1 import session -from keystoneauth1 import loading -from requests.auth import HTTPBasicAuth +from functest.utils import config RESULT_SPRIT_INDEX = { "transfer": 8, @@ -47,29 +43,19 @@ NUMBER_OF_DIGITS_FOR_AVG_JITTER = 3 NUMBER_OF_DIGITS_FOR_AVG_PKT_LOSS = 1 -class Utilvnf(object): +class Utilvnf(object): # pylint: disable=too-many-instance-attributes """ 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.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/" @@ -78,28 +64,6 @@ 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: test_env_config_yaml = yaml.safe_load(file_fd) file_fd.close() @@ -112,94 +76,34 @@ 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): - 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 + self.logger.debug("removed %s", self.test_result_json_file) - def get_address(self, server_name, network_name): - nova_client = self.get_nova_client() - servers_list = nova_client.servers.list() - server = None + self.cloud = None - for server in servers_list: - if server.name == server_name: - break + def set_credentials(self, cloud): + self.cloud = cloud + def get_address(self, server_name, network_name): + 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) response = requests.get( url, - auth=HTTPBasicAuth('admin', 'admin'), + auth=requests.auth.HTTPBasicAuth('admin', 'admin'), headers={'Tenant': 'default_tenant'}) resp_data = response.json() @@ -225,15 +129,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): @@ -248,12 +147,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"]) @@ -273,14 +169,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"]: @@ -288,14 +186,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"] @@ -333,11 +233,12 @@ class Utilvnf(object): 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): + @staticmethod + def get_test_scenario(file_path): test_scenario_file = open(file_path, 'r') test_scenario_yaml = yaml.safe_load(test_scenario_file)