Refactor to use a reporter of which github is one

Author: cmaloney

Lib/test/libregrtest/main.py

@@ -6,7 +6,7 @@
 import sysconfig
 import time
 import trace
-from _colorize import decolor, get_colors  # type: ignore[import-not-found]
+from _colorize import get_colors  # type: ignore[import-not-found]
 from typing import NoReturn
 
 from test.support import os_helper, MS_WINDOWS, flush_std_streams
@@ -15,6 +15,7 @@
 from .findtests import findtests, split_test_packages, list_cases, collect_cases
 from .logger import Logger
 from .pgo import setup_pgo_tests
+from .reporter import get_reporter
 from .result import TestResult
 from .results import TestResults, EXITCODE_INTERRUPTED
 from .runtests import RunTests, HuntRefleak
@@ -285,7 +286,7 @@ def _rerun_failed_tests(self, runtests: RunTests) -> RunTests:
             fail_fast=False,
             match_tests_dict=match_tests_dict,
             output_on_failure=False,
-            github_annotations=bool(self.github_summary))
+            github_summary=self.github_summary)
         self.logger.set_tests(runtests)
 
         msg = f"Re-running {len(tests)} failed tests in verbose mode"
@@ -298,7 +299,7 @@ def _rerun_failed_tests(self, runtests: RunTests) -> RunTests:
             self.run_tests_sequentially(runtests)
         return runtests
 
-    def rerun_failed_tests(self, runtests: RunTests) -> None:
+    def rerun_failed_tests(self, runtests: RunTests) -> RunTests:
         ansi = get_colors()
         red, reset = ansi.BOLD_RED, ansi.RESET
 
@@ -308,7 +309,7 @@ def rerun_failed_tests(self, runtests: RunTests) -> None:
                 "Re-running failed tests is not supported with --python "
                 "host runner option."
             )
-            return
+            return runtests
 
         self.first_state = self.get_state()
 
@@ -323,6 +324,7 @@ def rerun_failed_tests(self, runtests: RunTests) -> None:
             printlist(self.results.bad)
 
         self.display_result(rerun_runtests)
+        return rerun_runtests
 
     def _run_bisect(self, runtests: RunTests, test: str, progress: str) -> bool:
         print()
