1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
|
#!/usr/bin/env python3
#
# SPDX-FileCopyrightText: 2024 Kienan Stewart <kstewart@efficios.com>
# SPDX-License-Identifier: GPL-2.0-only
#
"""
This script runs the tests defined in the environment variable
`SERIAL_TESTS` one at a time in subprocesses.
For each serial test, `<test>.log` and `<test>.trs` will be produced. The log file
will contain the line-buffered combined stdout and stderr from the test itself.
The trs file will be formatted as an autotools trs file.
This test cannot know how many subtests are going to be run before hand, so all
tests are run and the output produced before emitting a TAP plan and creating it's
own test and trs log files.
For each test (that is - a TAP `ok`, `not ok`, and `skip` a single line following
the rough format of the autotools `make check` wrapper will be output to stdout.
PASS|SKIP|FAIL: testname NN - Description
"""
import logging
import os
import pathlib
import re
import shlex
import shutil
import subprocess
import sys
import signal
import time
test_utils_import_path = pathlib.Path(__file__).absolute().parents[1] / "utils"
sys.path.append(str(test_utils_import_path))
import lttngtest
# This is not tap 14 compliant, but should suffice
TAP_PLAN_LINE_RE = re.compile(r"^1\.\.(\d+)$", re.MULTILINE)
TAP_OK_LINE_RE = re.compile(r"^ok \d+.*$", re.MULTILINE)
TAP_SKIP_LINE_RE = re.compile(r"skip \d+.*$", re.MULTILINE)
TAP_NOT_OK_LINE_RE = re.compile(r"^not ok \d+.*$", re.MULTILINE)
test_being_aborted = False
current_test_process = None
def handle_abort(signum, frame):
global test_being_aborted
# signal.Signals only exists in python 3.5+
try:
signal_name = signal.Signals(signum).name
except AttributeError:
signal_name = "signal #{}".format(signum)
logging.warning("Received {}, aborting current test".format(signal_name))
test_being_aborted = True
if current_test_process:
logging.info("Forwarding {} to test process".format(signal_name))
try:
current_test_process.send_signal(signum)
current_test_process.wait(timeout=10)
except subprocess.TimeoutExpired:
logging.warning(
"Test process did not terminate before timeout, sending SIGKILL"
)
current_test_process.kill()
current_test_process.wait()
print(current_test_process.stdout.read().decode("utf-8"))
else:
logging.debug("No test process to terminate")
signal.signal(signal.SIGTERM, handle_abort)
signal.signal(signal.SIGINT, handle_abort)
signal.signal(signal.SIGABRT, handle_abort)
def run_tests(test_scripts):
"""
Returns True if all the tests pass or skip
"""
any_failures = False
results = []
tap = lttngtest.TapGenerator(len(test_scripts))
for test_script in test_scripts:
stdout, retcode = run_test(test_script)
result = {
"test": test_script,
"stdout": stdout,
"returncode": retcode,
"passed": len(TAP_OK_LINE_RE.findall(stdout)),
"failed": len(TAP_NOT_OK_LINE_RE.findall(stdout)),
"skipped": len(TAP_SKIP_LINE_RE.findall(stdout)),
}
plan_match = TAP_PLAN_LINE_RE.search(stdout)
if plan_match:
result["planned"] = int(plan_match.group(1))
else:
result["planned"] = "unknown"
results.append(result)
results[-1]["total_test_count"] = (
results[-1]["passed"] + results[-1]["failed"] + results[-1]["skipped"]
)
if retcode == 77:
# Special code for 'not platform applicable'
# See https://www.gnu.org/software/automake/manual/html_node/Scripts_002dbased-Testsuites.html
tap.skip("Test script '{}' returned code 77".format(test_script))
else:
any_failures = any_failures or (retcode != 0 or results[-1]["failed"] != 0)
tap.test(
retcode == 0 and results[-1]["failed"] == 0,
"Test script '{}' returned code {}. {} passed, {} failed, {} skipped, {} planned [total: {}]".format(
test_script,
retcode,
results[-1]["passed"],
results[-1]["failed"],
results[-1]["skipped"],
results[-1]["planned"],
results[-1]["total_test_count"],
),
)
if test_being_aborted:
logging.warning("Aborting test suite")
break
return not any_failures
def run_test(test_script):
global current_test_process
stdout = ""
logging.info("Starting test '{}'".format(test_script))
# Support logd from the modified tap driver
current_test_log_dir = os.environ.get("LTTNG_TEST_LOG_DIR", None)
test_env = os.environ.copy()
test_log = pathlib.Path("{}.log".format(test_script))
test_log_dir = None
if current_test_log_dir:
test_log_dir = pathlib.Path("{}.log.d".format(test_script))
try:
test_log_dir.mkdir(parents=True)
except FileExistsError:
pass
test_env["LTTNG_TEST_LOG_DIR"] = str(test_log_dir)
try:
current_test_process = subprocess.Popen(
[test_script],
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
env=test_env,
)
except Exception as e:
logging.error(
"Exception while starting test '{}': {}".format(test_script, str(e))
)
return (str(e), -1)
return_code = 0
communication_timeout = int(
os.getenv("LTTNG_TEST_SERIAL_TEST_POLL_PERIOD_SECONDS", 60)
)
# abort the test if there is more than 30 minutes without new output
test_timeout_minutes = int(os.getenv("LTTNG_TEST_SERIAL_TEST_TIMEOUT_MINUTES", 30))
last_output = time.time()
with open(str(test_log), "w") as f:
while current_test_process.returncode is None:
try:
out, err = current_test_process.communicate(
timeout=communication_timeout
)
data = out.decode("utf-8")
stdout += data
if f.write(data) > 0:
last_output = time.time()
except subprocess.TimeoutExpired:
logging.info("Test '{}' still running...".format(test_script))
if (time.time() - last_output) >= (test_timeout_minutes * 60):
logging.info(
"{}s since last output, aborting test '{}'".format(
time.time() - last_output, test_script
)
)
current_test_process.kill()
continue
return_code = current_test_process.returncode
current_test_process = None
logging.info(
"Test '{}' terminated with return code {}".format(test_script, return_code)
)
# Some test scripts can fail but still return a zero exit code
if return_code == 0 and len(TAP_NOT_OK_LINE_RE.findall(stdout)) == 0:
if test_log_dir:
shutil.rmtree(str(test_log_dir))
return (stdout, return_code)
if __name__ == "__main__":
logging.basicConfig(
level=logging.DEBUG, format="# [%(created)f] %(levelname)s:%(name)s:%(message)s"
)
tests = shlex.split(os.environ.get("SERIAL_TESTS", ""))
skip_tests = shlex.split(os.environ.get("SKIP_SERIAL_TESTS", ""))
if skip_tests:
logging.debug("Skipped serial tests: {}".format(skip_tests))
tests = [test for test in tests if test not in skip_tests]
limit_tests = shlex.split(os.environ.get("TESTS", ""))
if limit_tests:
logging.debug("Limiting tests to: {}".format(limit_tests))
tests = [test for test in tests if test in limit_tests]
logging.info("Serial tests received %d tests", len(tests))
logging.debug("Serial tests: {}".format(tests))
if len(tests) == 0:
tap = lttngtest.TapGenerator(1)
tap.skip("No tests specified by the environment")
sys.exit(0)
if not run_tests(tests):
sys.exit(1)
|
