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)
- """
- try:
- installer = os.environ['INSTALLER_TYPE']
- except KeyError:
- logger.error("Impossible to retrieve the installer type")
- installer = "Unknown_installer"
-
- return installer
-
-
-def get_scenario():
- """
- Get scenario
- """
- try:
- scenario = os.environ['DEPLOY_SCENARIO']
- except KeyError:
- logger.info("Impossible to retrieve the scenario."
- "Use default os-nosdn-nofeature-noha")
- scenario = "os-nosdn-nofeature-noha"
-
- return scenario
-
-
def get_version():
"""
Get version
# 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"
-def get_pod_name():
- """
- Get PoD Name from env variable NODE_NAME
- """
- try:
- return os.environ['NODE_NAME']
- except KeyError:
- logger.info(
- "Unable to retrieve the POD name from environment. " +
- "Using pod name 'unknown-pod'")
- 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()
+ pod_name = CONST.__getattribute__('NODE_NAME')
+ scenario = CONST.__getattribute__('DEPLOY_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']
error = None
headers = {'Content-Type': 'application/json'}
try:
- r = requests.post(url, data=json.dumps(params), headers=headers)
+ r = requests.post(url, data=json.dumps(params, sort_keys=True),
+ headers=headers)
logger.debug(r)
r.raise_for_status()
except requests.RequestException as exc:
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"):
def get_functest_config(parameter):
- yaml_ = os.environ["CONFIG_FUNCTEST_YAML"]
+ yaml_ = constants.CONST.__getattribute__('CONFIG_FUNCTEST_YAML')
return get_parameter_from_yaml(parameter, yaml_)
-def check_success_rate(case_name, result):
- # It should be removed as TestCase tests criteria
- # and result.
- logger.warning('check_success_rate will be removed soon')
- criteria = get_criteria_by_test(case_name)
- if type(criteria) == int and result >= criteria:
- return 100
- return 0
-
-
def merge_dicts(dict1, dict2):
for k in set(dict1.keys()).union(dict2.keys()):
if k in dict1 and k in dict2:
yield (k, dict2[k])
-def get_testcases_file_dir():
- return get_functest_config('general.functest.testcases_yaml')
-
-
def get_functest_yaml():
- with open(os.environ["CONFIG_FUNCTEST_YAML"]) as f:
+ with open(constants.CONST.__getattribute__('CONFIG_FUNCTEST_YAML')) as f:
functest_yaml = yaml.safe_load(f)
f.close()
return functest_yaml