3 # Copyright (c) 2016 Ericsson AB and others.
5 # All rights reserved. This program and the accompanying materials
6 # are made available under the terms of the Apache License, Version 2.0
7 # which accompanies this distribution, and is available at
8 # http://www.apache.org/licenses/LICENSE-2.0
10 """ The entry of running tests:
11 1) Parses functest/ci/testcases.yaml to check which testcase(s) to be run
12 2) Execute the common operations on every testcase (run, push results to db...)
13 3) Return the right status code
29 import functest.ci.tier_builder as tb
30 import functest.core.testcase as testcase
31 import functest.utils.functest_utils as ft_utils
32 from functest.utils.constants import CONST
34 # __name__ cannot be used here
35 LOGGER = logging.getLogger('functest.ci.run_tests')
37 CONFIG_FUNCTEST_PATH = pkg_resources.resource_filename(
38 'functest', 'ci/config_functest.yaml')
41 class Result(enum.Enum):
42 """The overall result in enumerated type"""
43 # pylint: disable=too-few-public-methods
48 class BlockingTestFailed(Exception):
49 """Exception when the blocking test fails"""
53 class TestNotEnabled(Exception):
54 """Exception when the test is not enabled"""
58 class RunTestsParser(object):
59 """Parser to run tests"""
60 # pylint: disable=too-few-public-methods
63 self.parser = argparse.ArgumentParser()
64 self.parser.add_argument("-t", "--test", dest="test", action='store',
65 help="Test case or tier (group of tests) "
66 "to be executed. It will run all the test "
68 self.parser.add_argument("-n", "--noclean", help="Do not clean "
69 "OpenStack resources after running each "
70 "test (default=false).",
72 self.parser.add_argument("-r", "--report", help="Push results to "
73 "database (default=false).",
76 def parse_args(self, argv=None):
79 It can call sys.exit if arguments are incorrect.
82 the arguments from cmdline
84 return vars(self.parser.parse_args(argv))
91 self.executed_test_cases = {}
92 self.overall_result = Result.EX_OK
93 self.clean_flag = True
94 self.report_flag = False
95 self.tiers = tb.TierBuilder(
96 CONST.__getattribute__('INSTALLER_TYPE'),
97 CONST.__getattribute__('DEPLOY_SCENARIO'),
98 pkg_resources.resource_filename('functest', 'ci/testcases.yaml'))
101 def source_envfile(rc_file):
102 """Source the env file passed as arg"""
103 with open(rc_file, "r") as rcfd:
105 var = (line.rstrip('"\n').replace('export ', '').split(
106 "=") if re.search(r'(.*)=(.*)', line) else None)
107 # The two next lines should be modified as soon as rc_file
108 # conforms with common rules. Be aware that it could induce
109 # issues if value starts with '
111 key = re.sub(r'^["\' ]*|[ \'"]*$', '', var[0])
112 value = re.sub(r'^["\' ]*|[ \'"]*$', '', "".join(var[1:]))
113 os.environ[key] = value
114 setattr(CONST, key, value)
117 def get_run_dict(testname):
118 """Obtain the 'run' block of the testcase from testcases.yaml"""
120 dic_testcase = ft_utils.get_dict_by_test(testname)
122 LOGGER.error("Cannot get %s's config options", testname)
123 elif 'run' in dic_testcase:
124 return dic_testcase['run']
126 except Exception: # pylint: disable=broad-except
127 LOGGER.exception("Cannot get %s's config options", testname)
130 def run_test(self, test):
131 """Run one test case"""
132 if not test.is_enabled():
133 raise TestNotEnabled(
134 "The test case {} is not enabled".format(test.get_name()))
135 LOGGER.info("Running test case '%s'...", test.get_name())
136 result = testcase.TestCase.EX_RUN_ERROR
137 run_dict = self.get_run_dict(test.get_name())
140 module = importlib.import_module(run_dict['module'])
141 cls = getattr(module, run_dict['class'])
142 test_dict = ft_utils.get_dict_by_test(test.get_name())
143 test_case = cls(**test_dict)
144 self.executed_test_cases[test.get_name()] = test_case
146 kwargs = run_dict['args']
147 test_case.run(**kwargs)
151 test_case.push_to_db()
152 if test.get_project() == "functest":
153 result = test_case.is_successful()
155 result = testcase.TestCase.EX_OK
156 LOGGER.info("Test result:\n\n%s\n", test_case)
160 LOGGER.exception("Cannot import module %s", run_dict['module'])
161 except AttributeError:
162 LOGGER.exception("Cannot get class %s", run_dict['class'])
164 raise Exception("Cannot import the class for the test case.")
167 def run_tier(self, tier):
169 tier_name = tier.get_name()
170 tests = tier.get_tests()
172 LOGGER.info("There are no supported test cases in this tier "
173 "for the given scenario")
174 self.overall_result = Result.EX_ERROR
176 LOGGER.info("Running tier '%s'", tier_name)
179 test_case = self.executed_test_cases[test.get_name()]
180 if test_case.is_successful() != testcase.TestCase.EX_OK:
181 LOGGER.error("The test case '%s' failed.", test.get_name())
182 if test.get_project() == "functest":
183 self.overall_result = Result.EX_ERROR
184 if test.is_blocking():
185 raise BlockingTestFailed(
186 "The test case {} failed and is blocking".format(
188 return self.overall_result
191 """Run all available testcases"""
193 msg = prettytable.PrettyTable(
194 header_style='upper', padding_width=5,
195 field_names=['tiers', 'order', 'CI Loop', 'description',
197 for tier in self.tiers.get_tiers():
198 if (tier.get_tests() and
199 re.search(CONST.__getattribute__('CI_LOOP'),
200 tier.get_ci_loop()) is not None):
201 tiers_to_run.append(tier)
202 msg.add_row([tier.get_name(), tier.get_order(),
204 textwrap.fill(tier.description, width=40),
205 textwrap.fill(' '.join([str(x.get_name(
206 )) for x in tier.get_tests()]), width=40)])
207 LOGGER.info("TESTS TO BE EXECUTED:\n\n%s\n", msg)
208 for tier in tiers_to_run:
211 def main(self, **kwargs):
212 """Entry point of class Runner"""
213 if 'noclean' in kwargs:
214 self.clean_flag = not kwargs['noclean']
215 if 'report' in kwargs:
216 self.report_flag = kwargs['report']
219 LOGGER.debug("Sourcing the credential file...")
220 self.source_envfile(getattr(CONST, 'env_file'))
222 LOGGER.debug("Test args: %s", kwargs['test'])
223 if self.tiers.get_tier(kwargs['test']):
224 self.run_tier(self.tiers.get_tier(kwargs['test']))
225 elif self.tiers.get_test(kwargs['test']):
226 result = self.run_test(
227 self.tiers.get_test(kwargs['test']))
228 if result != testcase.TestCase.EX_OK:
229 LOGGER.error("The test case '%s' failed.",
231 self.overall_result = Result.EX_ERROR
232 elif kwargs['test'] == "all":
235 LOGGER.error("Unknown test case or tier '%s', or not "
236 "supported by the given scenario '%s'.",
238 CONST.__getattribute__('DEPLOY_SCENARIO'))
239 LOGGER.debug("Available tiers are:\n\n%s",
241 return Result.EX_ERROR
244 except BlockingTestFailed:
246 except Exception: # pylint: disable=broad-except
247 LOGGER.exception("Failures when running testcase(s)")
248 self.overall_result = Result.EX_ERROR
249 if not self.tiers.get_test(kwargs['test']):
250 self.summary(self.tiers.get_tier(kwargs['test']))
251 LOGGER.info("Execution exit value: %s", self.overall_result)
252 return self.overall_result
254 def summary(self, tier=None):
255 """To generate functest report showing the overall results"""
256 msg = prettytable.PrettyTable(
257 header_style='upper', padding_width=5,
258 field_names=['env var', 'value'])
259 for env_var in ['INSTALLER_TYPE', 'DEPLOY_SCENARIO', 'BUILD_TAG',
261 msg.add_row([env_var, CONST.__getattribute__(env_var)])
262 LOGGER.info("Deployment description:\n\n%s\n", msg)
263 msg = prettytable.PrettyTable(
264 header_style='upper', padding_width=5,
265 field_names=['test case', 'project', 'tier',
266 'duration', 'result'])
267 tiers = [tier] if tier else self.tiers.get_tiers()
268 for each_tier in tiers:
269 for test in each_tier.get_tests():
271 test_case = self.executed_test_cases[test.get_name()]
273 msg.add_row([test.get_name(), test.get_project(),
274 each_tier.get_name(), "00:00", "SKIP"])
276 result = 'PASS' if(test_case.is_successful(
277 ) == test_case.EX_OK) else 'FAIL'
279 [test_case.case_name, test_case.project_name,
280 self.tiers.get_tier_name(test_case.case_name),
281 test_case.get_duration(), result])
282 for test in each_tier.get_skipped_test():
283 msg.add_row([test.get_name(), test.get_project(),
284 each_tier.get_name(), "00:00", "SKIP"])
285 LOGGER.info("FUNCTEST REPORT:\n\n%s\n", msg)
290 logging.config.fileConfig(pkg_resources.resource_filename(
291 'functest', 'ci/logging.ini'))
292 logging.captureWarnings(True)
293 parser = RunTestsParser()
294 args = parser.parse_args(sys.argv[1:])
296 return runner.main(**args).value