"""
import argparse
-import enum
import importlib
import logging
import logging.config
import textwrap
import pkg_resources
+import enum
import prettytable
import functest.ci.tier_builder as tb
self.overall_result = Result.EX_OK
self.clean_flag = True
self.report_flag = False
- self._tiers = tb.TierBuilder(
+ self.tiers = tb.TierBuilder(
CONST.__getattribute__('INSTALLER_TYPE'),
CONST.__getattribute__('DEPLOY_SCENARIO'),
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')
- 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)
-
@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)
if not dic_testcase:
test_dict = ft_utils.get_dict_by_test(test.get_name())
test_case = cls(**test_dict)
self.executed_test_cases[test.get_name()] = test_case
- if self.clean_flag:
- if test_case.create_snapshot() != test_case.EX_OK:
- return testcase.TestCase.EX_RUN_ERROR
try:
kwargs = run_dict['args']
test_case.run(**kwargs)
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():
if (tier.get_tests() and
re.search(CONST.__getattribute__('CI_LOOP'),
tier.get_ci_loop()) is not None):
self.report_flag = kwargs['report']
try:
if 'test' in kwargs:
- self.source_rc_file()
+ LOGGER.debug("Sourcing the credential file...")
+ os_utils.source_credentials(CONST.__getattribute__('env_file'))
+
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'])
kwargs['test'],
CONST.__getattribute__('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
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(),