@@ -396,7 +398,6 @@ def run_test(
             result = run_single_test(test_name, runtests)
 
         self.results.accumulate_result(result, runtests)
-        result.print_github_annotation(runtests)
 
         return result
 
@@ -463,7 +464,8 @@ def _run_tests_mp(self, runtests: RunTests, num_workers: int) -> None:
         from .run_workers import RunWorkers
         RunWorkers(num_workers, runtests, self.logger, self.results).run()
 
-    def finalize_tests(self, coverage: trace.CoverageResults | None) -> None:
+    def finalize_tests(self, coverage: trace.CoverageResults | None,
+                       runtests: RunTests) -> None:
         if self.next_single_filename:
             if self.next_single_test:
                 with open(self.next_single_filename, 'w') as fp:
@@ -482,24 +484,11 @@ def finalize_tests(self, coverage: trace.CoverageResults | None) -> None:
         if self.junit_filename:
             self.results.write_junit(self.junit_filename)
 
-        self.write_github_summary()
-
-    def write_github_summary(self) -> None:
-        # Tests which failed in the last run (the re-run, if any)
-        failed = self.results.rerun_results
-        # Only write a summary on failure: a CI run has many test jobs
-        exitcode = self.results.get_exitcode(self.fail_env_changed,
-                                             self.fail_rerun)
-        if not self.github_summary or not failed or not exitcode:
-            return
-        ncases = sum(len(result.failed_test_cases) for result in failed)
-        with open(self.github_summary, "a", encoding="utf-8") as fp:
-            fp.write(f"## {decolor(self.get_state())}: "
-                     f"{count(len(failed), 'test file')} and "
-                     f"{count(ncases, 'test case')} failed\n\n")
-            for result in failed:
-                # Expand the tracebacks of up to 5 test cases
-                fp.write(result.format_github_summary(expand=ncases <= 5))
+        # Test files which failed in the last run (the re-run, if any)
+        failed = [result for result in self.results.rerun_results
+                  if result.is_failed(self.fail_env_changed)]
+        if failed:
+            get_reporter(runtests).run_failed(failed, self.get_state())
 
     def display_summary(self) -> None:
         if self.first_runtests is None:
@@ -538,10 +527,10 @@ def create_run_tests(self, tests: TestTuple) -> RunTests:
             hunt_refleak=self.hunt_refleak,
             test_dir=self.test_dir,
             use_junit=(self.junit_filename is not None),
-            # Only annotate the failure that results in job failure.
-            # Ex: When reruning failed tests, only annotate the the final run.
-            github_annotations=(bool(self.github_summary) and
-                                not (self.want_rerun and not self.python_cmd)),
+            # Only report the failures which fail the job: when failed tests
+            # are re-run, only report the re-run.
+            github_summary=(None if self.want_rerun and not self.python_cmd
+                            else self.github_summary),
             coverage=self.coverage,
             memory_limit=self.memory_limit,
             gc_threshold=self.gc_threshold,
@@ -612,8 +601,10 @@ def _run_tests(self, selected: TestTuple, tests: TestList | None) -> int:
             coverage = self.results.get_coverage_results()
             self.display_result(runtests)
 
+            # The re-run, if any, is the last run
+            last_runtests = runtests
             if self.want_rerun and self.results.need_rerun():
-                self.rerun_failed_tests(runtests)
+                last_runtests = self.rerun_failed_tests(runtests)
 
             if self.want_bisect and self.results.need_rerun():
                 self.run_bisect(runtests)
@@ -622,7 +613,7 @@ def _run_tests(self, selected: TestTuple, tests: TestList | None) -> int:
                 self.logger.stop_load_tracker()
 
         self.display_summary()
-        self.finalize_tests(coverage)
+        self.finalize_tests(coverage, last_runtests)
 
         return self.results.get_exitcode(self.fail_env_changed,
                                          self.fail_rerun)

Lib/test/libregrtest/reporter.py

@@ -0,0 +1,130 @@
+import html
+import os
+import pathlib
+import re
+from _colorize import decolor  # type: ignore[import-not-found]
+from typing import TextIO
+
+from .findtests import findtestdir
+from .result import TestResult
+from .runtests import RunTests
+from .utils import count
+
+
+class Reporter:
+    """Report test failures, for example to a CI service.
+
+    The default reporter does nothing.
+    """
+
+    def test_case_failed(self, test_case: str, traceback: str,
+                         stream: TextIO) -> None:
+        """Report a failed test case.
+
+        Called just after unittest printed the failure, or at the end of the
+        test file in quiet mode (failures are not printed).
+        """
+
+    def test_result(self, result: TestResult, runtests: RunTests) -> None:
+        """Report the result of a test file, in the main process."""
+
+    def run_failed(self, failed: list[TestResult], state: str) -> None:
+        """Report a failed run: failed is the list of failed test files."""
+
+
+def get_reporter(runtests: RunTests) -> Reporter:
+    if runtests.github_summary:
+        return GitHubActionsReporter(runtests.github_summary)
+    return Reporter()
+
+
+class GitHubActionsReporter(Reporter):
+    """Report test failures to GitHub Actions.
+
+    Annotate failures in the job log and list them in the job summary.
+    """
+
+    # Escape the data and the properties of workflow commands
+    ESCAPE_DATA = str.maketrans({"%": "%25", "\r": "%0D", "\n": "%0A"})
+    ESCAPE_PROPERTY = ESCAPE_DATA | str.maketrans({":": "%3A", ",": "%2C"})
+
+    def __init__(self, summary_filename: str) -> None:
+        self.summary_filename = summary_filename
+
+    def test_case_failed(self, test_case: str, traceback: str,
+                         stream: TextIO) -> None:
+        traceback = decolor(traceback)
+        # Get a file + line number from the backtrace: the last frame in a
+        # test file, or else the first frame (the test). Without frames, the
+        # failure is still annotated, just not located.
+        filename = line = None
+        if frames := re.findall(r'^ *File "(.*)", line (\d+)', traceback,
+                                re.MULTILINE):
+            test_frames = [frame for frame in frames
+                           if pathlib.Path(frame[0]).name.startswith('test_')]
+            filename, line = test_frames[-1] if test_frames else frames[0]
+        # Only keep the exception: strip the traceback frames, each one a
+        # '  File' line followed by its indented source lines
+        message = re.split(r'^  File .*\n(?:    .*\n)*', traceback,
+                           flags=re.MULTILINE)[-1].strip()
+        self.annotate(test_case, message, filename, line, stream)
+
+    def test_result(self, result: TestResult, runtests: RunTests) -> None:
+        # Annotate a test file which failed without failed test cases, for
+        # example: crash, timeout, env changed. Failed test cases are
+        # annotated by test_case_failed().
+        if (not result.is_failed(runtests.fail_env_changed)
+                or result.failed_test_cases):
+            return
+        filename = result.filename
+        if not filename:
+            # The worker crashed or timed out before reporting the test file:
+            # convert python import test name to relative file path.
+            # Ex: "test.test_x.test_y" is test_x/test_y.py
+            name = result.test_name.removeprefix("test.")
+            path = pathlib.Path(findtestdir(runtests.test_dir),
+                                *name.split("."))
+            if path.with_suffix(".py").exists():
+                filename = str(path.with_suffix(".py"))
+            elif (path / "__init__.py").exists():
+                filename = str(path / "__init__.py")
+            # else not a test file (ex: "<regrtest worker>" bug): no location
+        message = "\n".join([str(result), *(result.env_changed_reasons or ())])
+        self.annotate(result.test_name, decolor(message), filename)
+
+    def annotate(self, title: str, message: str, filename: str | None,
+                 line: str | None = None, stream: TextIO | None = None) -> None:
+        if filename:
+            # Normalize stdlib path to be relative inside the repository
+            # checkout.
+            path = pathlib.Path(filename)
+            stdlib_dir = pathlib.Path(os.__file__).parent
+            if path.is_relative_to(stdlib_dir):
+                filename = ("Lib" / path.relative_to(stdlib_dir)).as_posix()
+        props = {"file": filename, "line": line, "title": title}
+        props_text = ",".join(f"{key}={value.translate(self.ESCAPE_PROPERTY)}"
+                              for key, value in props.items() if value)
+        print(f"::error {props_text}::{message.translate(self.ESCAPE_DATA)}",
+              file=stream, flush=True)
+
+    def run_failed(self, failed: list[TestResult], state: str) -> None:
+        ncases = sum(len(result.failed_test_cases) for result in failed)
+        blocks = [f"## {decolor(state)}: "
+                  f"{count(len(failed), 'test file')} and "
+                  f"{count(ncases, 'test case')} failed"]
+        for result in failed:
+            blocks.append(f"### {decolor(str(result))}")
+            if result.env_changed_reasons:
+                blocks.append("\n".join(f"- {reason}"
+                                        for reason in result.env_changed_reasons))
+            for name, traceback in result.failed_test_cases:
+                # Only default open short tracebacks of up to 5 test cases.
+                is_open = ncases <= 5 and traceback.count("\n") < 30
+                blocks += [f"<details{' open' if is_open else ''}>"
+                           f"<summary>{html.escape(name)}</summary>",
+                           f"```pytb\n{decolor(traceback).rstrip()}\n```",
+                           "</details>"]
+        # Separate Markdown blocks with an empty line
+        with open(self.summary_filename, "a", encoding="utf-8") as fp:
+            fp.write("".join(f"{block}\n\n" for block in blocks))
+

Lib/test/libregrtest/result.py

@@ -1,16 +1,11 @@
 import dataclasses
-import html
 import json
-import pathlib
-from _colorize import decolor, get_colors  # type: ignore[import-not-found]
+from _colorize import get_colors  # type: ignore[import-not-found]
 from typing import Any
 
-from .findtests import findtestdir
-from .runtests import RunTests
 from .utils import (
     StrJSON, TestName, FilterTuple,
-    format_duration, normalize_test_name, print_warning,
-    print_github_annotation)
+    format_duration, normalize_test_name, print_warning)
 
 
 @dataclasses.dataclass(slots=True)
