X-Git-Url: https://gerrit.opnfv.org/gerrit/gitweb?a=blobdiff_plain;f=functest%2Futils%2Ffunctest_utils.py;h=b49337de2a88d850410060974b8bd3b2e1350128;hb=9e4e6f049fa7d5cce8db61dfa2ebebc55f91c1d8;hp=c1491855471d39347975568984aeeca74389032f;hpb=8e416aee220bf6959e2e21f18a5ebfec1421858b;p=functest.git diff --git a/functest/utils/functest_utils.py b/functest/utils/functest_utils.py index c14918554..b49337de2 100644 --- a/functest/utils/functest_utils.py +++ b/functest/utils/functest_utils.py @@ -11,91 +11,15 @@ from __future__ import print_function import logging -import os -import re -import shutil import subprocess import sys - -import pkg_resources -import dns.resolver -from six.moves import urllib import yaml -from functest.utils import constants +import six LOGGER = logging.getLogger(__name__) -# ---------------------------------------------------------- -# -# INTERNET UTILS -# -# ----------------------------------------------------------- -def check_internet_connectivity(url='http://www.opnfv.org/'): - """ - Check if there is access to the internet - """ - try: - urllib.request.urlopen(url, timeout=5) - return True - except urllib.error.URLError: - return False - - -def download_url(url, dest_path): - """ - Download a file to a destination path given a URL - """ - name = url.rsplit('/')[-1] - dest = dest_path + "/" + name - try: - response = urllib.request.urlopen(url) - except (urllib.error.HTTPError, urllib.error.URLError): - return False - - with open(dest, 'wb') as lfile: - shutil.copyfileobj(response, lfile) - return True - - -# ---------------------------------------------------------- -# -# CI UTILS -# -# ----------------------------------------------------------- -def get_resolvconf_ns(): - """ - Get nameservers from current resolv.conf - """ - nameservers = [] - rconf = open("/etc/resolv.conf", "r") - line = rconf.readline() - resolver = dns.resolver.Resolver() - while line: - addr_ip = re.search(r"\b(?:[0-9]{1,3}\.){3}[0-9]{1,3}\b", line) - if addr_ip: - resolver.nameservers = [addr_ip.group(0)] - try: - result = resolver.query('opnfv.org')[0] - if result != "": - nameservers.append(addr_ip.group()) - except dns.exception.Timeout: - pass - line = rconf.readline() - return nameservers - - -def get_ci_envvars(): - """ - Get the CI env variables - """ - ci_env_var = { - "installer": os.environ.get('INSTALLER_TYPE'), - "scenario": os.environ.get('DEPLOY_SCENARIO')} - return ci_env_var - - def execute_command_raise(cmd, info=False, error_msg="", verbose=True, output_file=None): ret = execute_command(cmd, info, error_msg, verbose, output_file) @@ -122,7 +46,7 @@ def execute_command(cmd, info=False, error_msg="", ofd.write(line) else: line = line.replace('\n', '') - print (line) + print(line) sys.stdout.flush() if output_file: ofd.close() @@ -135,33 +59,6 @@ def execute_command(cmd, info=False, error_msg="", return returncode -def get_dict_by_test(testname): - # pylint: disable=bad-continuation - with open(pkg_resources.resource_filename( - 'functest', 'ci/testcases.yaml')) as tyaml: - testcases_yaml = yaml.safe_load(tyaml) - - for dic_tier in testcases_yaml.get("tiers"): - for dic_testcase in dic_tier['testcases']: - if dic_testcase['case_name'] == testname: - return dic_testcase - - LOGGER.error('Project %s is not defined in testcases.yaml', testname) - return None - - -def get_criteria_by_test(testname): - tdict = get_dict_by_test(testname) - if tdict: - return tdict['criteria'] - return None - - -# ---------------------------------------------------------- -# -# YAML UTILS -# -# ----------------------------------------------------------- def get_parameter_from_yaml(parameter, yfile): """ Returns the value of a given parameter in file.yaml @@ -179,18 +76,93 @@ def get_parameter_from_yaml(parameter, yfile): return value -def get_functest_config(parameter): - yaml_ = constants.CONST.__getattribute__('CONFIG_FUNCTEST_YAML') - return get_parameter_from_yaml(parameter, yaml_) +def get_nova_version(cloud): + """ Get Nova API microversion + + Returns: + + - Nova API microversion + - None on operation error + """ + # pylint: disable=protected-access + try: + request = cloud._compute_client.request("/", "GET") + LOGGER.debug('cloud._compute_client.request: %s', request) + version = request["version"]["version"] + major, minor = version.split('.') + LOGGER.debug('nova version: %s', (int(major), int(minor))) + return (int(major), int(minor)) + except Exception: # pylint: disable=broad-except + LOGGER.exception("Cannot detect Nova version") + return None + + +def get_openstack_version(cloud): + """ Detect OpenStack version via Nova API microversion + It follows `MicroversionHistory + `_. -def get_functest_yaml(): - # pylint: disable=bad-continuation - with open(constants.CONST.__getattribute__( - 'CONFIG_FUNCTEST_YAML')) as yaml_fd: - functest_yaml = yaml.safe_load(yaml_fd) - return functest_yaml + Returns: + + - OpenStack release + - Unknown on operation error + """ + version = get_nova_version(cloud) + try: + assert version + if version > (2, 65): + osversion = "Master" + elif version > (2, 60): + osversion = "Rocky" + elif version > (2, 53): + osversion = "Queens" + elif version > (2, 42): + osversion = "Pike" + elif version > (2, 38): + osversion = "Ocata" + elif version > (2, 25): + osversion = "Newton" + elif version > (2, 12): + osversion = "Mitaka" + elif version > (2, 3): + osversion = "Liberty" + elif version >= (2, 1): + osversion = "Kilo" + else: + osversion = "Unknown" + LOGGER.info('Detect OpenStack version: %s', osversion) + return osversion + except AssertionError: + LOGGER.exception("Cannot detect OpenStack version") + return "Unknown" + + +def convert_dict_to_ini(value): + "Convert dict to oslo.conf input" + assert isinstance(value, dict) + return ",".join("{}:{}".format( + key, val) for (key, val) in six.iteritems(value)) + + +def convert_list_to_ini(value): + "Convert list to oslo.conf input" + assert isinstance(value, list) + return ",".join("{}".format(val) for val in value) + + +def convert_ini_to_dict(value): + "Convert oslo.conf input to dict" + assert isinstance(value, str) + try: + return {k: v for k, v in (x.rsplit(':', 1) for x in value.split(','))} + except ValueError: + return {} -def print_separator(): - LOGGER.info("==============================================") +def convert_ini_to_list(value): + "Convert list to oslo.conf input" + assert isinstance(value, str) + if not value: + return [] + return [x for x in value.split(',')]