diff options
Diffstat (limited to 'testcases/functest_utils.py')
| -rw-r--r-- | testcases/functest_utils.py | 318 |
1 files changed, 0 insertions, 318 deletions
diff --git a/testcases/functest_utils.py b/testcases/functest_utils.py deleted file mode 100644 index 8eba43be..00000000 --- a/testcases/functest_utils.py +++ /dev/null @@ -1,318 +0,0 @@ -#!/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 |