ENV_CHANGED = "ENV_CHANGED"
RESOURCE_DENIED = "RESOURCE_DENIED"
INTERRUPTED = "INTERRUPTED"
- MULTIPROCESSING_ERROR = "MULTIPROCESSING_ERROR"
+ WORKER_FAILED = "WORKER_FAILED" # non-zero worker process exit code
+ WORKER_BUG = "WORKER_BUG" # exception when running a worker
DID_NOT_RUN = "DID_NOT_RUN"
TIMEOUT = "TIMEOUT"
State.FAILED,
State.UNCAUGHT_EXC,
State.REFLEAK,
- State.MULTIPROCESSING_ERROR,
+ State.WORKER_FAILED,
+ State.WORKER_BUG,
State.TIMEOUT}
@staticmethod
State.SKIPPED,
State.RESOURCE_DENIED,
State.INTERRUPTED,
- State.MULTIPROCESSING_ERROR,
+ State.WORKER_FAILED,
+ State.WORKER_BUG,
State.DID_NOT_RUN}
@staticmethod
def must_stop(state):
return state in {
State.INTERRUPTED,
- State.MULTIPROCESSING_ERROR}
+ State.WORKER_BUG,
+ }
@dataclasses.dataclass(slots=True)
return f"{self.test_name} skipped (resource denied)"
case State.INTERRUPTED:
return f"{self.test_name} interrupted"
- case State.MULTIPROCESSING_ERROR:
- return f"{self.test_name} process crashed"
+ case State.WORKER_FAILED:
+ return f"{self.test_name} worker non-zero exit code"
+ case State.WORKER_BUG:
+ return f"{self.test_name} worker bug"
case State.DID_NOT_RUN:
return f"{self.test_name} ran no tests"
case State.TIMEOUT:
self.rerun_results: list[TestResult] = []
self.interrupted: bool = False
+ self.worker_bug: bool = False
self.test_times: list[tuple[float, TestName]] = []
self.stats = TestStats()
# used by --junit-xml
def is_all_good(self):
return (not self.bad
and not self.skipped
- and not self.interrupted)
+ and not self.interrupted
+ and not self.worker_bug)
def get_executed(self):
return (set(self.good) | set(self.bad) | set(self.skipped)
if self.interrupted:
state.append("INTERRUPTED")
+ if self.worker_bug:
+ state.append("WORKER BUG")
if not state:
state.append("SUCCESS")
exitcode = EXITCODE_NO_TESTS_RAN
elif fail_rerun and self.rerun:
exitcode = EXITCODE_RERUN_FAIL
+ elif self.worker_bug:
+ exitcode = EXITCODE_BAD_TEST
return exitcode
def accumulate_result(self, result: TestResult, runtests: RunTests):
else:
raise ValueError(f"invalid test state: {result.state!r}")
+ if result.state == State.WORKER_BUG:
+ self.worker_bug = True
+
if result.has_meaningful_duration() and not rerun:
self.test_times.append((result.duration, test_name))
if result.stats is not None:
f.write(s)
def display_result(self, tests: TestTuple, quiet: bool, print_slowest: bool):
- omitted = set(tests) - self.get_executed()
- if omitted:
- print()
- print(count(len(omitted), "test"), "omitted:")
- printlist(omitted)
-
if print_slowest:
self.test_times.sort(reverse=True)
print()
for test_time, test in self.test_times[:10]:
print("- %s: %s" % (test, format_duration(test_time)))
- all_tests = [
- (self.bad, "test", "{} failed:"),
- (self.env_changed, "test", "{} altered the execution environment (env changed):"),
- ]
+ all_tests = []
+ omitted = set(tests) - self.get_executed()
+
+ # less important
+ all_tests.append((omitted, "test", "{} omitted:"))
if not quiet:
all_tests.append((self.skipped, "test", "{} skipped:"))
all_tests.append((self.resource_denied, "test", "{} skipped (resource denied):"))
- all_tests.append((self.rerun, "re-run test", "{}:"))
all_tests.append((self.run_no_tests, "test", "{} run no tests:"))
+ # more important
+ all_tests.append((self.env_changed, "test", "{} altered the execution environment (env changed):"))
+ all_tests.append((self.rerun, "re-run test", "{}:"))
+ all_tests.append((self.bad, "test", "{} failed:"))
+
for tests_list, count_text, title_format in all_tests:
if tests_list:
print()
from .single import PROGRESS_MIN_TIME
from .utils import (
StrPath, TestName, MS_WINDOWS,
- format_duration, print_warning, count, plural)
+ format_duration, print_warning, count, plural, get_signal_name)
from .worker import create_worker_process, USE_PROCESS_GROUP
if MS_WINDOWS:
test_name: TestName,
err_msg: str | None,
stdout: str | None,
- state: str = State.MULTIPROCESSING_ERROR):
+ state: str):
result = TestResult(test_name, state=state)
self.mp_result = MultiprocessResult(result, stdout, err_msg)
super().__init__()
# gh-101634: Catch UnicodeDecodeError if stdout cannot be
# decoded from encoding
raise WorkerError(self.test_name,
- f"Cannot read process stdout: {exc}", None)
+ f"Cannot read process stdout: {exc}",
+ stdout=None,
+ state=State.WORKER_BUG)
def read_json(self, json_file: JsonFile, json_tmpfile: TextIO | None,
stdout: str) -> tuple[TestResult, str]:
# decoded from encoding
err_msg = f"Failed to read worker process JSON: {exc}"
raise WorkerError(self.test_name, err_msg, stdout,
- state=State.MULTIPROCESSING_ERROR)
+ state=State.WORKER_BUG)
if not worker_json:
- raise WorkerError(self.test_name, "empty JSON", stdout)
+ raise WorkerError(self.test_name, "empty JSON", stdout,
+ state=State.WORKER_BUG)
try:
result = TestResult.from_json(worker_json)
# decoded from encoding
err_msg = f"Failed to parse worker process JSON: {exc}"
raise WorkerError(self.test_name, err_msg, stdout,
- state=State.MULTIPROCESSING_ERROR)
+ state=State.WORKER_BUG)
return (result, stdout)
stdout = self.read_stdout(stdout_file)
if retcode is None:
- raise WorkerError(self.test_name, None, stdout, state=State.TIMEOUT)
+ raise WorkerError(self.test_name, stdout=stdout,
+ err_msg=None,
+ state=State.TIMEOUT)
if retcode != 0:
- raise WorkerError(self.test_name, f"Exit code {retcode}", stdout)
+ name = get_signal_name(retcode)
+ if name:
+ retcode = f"{retcode} ({name})"
+ raise WorkerError(self.test_name, f"Exit code {retcode}", stdout,
+ state=State.WORKER_FAILED)
result, stdout = self.read_json(json_file, json_tmpfile, stdout)
text = str(result)
if mp_result.err_msg:
- # MULTIPROCESSING_ERROR
+ # WORKER_BUG
text += ' (%s)' % mp_result.err_msg
elif (result.duration >= PROGRESS_MIN_TIME and not pgo):
text += ' (%s)' % format_duration(result.duration)
# Thread got an exception
format_exc = item[1]
print_warning(f"regrtest worker thread failed: {format_exc}")
- result = TestResult("<regrtest worker>", state=State.MULTIPROCESSING_ERROR)
+ result = TestResult("<regrtest worker>", state=State.WORKER_BUG)
self.results.accumulate_result(result, self.runtests)
return result
import os.path
import platform
import random
+import signal
import sys
import sysconfig
import tempfile
else:
print("Remove file: %s" % name)
os_helper.unlink(name)
+
+WINDOWS_STATUS = {
+ 0xC0000005: "STATUS_ACCESS_VIOLATION",
+ 0xC00000FD: "STATUS_STACK_OVERFLOW",
+ 0xC000013A: "STATUS_CONTROL_C_EXIT",
+}
+
+def get_signal_name(exitcode):
+ if exitcode < 0:
+ signum = -exitcode
+ try:
+ return signal.Signals(signum).name
+ except ValueError:
+ pass
+
+ try:
+ return WINDOWS_STATUS[exitcode]
+ except KeyError:
+ pass
+
+ return None
import random
import re
import shlex
+import signal
import subprocess
import sys
import sysconfig
self.assertIsNone(normalize('setUpModule (test.test_x)', is_error=True))
self.assertIsNone(normalize('tearDownModule (test.test_module)', is_error=True))
+ def test_get_signal_name(self):
+ for exitcode, expected in (
+ (-int(signal.SIGINT), 'SIGINT'),
+ (-int(signal.SIGSEGV), 'SIGSEGV'),
+ (3221225477, "STATUS_ACCESS_VIOLATION"),
+ (0xC00000FD, "STATUS_STACK_OVERFLOW"),
+ ):
+ self.assertEqual(utils.get_signal_name(exitcode), expected, exitcode)
+
if __name__ == '__main__':
unittest.main()
projects / thirdparty / Python / cpython.git / commitdiff
