# 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,
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/"
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.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()
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):
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"])
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"]:
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"]
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)
Code Review / functest.git / blobdiff