pythongh-109276: regrtest: add WORKER_FAILED state

WORKER_FAILED does not stop the manager, whereas WORKER_BUG does.

Change also TestResults.display_result() order: display failed tests
at the end, the important important information.

Author: vstinner

Lib/test/libregrtest/result.py

@@ -19,7 +19,8 @@ class State:
     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"
 
@@ -29,7 +30,8 @@ def is_failed(state):
             State.FAILED,
             State.UNCAUGHT_EXC,
             State.REFLEAK,
-            State.MULTIPROCESSING_ERROR,
+            State.WORKER_FAILED,
+            State.WORKER_BUG,
             State.TIMEOUT}
 
     @staticmethod
@@ -42,14 +44,16 @@ def has_meaningful_duration(state):
             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)
@@ -108,8 +112,10 @@ def __str__(self) -> str:
                 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:

Lib/test/libregrtest/results.py

@@ -30,6 +30,7 @@ def __init__(self):
         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
@@ -38,7 +39,8 @@ def __init__(self):
     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)
@@ -60,6 +62,8 @@ def get_state(self, fail_env_changed):
 
         if self.interrupted:
             state.append("INTERRUPTED")
+        if self.worker_bug:
+            state.append("WORKER BUG")
         if not state:
             state.append("SUCCESS")
 
@@ -77,6 +81,8 @@ def get_exitcode(self, fail_env_changed, fail_rerun):
             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):
@@ -105,6 +111,9 @@ 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:
@@ -173,29 +182,28 @@ def write_junit(self, filename: StrPath):
                 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()
             print("10 slowest tests:")
             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()

Lib/test/libregrtest/run_workers.py

@@ -92,7 +92,7 @@ def __init__(self,
                  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__()
@@ -298,7 +298,9 @@ def read_stdout(self, stdout_file: TextIO) -> str:
             # 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]:
@@ -317,10 +319,11 @@ def read_json(self, json_file: JsonFile, json_tmpfile: TextIO | None,
             # 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)
@@ -329,7 +332,7 @@ def read_json(self, json_file: JsonFile, json_tmpfile: TextIO | None,
             # 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)
 
@@ -345,9 +348,11 @@ def _runtest(self, test_name: TestName) -> MultiprocessResult:
             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, None, stdout,
+                                  state=State.TIMEOUT)
             if retcode != 0:
-                raise WorkerError(self.test_name, f"Exit code {retcode}", stdout)
+                raise WorkerError(self.test_name, f"Exit code {retcode}", stdout,
+                                  state=State.WORKER_FAILED)
 
             result, stdout = self.read_json(json_file, json_tmpfile, stdout)
 
@@ -527,7 +532,7 @@ def display_result(self, mp_result: MultiprocessResult) -> None:
 
         text = str(result)
         if mp_result.err_msg:
-            # MULTIPROCESSING_ERROR
+            # WORKER_ERROR
             text += ' (%s)' % mp_result.err_msg
         elif (result.duration >= PROGRESS_MIN_TIME and not pgo):
             text += ' (%s)' % format_duration(result.duration)
@@ -543,7 +548,7 @@ def _process_result(self, item: QueueOutput) -> TestResult:
             # 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_ERROR)
             self.results.accumulate_result(result, self.runtests)
             return result