Merge "Remove useless odlreport2db args"
[functest.git] / ci / run_tests.py
1 #!/usr/bin/env python
2 #
3 # Author: Jose Lausuch (jose.lausuch@ericsson.com)
4 #
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
9 #
10
11 import argparse
12 import datetime
13 import os
14 import re
15 import sys
16 import functest.ci.generate_report as generate_report
17 import functest.ci.tier_builder as tb
18 import functest.utils.functest_logger as ft_logger
19 import functest.utils.functest_utils as ft_utils
20 import functest.utils.openstack_clean as os_clean
21 import functest.utils.openstack_snapshot as os_snapshot
22 import functest.utils.openstack_utils as os_utils
23
24
25 parser = argparse.ArgumentParser()
26 parser.add_argument("-t", "--test", dest="test", action='store',
27                     help="Test case or tier (group of tests) to be executed. "
28                     "It will run all the test if not specified.")
29 parser.add_argument("-n", "--noclean", help="Do not clean OpenStack resources"
30                     " after running each test (default=false).",
31                     action="store_true")
32 parser.add_argument("-r", "--report", help="Push results to database "
33                     "(default=false).", action="store_true")
34 args = parser.parse_args()
35
36
37 """ logging configuration """
38 logger = ft_logger.Logger("run_tests").getLogger()
39
40
41 """ global variables """
42 REPOS_DIR = os.getenv('repos_dir')
43 FUNCTEST_REPO = ("%s/functest/" % REPOS_DIR)
44 EXEC_SCRIPT = ("%sci/exec_test.sh" % FUNCTEST_REPO)
45 CLEAN_FLAG = True
46 REPORT_FLAG = False
47 EXECUTED_TEST_CASES = []
48
49 # This will be the return code of this script. If any of the tests fails,
50 # this variable will change to -1
51 OVERALL_RESULT = 0
52
53
54 def print_separator(str, count=45):
55     line = ""
56     for i in range(0, count - 1):
57         line += str
58     logger.info("%s" % line)
59
60
61 def source_rc_file():
62     rc_file = os.getenv('creds')
63     if not os.path.isfile(rc_file):
64         logger.error("RC file %s does not exist..." % rc_file)
65         sys.exit(1)
66     logger.debug("Sourcing the OpenStack RC file...")
67     os_utils.source_credentials(rc_file)
68
69
70 def generate_os_snapshot():
71     os_snapshot.main()
72
73
74 def cleanup():
75     os_clean.main()
76
77
78 def update_test_info(test_name, result, duration):
79     for test in EXECUTED_TEST_CASES:
80         if test['test_name'] == test_name:
81             test.update({"result": result,
82                          "duration": duration})
83
84
85 def run_test(test, tier_name):
86     global OVERALL_RESULT, EXECUTED_TEST_CASES
87     result_str = "PASS"
88     start = datetime.datetime.now()
89     test_name = test.get_name()
90     logger.info("\n")  # blank line
91     print_separator("=")
92     logger.info("Running test case '%s'..." % test_name)
93     print_separator("=")
94     logger.debug("\n%s" % test)
95
96     if CLEAN_FLAG:
97         generate_os_snapshot()
98
99     flags = (" -t %s" % (test_name))
100     if REPORT_FLAG:
101         flags += " -r"
102
103     cmd = ("%s%s" % (EXEC_SCRIPT, flags))
104     logger.debug("Executing command '%s'" % cmd)
105     result = ft_utils.execute_command(cmd, logger, exit_on_error=False)
106
107     if CLEAN_FLAG:
108         cleanup()
109     end = datetime.datetime.now()
110     duration = (end - start).seconds
111     duration_str = ("%02d:%02d" % divmod(duration, 60))
112     logger.info("Test execution time: %s" % duration_str)
113
114     if result != 0:
115         logger.error("The test case '%s' failed. " % test_name)
116         OVERALL_RESULT = -1
117         result_str = "FAIL"
118
119         if test.is_blocking():
120             if not args.test or args.test == "all":
121                 logger.info("This test case is blocking. Aborting overall "
122                             "execution.")
123                 # if it is a single test we don't print the whole results table
124                 update_test_info(test_name, result_str, duration_str)
125                 generate_report.main(EXECUTED_TEST_CASES)
126             logger.info("Execution exit value: %s" % OVERALL_RESULT)
127             sys.exit(OVERALL_RESULT)
128
129     update_test_info(test_name, result_str, duration_str)
130
131
132 def run_tier(tier):
133     tier_name = tier.get_name()
134     tests = tier.get_tests()
135     if tests is None or len(tests) == 0:
136         logger.info("There are no supported test cases in this tier "
137                     "for the given scenario")
138         return 0
139     logger.info("\n\n")  # blank line
140     print_separator("#")
141     logger.info("Running tier '%s'" % tier_name)
142     print_separator("#")
143     logger.debug("\n%s" % tier)
144     for test in tests:
145         run_test(test, tier_name)
146
147
148 def run_all(tiers):
149     global EXECUTED_TEST_CASES
150     summary = ""
151     BUILD_TAG = os.getenv('BUILD_TAG')
152     if BUILD_TAG is not None and re.search("daily", BUILD_TAG) is not None:
153         CI_LOOP = "daily"
154     else:
155         CI_LOOP = "weekly"
156
157     tiers_to_run = []
158
159     for tier in tiers.get_tiers():
160         if (len(tier.get_tests()) != 0 and
161                 re.search(CI_LOOP, tier.get_ci_loop()) is not None):
162             tiers_to_run.append(tier)
163             summary += ("\n    - %s:\n\t   %s"
164                         % (tier.get_name(),
165                            tier.get_test_names()))
166
167     logger.info("Tests to be executed:%s" % summary)
168     EXECUTED_TEST_CASES = generate_report.init(tiers_to_run)
169     for tier in tiers_to_run:
170         run_tier(tier)
171
172     generate_report.main(EXECUTED_TEST_CASES)
173
174
175 def main():
176     global CLEAN_FLAG
177     global REPORT_FLAG
178
179     CI_INSTALLER_TYPE = os.getenv('INSTALLER_TYPE')
180     CI_SCENARIO = os.getenv('DEPLOY_SCENARIO')
181
182     file = FUNCTEST_REPO + "/ci/testcases.yaml"
183     _tiers = tb.TierBuilder(CI_INSTALLER_TYPE, CI_SCENARIO, file)
184
185     if args.noclean:
186         CLEAN_FLAG = False
187
188     if args.report:
189         REPORT_FLAG = True
190
191     if args.test:
192         source_rc_file()
193         if _tiers.get_tier(args.test):
194             run_tier(_tiers.get_tier(args.test))
195
196         elif _tiers.get_test(args.test):
197             run_test(_tiers.get_test(args.test), _tiers.get_tier(args.test))
198
199         elif args.test == "all":
200             run_all(_tiers)
201
202         else:
203             logger.error("Unknown test case or tier '%s', or not supported by "
204                          "the given scenario '%s'."
205                          % (args.test, CI_SCENARIO))
206             logger.debug("Available tiers are:\n\n%s"
207                          % _tiers)
208     else:
209         run_all(_tiers)
210
211     logger.info("Execution exit value: %s" % OVERALL_RESULT)
212     sys.exit(OVERALL_RESULT)
213
214 if __name__ == '__main__':
215     main()