diff options
Diffstat (limited to 'utils/functest_utils.py')
| -rw-r--r-- | utils/functest_utils.py | 318 |
1 files changed, 318 insertions, 0 deletions
diff --git a/utils/functest_utils.py b/utils/functest_utils.py new file mode 100644 index 000000000..8eba43be8 --- /dev/null +++ b/utils/functest_utils.py @@ -0,0 +1,318 @@ +#!/usr/bin/env python +# +# jose.lausuch@ericsson.com +# valentin.boucher@orange.com +# 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 +# + +import json +import os +import os.path +import re +import requests +import shutil +import socket +import subprocess +import urllib2 +from git import Repo + + +# ---------------------------------------------------------- +# +# INTERNET UTILS +# +# ----------------------------------------------------------- +def check_internet_connectivity(url='http://www.opnfv.org/'): + """ + Check if there is access to the internet + """ + try: + urllib2.urlopen(url, timeout=5) + return True + except urllib2.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 = urllib2.urlopen(url) + except (urllib2.HTTPError, urllib2.URLError): + return False + + with open(dest, 'wb') as f: + shutil.copyfileobj(response, f) + return True + + +# ---------------------------------------------------------- +# +# CI UTILS +# +# ----------------------------------------------------------- +def get_git_branch(repo_path): + """ + Get git branch name + """ + repo = Repo(repo_path) + branch = repo.active_branch + return branch.name + + +def get_installer_type(logger=None): + """ + Get installer type (fuel, apex, joid, compass) + """ + try: + installer = os.environ['INSTALLER_TYPE'] + except KeyError: + if logger: + logger.error("Impossible to retrieve the installer type") + installer = "Unknown_installer" + + return installer + + +def get_scenario(logger=None): + """ + Get scenario + """ + try: + scenario = os.environ['DEPLOY_SCENARIO'] + except KeyError: + if logger: + logger.error("Impossible to retrieve the scenario") + scenario = "Unknown_scenario" + + return scenario + + +def get_version(logger=None): + """ + Get version + """ + # Use the build tag to retrieve the version + # By default version is unknown + # if launched through CI the build tag has the following format + # jenkins-<project>-<installer>-<pod>-<job>-<branch>-<id> + # e.g. jenkins-functest-fuel-opnfv-jump-2-daily-master-190 + # use regex to match branch info + rule = "daily-(.+?)-[0-9]*" + build_tag = get_build_tag(logger) + m = re.search(rule, build_tag) + if m: + return m.group(1) + else: + return "unknown" + + +def get_pod_name(logger=None): + """ + Get PoD Name from env variable NODE_NAME + """ + try: + return os.environ['NODE_NAME'] + except KeyError: + if logger: + logger.error( + "Unable to retrieve the POD name from environment. " + + "Using pod name 'unknown-pod'") + return "unknown-pod" + + +def get_build_tag(logger=None): + """ + Get build tag of jenkins jobs + """ + try: + build_tag = os.environ['BUILD_TAG'] + except KeyError: + if logger: + logger.error("Impossible to retrieve the build tag") + build_tag = "unknown_build_tag" + + return build_tag + + +def push_results_to_db(db_url, project, case_name, logger, pod_name, + version, scenario, criteria, build_tag, payload): + """ + POST results to the Result target DB + """ + url = db_url + "/results" + installer = get_installer_type(logger) + params = {"project_name": project, "case_name": case_name, + "pod_name": pod_name, "installer": installer, + "version": version, "scenario": scenario, "criteria": criteria, + "build_tag": build_tag, "details": payload} + + headers = {'Content-Type': 'application/json'} + try: + r = requests.post(url, data=json.dumps(params), headers=headers) + if logger: + logger.debug(r) + return True + except Exception, e: + print ("Error [push_results_to_db('%s', '%s', '%s', " + + "'%s', '%s', '%s', '%s', '%s', '%s')]:" % + (db_url, project, case_name, pod_name, version, + scenario, criteria, build_tag, payload), e) + return False + + +def get_resolvconf_ns(): + """ + Get nameservers from current resolv.conf + """ + nameservers = [] + rconf = open("/etc/resolv.conf", "r") + line = rconf.readline() + while line: + ip = re.search(r"\b(?:[0-9]{1,3}\.){3}[0-9]{1,3}\b", line) + sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + if ip: + result = sock.connect_ex((ip.group(), 53)) + if result == 0: + nameservers.append(ip.group()) + line = rconf.readline() + return nameservers + + +def getTestEnv(test, functest_yaml): + """ + Get the config of the testcase based on functest_config.yaml + 2 options + - test = test project e.g; ovno + - test = testcase e.g. functest/odl + look for the / to see if it is a test project or a testcase + """ + try: + TEST_ENV = functest_yaml.get("test-dependencies") + + if test.find("/") < 0: + config_test = TEST_ENV[test] + else: + test_split = test.split("/") + testproject = test_split[0] + testcase = test_split[1] + config_test = TEST_ENV[testproject][testcase] + except KeyError: + # if not defined in dependencies => no dependencies + config_test = "" + except Exception, e: + print "Error [getTestEnv]:", e + + return config_test + + +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 isTestRunnable(test, functest_yaml): + """ + Return True if the test is runnable in the current scenario + """ + # By default we assume that all the tests are always runnable... + is_runnable = True + # Retrieve CI environment + ci_env = get_ci_envvars() + # Retrieve test environement from config file + test_env = getTestEnv(test, functest_yaml) + + # if test_env not empty => dependencies to be checked + if test_env is not None and len(test_env) > 0: + # possible criteria = ["installer", "scenario"] + # consider test criteria from config file + # compare towards CI env through CI en variable + for criteria in test_env: + if re.search(test_env[criteria], ci_env[criteria]) is None: + # print "Test "+ test + " cannot be run on the environment" + is_runnable = False + return is_runnable + + +def generateTestcaseList(functest_yaml): + """ + Generate a test file with the runnable test according to + the current scenario + """ + test_list = "" + # get testcases + testcase_list = functest_yaml.get("test-dependencies") + projects = testcase_list.keys() + + for project in projects: + testcases = testcase_list[project] + # 1 or 2 levels for testcases project[/case]l + # if only project name without controller or scenario + # => shall be runnable on any controller/scenario + if testcases is None: + test_list += project + " " + else: + for testcase in testcases: + if testcase == "installer" or testcase == "scenario": + # project (1 level) + if isTestRunnable(project, functest_yaml): + test_list += project + " " + else: + # project/testcase (2 levels) + thetest = project + "/" + testcase + if isTestRunnable(thetest, functest_yaml): + test_list += testcase + " " + + # sort the list to execute the test in the right order + test_order_list = functest_yaml.get("test_exec_priority") + test_sorted_list = "" + for test in test_order_list: + if test_order_list[test] in test_list: + test_sorted_list += test_order_list[test] + " " + + # create a file that could be consumed by run-test.sh + # this method is used only for CI + # so it can be run only in container + # reuse default conf directory to store the list of runnable tests + file = open("/home/opnfv/functest/conf/testcase-list.txt", 'w') + file.write(test_sorted_list) + file.close() + + return test_sorted_list + + +def execute_command(cmd, logger=None, exit_on_error=True): + """ + Execute Linux command + prints stdout to a file and depending on if there + is a logger defined, it will print it or not. + """ + if logger: + logger.debug('Executing command : {}'.format(cmd)) + output_file = "output.txt" + f = open(output_file, 'w+') + p = subprocess.call(cmd, shell=True, stdout=f, stderr=subprocess.STDOUT) + f.close() + f = open(output_file, 'r') + result = f.read() + if result != "" and logger: + logger.debug(result) + if p == 0: + return True + else: + if logger: + logger.error("Error when executing command %s" % cmd) + if exit_on_error: + exit(-1) + return False |