import json
import logging
import os
+import pkg_resources
import re
import shutil
import subprocess
import requests
from six.moves import urllib
import yaml
-from git import Repo
from functest.utils import constants
from functest.utils import decorators
+from functest.utils.constants import CONST
logger = logging.getLogger(__name__)
# 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():
"""
Get installer type (fuel, apex, joid, compass)
# jenkins-functest-fuel-baremetal-weekly-master-8
# use regex to match branch info
rule = "(dai|week)ly-(.+?)-[0-9]*"
- build_tag = get_build_tag()
+ build_tag = CONST.__getattribute__('BUILD_TAG')
+ if not build_tag:
+ build_tag = 'none'
m = re.search(rule, build_tag)
if m:
return m.group(2)
return "unknown-pod"
-def get_build_tag():
- """
- Get build tag of jenkins jobs
- """
- try:
- build_tag = os.environ['BUILD_TAG']
- except KeyError:
- logger.info("Impossible to retrieve the build tag")
- build_tag = "none"
-
- return build_tag
-
-
-def get_db_url():
+def logger_test_results(project, case_name, status, details):
"""
- Returns DB URL
+ Format test case results for the logger
"""
- # TODO use CONST mechanism
- try:
- # if TEST_DB_URL declared in env variable, use it!
- db_url = os.environ['TEST_DB_URL']
- except KeyError:
- db_url = get_functest_config('results.test_db_url')
- return db_url
-
-
-def logger_test_results(project, case_name, status, details):
pod_name = get_pod_name()
scenario = get_scenario()
version = get_version()
- build_tag = get_build_tag()
+ build_tag = CONST.__getattribute__('BUILD_TAG')
+ db_url = CONST.__getattribute__("results_test_db_url")
logger.info(
"\n"
"details:\t%(d)s\n"
% {'p': project,
'n': case_name,
- 'db': get_db_url(),
+ 'db': db_url,
'pod': pod_name,
'v': version,
's': scenario,
POST results to the Result target DB
"""
# Retrieve params from CI and conf
- url = get_db_url()
+ url = CONST.__getattribute__("results_test_db_url")
try:
installer = os.environ['INSTALLER_TYPE']
def execute_command_raise(cmd, info=False, error_msg="",
- verbose=True, output_file=None):
- ret = execute_command(cmd, info, error_msg, verbose, output_file)
+ verbose=True, output_file=None, env=None):
+ ret = execute_command(cmd, info, error_msg, verbose, output_file, env)
if ret != 0:
raise Exception(error_msg)
def execute_command(cmd, info=False, error_msg="",
- verbose=True, output_file=None):
+ verbose=True, output_file=None, env=None):
if not error_msg:
error_msg = ("The command '%s' failed." % cmd)
msg_exec = ("Executing command: '%s'" % cmd)
logger.info(msg_exec)
else:
logger.debug(msg_exec)
- p = subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE,
+ p = subprocess.Popen(cmd, env=env, shell=True, stdout=subprocess.PIPE,
stderr=subprocess.STDOUT)
if output_file:
f = open(output_file, "w")
def get_dict_by_test(testname):
- with open(get_testcases_file_dir()) as f:
+ with open(pkg_resources.resource_filename(
+ 'functest', 'ci/testcases.yaml')) as f:
testcases_yaml = yaml.safe_load(f)
for dic_tier in testcases_yaml.get("tiers"):
yield (k, dict2[k])
-def get_testcases_file_dir():
- return get_functest_config('general.functest.testcases_yaml')
-
-
def get_functest_yaml():
with open(constants.CONST.__getattribute__('CONFIG_FUNCTEST_YAML')) as f:
functest_yaml = yaml.safe_load(f)