@@ -190,44 +185,6 @@ def has_meaningful_duration(self):
     def failed_test_cases(self) -> list[tuple[str, str]]:
         return (self.errors or []) + (self.failures or [])
 
-    def format_github_summary(self, expand: bool) -> str:
-        """Format the Markdown section of the GitHub Actions job summary."""
-        blocks = [f"### {decolor(str(self))}"]
-        if self.env_changed_reasons:
-            blocks.append("\n".join(f"- {reason}"
-                                    for reason in self.env_changed_reasons))
-        for name, traceback in self.failed_test_cases:
-            # Only default open short tracebacks.
-            is_open = expand and traceback.count("\n") < 30
-            blocks += [f"<details{' open' if is_open else ''}>"
-                       f"<summary>{html.escape(name)}</summary>",
-                       f"```pytb\n{decolor(traceback).rstrip()}\n```",
-                       "</details>"]
-
-        # Separate Markdown blocks with an empty line
-        return "".join(f"{block}\n\n" for block in blocks)
-
-    def print_github_annotation(self, runtests: RunTests) -> None:
-        """Annotate a failed test file without failed test cases."""
-        if (not runtests.github_annotations
-                or not self.is_failed(runtests.fail_env_changed)
-                or self.failed_test_cases):
-            return
-        filename = self.filename
-        if not filename:
-            # The worker crashed or timed out before reporting the test file:
-            # convert python import test name to relative file path.
-            # Ex: "test.test_x.test_y" is test_x/test_y.py
-            name = self.test_name.removeprefix("test.")
-            path = pathlib.Path(findtestdir(runtests.test_dir),
-                                *name.split("."))
-            if path.with_suffix(".py").exists():
-                filename = str(path.with_suffix(".py"))
-            else:
-                filename = str(path / "__init__.py")
-        message = "\n".join([str(self), *(self.env_changed_reasons or ())])
-        print_github_annotation(self.test_name, message, filename)
-
     def set_env_changed(self, *reasons):
         if self.state is None or self.state == State.PASSED:
             self.state = State.ENV_CHANGED

