"""
import argparse
-import enum
import importlib
import logging
import logging.config
import textwrap
import pkg_resources
+import enum
import prettytable
+import yaml
import functest.ci.tier_builder as tb
import functest.core.testcase as testcase
-import functest.utils.functest_utils as ft_utils
-import functest.utils.openstack_utils as os_utils
-from functest.utils.constants import CONST
-# __name__ cannot be used here
LOGGER = logging.getLogger('functest.ci.run_tests')
-
-CONFIG_FUNCTEST_PATH = pkg_resources.resource_filename(
- 'functest', 'ci/config_functest.yaml')
+ENV_FILE = "/home/opnfv/functest/conf/env_file"
class Result(enum.Enum):
self.overall_result = Result.EX_OK
self.clean_flag = True
self.report_flag = False
- self._tiers = tb.TierBuilder(
- CONST.__getattribute__('INSTALLER_TYPE'),
- CONST.__getattribute__('DEPLOY_SCENARIO'),
+ self.tiers = tb.TierBuilder(
+ os.environ.get('INSTALLER_TYPE', None),
+ os.environ.get('DEPLOY_SCENARIO', None),
pkg_resources.resource_filename('functest', 'ci/testcases.yaml'))
@staticmethod
- def source_rc_file():
- """Set the environmental vars from openstack.creds"""
-
- rc_file = CONST.__getattribute__('openstack_creds')
+ def source_envfile(rc_file=ENV_FILE):
+ """Source the env file passed as arg"""
if not os.path.isfile(rc_file):
- raise Exception("RC file %s does not exist..." % rc_file)
- LOGGER.debug("Sourcing the OpenStack RC file...")
- os_utils.source_credentials(rc_file)
+ LOGGER.debug("No env file %s found", rc_file)
+ return
+ with open(rc_file, "r") as rcfd:
+ LOGGER.info("Sourcing env file %s", rc_file)
+ for line in rcfd:
+ var = (line.rstrip('"\n').replace('export ', '').split(
+ "=") if re.search(r'(.*)=(.*)', line) else None)
+ # The two next lines should be modified as soon as rc_file
+ # conforms with common rules. Be aware that it could induce
+ # issues if value starts with '
+ if var:
+ key = re.sub(r'^["\' ]*|[ \'"]*$', '', var[0])
+ value = re.sub(r'^["\' ]*|[ \'"]*$', '', "".join(var[1:]))
+ os.environ[key] = value
+
+ @staticmethod
+ def get_dict_by_test(testname):
+ # pylint: disable=bad-continuation,missing-docstring
+ 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
@staticmethod
def get_run_dict(testname):
- """Obtain the the 'run' block of the testcase from testcases.yaml"""
+ """Obtain the 'run' block of the testcase from testcases.yaml"""
try:
- dic_testcase = ft_utils.get_dict_by_test(testname)
+ dic_testcase = Runner.get_dict_by_test(testname)
if not dic_testcase:
LOGGER.error("Cannot get %s's config options", testname)
elif 'run' in dic_testcase:
try:
module = importlib.import_module(run_dict['module'])
cls = getattr(module, run_dict['class'])
- test_dict = ft_utils.get_dict_by_test(test.get_name())
+ test_dict = Runner.get_dict_by_test(test.get_name())
test_case = cls(**test_dict)
self.executed_test_cases[test.get_name()] = test_case
try:
header_style='upper', padding_width=5,
field_names=['tiers', 'order', 'CI Loop', 'description',
'testcases'])
- for tier in self._tiers.get_tiers():
+ for tier in self.tiers.get_tiers():
+ ci_loop = os.environ.get('CI_LOOP', 'daily')
if (tier.get_tests() and
- re.search(CONST.__getattribute__('CI_LOOP'),
- tier.get_ci_loop()) is not None):
+ re.search(ci_loop, tier.get_ci_loop()) is not None):
tiers_to_run.append(tier)
msg.add_row([tier.get_name(), tier.get_order(),
tier.get_ci_loop(),
self.report_flag = kwargs['report']
try:
if 'test' in kwargs:
- self.source_rc_file()
+ LOGGER.debug("Sourcing the credential file...")
+ self.source_envfile()
+
LOGGER.debug("Test args: %s", kwargs['test'])
- if self._tiers.get_tier(kwargs['test']):
- self.run_tier(self._tiers.get_tier(kwargs['test']))
- elif self._tiers.get_test(kwargs['test']):
+ if self.tiers.get_tier(kwargs['test']):
+ self.run_tier(self.tiers.get_tier(kwargs['test']))
+ elif self.tiers.get_test(kwargs['test']):
result = self.run_test(
- self._tiers.get_test(kwargs['test']))
+ self.tiers.get_test(kwargs['test']))
if result != testcase.TestCase.EX_OK:
LOGGER.error("The test case '%s' failed.",
kwargs['test'])
LOGGER.error("Unknown test case or tier '%s', or not "
"supported by the given scenario '%s'.",
kwargs['test'],
- CONST.__getattribute__('DEPLOY_SCENARIO'))
+ os.environ.get('DEPLOY_SCENARIO', ""))
LOGGER.debug("Available tiers are:\n\n%s",
- self._tiers)
+ self.tiers)
return Result.EX_ERROR
else:
self.run_all()
except Exception: # pylint: disable=broad-except
LOGGER.exception("Failures when running testcase(s)")
self.overall_result = Result.EX_ERROR
- if not self._tiers.get_test(kwargs['test']):
- self.summary(self._tiers.get_tier(kwargs['test']))
+ if not self.tiers.get_test(kwargs['test']):
+ self.summary(self.tiers.get_tier(kwargs['test']))
LOGGER.info("Execution exit value: %s", self.overall_result)
return self.overall_result
field_names=['env var', 'value'])
for env_var in ['INSTALLER_TYPE', 'DEPLOY_SCENARIO', 'BUILD_TAG',
'CI_LOOP']:
- msg.add_row([env_var, CONST.__getattribute__(env_var)])
+ msg.add_row([env_var, os.environ.get(env_var, "")])
LOGGER.info("Deployment description:\n\n%s\n", msg)
msg = prettytable.PrettyTable(
header_style='upper', padding_width=5,
field_names=['test case', 'project', 'tier',
'duration', 'result'])
- tiers = [tier] if tier else self._tiers.get_tiers()
+ tiers = [tier] if tier else self.tiers.get_tiers()
for each_tier in tiers:
for test in each_tier.get_tests():
try:
) == test_case.EX_OK) else 'FAIL'
msg.add_row(
[test_case.case_name, test_case.project_name,
- self._tiers.get_tier_name(test_case.case_name),
+ self.tiers.get_tier_name(test_case.case_name),
test_case.get_duration(), result])
for test in each_tier.get_skipped_test():
msg.add_row([test.get_name(), test.get_project(),
Code Review / functest.git / blobdiff