Lib/test/libregrtest/results.py

@@ -3,6 +3,7 @@
 from _colorize import get_colors  # type: ignore[import-not-found]
 lazy from xml.etree.ElementTree import Element
 
+from .reporter import get_reporter
 from .runtests import RunTests
 from .result import State, TestResult, TestStats, Location
 from .utils import (
@@ -142,6 +143,7 @@ def accumulate_result(self, result: TestResult, runtests: RunTests) -> None:
         xml_data = result.xml_data
         if xml_data:
             self.add_junit(xml_data)
+        get_reporter(runtests).test_result(result, runtests)
 
     def get_coverage_results(self) -> trace.CoverageResults:
         counts = {loc: 1 for loc in self.covered_lines}

Lib/test/libregrtest/run_workers.py

@@ -624,7 +624,6 @@ def _process_result(self, item: QueueOutput) -> TestResult:
             stdout = mp_result.worker_stdout
             if stdout:
                 print(stdout, flush=True)
-        result.print_github_annotation(self.runtests)
 
         return result
 

Lib/test/libregrtest/runtests.py

@@ -93,7 +93,8 @@ class RunTests:
     hunt_refleak: HuntRefleak | None
     test_dir: StrPath | None
     use_junit: bool
-    github_annotations: bool
+    # GitHub Actions job summary file, to report the failures of this run
+    github_summary: str | None
     coverage: bool
     memory_limit: str | None
     gc_threshold: int | None

Lib/test/libregrtest/setup.py

@@ -11,6 +11,7 @@
 from _colorize import can_colorize  # type: ignore[import-not-found]
 
 from .filter import set_match_tests
+from .reporter import get_reporter
 from .runtests import RunTests
 lazy from .testresult import RegressionTestResult
 from .utils import (
@@ -113,7 +114,7 @@ def setup_tests(runtests: RunTests) -> None:
 
     set_match_tests(runtests.match_tests)
 
-    RegressionTestResult.GITHUB_ANNOTATIONS = runtests.github_annotations
+    RegressionTestResult.reporter = get_reporter(runtests)
     if runtests.use_junit:
         support.junit_xml_list = []
         RegressionTestResult.USE_XML = True