gh-158895: Teach libregrtest to output github job summaries

Lib/test/libregrtest/main.py

@@ -1,3 +1,4 @@
+import html
 import os
 import random
 import re
@@ -6,7 +7,7 @@
 import sysconfig
 import time
 import trace
-from _colorize import get_colors  # type: ignore[import-not-found]
+from _colorize import decolor, get_colors  # type: ignore[import-not-found]
 from typing import NoReturn
 
 from test.support import os_helper, MS_WINDOWS, flush_std_streams
@@ -284,7 +285,8 @@ def _rerun_failed_tests(self, runtests: RunTests) -> RunTests:
             forever=False,
             fail_fast=False,
             match_tests_dict=match_tests_dict,
-            output_on_failure=False)
+            output_on_failure=False,
+            github_annotations=bool(os.environ.get("GITHUB_STEP_SUMMARY")))
         self.logger.set_tests(runtests)
 
         msg = f"Re-running {len(tests)} failed tests in verbose mode"
@@ -395,6 +397,8 @@ def run_test(
             result = run_single_test(test_name, runtests)
 
         self.results.accumulate_result(result, runtests)
+        if runtests.github_annotations:
+            result.print_github_annotation(runtests)
 
         return result
 
@@ -480,6 +484,42 @@ 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:
+        filename = os.environ.get("GITHUB_STEP_SUMMARY")
+        # 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 filename or not failed or not exitcode:
+            return
+        cases = [(result.errors or []) + (result.failures or [])
+                 for result in failed]
+        ncases = sum(map(len, cases))
+        with open(filename, "a", encoding="utf-8") as fp:
+            def write(text: str) -> None:
+                # Separate Markdown blocks with an empty line
+                print(text, end="\n\n", file=fp)
+
+            write(f"## {decolor(self.get_state())}: "
+                  f"{count(len(failed), 'test file')} and "
+                  f"{count(ncases, 'test case')} failed")
+            for result, result_cases in zip(failed, cases):
+                write(f"### {decolor(str(result))}")
+                if result.env_changed_reasons:
+                    write("\n".join(f"- {reason}"
+                                    for reason in result.env_changed_reasons))
+                for name, traceback in result_cases:
+                    # Expand the tracebacks if they fit on a screen: up to 5
+                    # test cases with a traceback of less than 30 lines
+                    is_open = ncases <= 5 and traceback.count("\n") < 30
+                    write(f"<details{' open' if is_open else ''}>"
+                          f"<summary>{html.escape(name)}</summary>")
+                    write(f"```pytb\n{decolor(traceback).rstrip()}\n```")
+                    write("</details>")
+
     def display_summary(self) -> None:
         if self.first_runtests is None:
             raise ValueError(
@@ -500,6 +540,12 @@ def display_summary(self) -> None:
         print(f"Result: {state}")
 
     def create_run_tests(self, tests: TestTuple) -> RunTests:
+        # Annotate test failures in the GitHub Actions job log of the last
+        # run (the re-run, if any), if it reports failures
+        will_rerun = self.want_rerun and not self.python_cmd
+        github_annotations = (bool(os.environ.get("GITHUB_STEP_SUMMARY"))
+                              and not will_rerun
+                              and bool(self.verbose or self.output_on_failure))
         return RunTests(
             tests,
             fail_fast=self.fail_fast,
@@ -517,6 +563,7 @@ 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),
+            github_annotations=github_annotations,
             coverage=self.coverage,
             memory_limit=self.memory_limit,
             gc_threshold=self.gc_threshold,

Lib/test/libregrtest/result.py

@@ -1,11 +1,13 @@
 import dataclasses
+import importlib.util
 import json
 from _colorize import get_colors  # type: ignore[import-not-found]
 from typing import Any
 
+from .runtests import RunTests
 from .utils import (
-    StrJSON, TestName, FilterTuple,
-    format_duration, normalize_test_name, print_warning)
+    StrJSON, TestName, FilterTuple, abs_module_name,
+    format_duration, normalize_test_name, print_warning, github_annotation)
 
 
 @dataclasses.dataclass(slots=True)
@@ -178,6 +180,22 @@ def __str__(self) -> str:
     def has_meaningful_duration(self):
         return State.has_meaningful_duration(self.state)
 
+    def print_github_annotation(self, runtests: RunTests) -> None:
+        """Annotate a failed test without test case failures.
+
+        For example: crash, timeout, env changed. Locate the annotation in the
+        test file. Test case failures are annotated by the test runner, where
+        they are reported (see RegressionTestResult.printErrorList()).
+        """
+        if (self.is_failed(runtests.fail_env_changed)
+                and not self.errors and not self.failures):
+            message = "\n".join([str(self), *(self.env_changed_reasons or ())])
+            spec = importlib.util.find_spec(
+                abs_module_name(self.test_name, runtests.test_dir))
+            filename = spec.origin if spec is not None else None
+            print(github_annotation(self.test_name, message, filename),
+                  flush=True)
+
     def set_env_changed(self, *reasons):
         if self.state is None or self.state == State.PASSED:
             self.state = State.ENV_CHANGED

Lib/test/libregrtest/run_workers.py

@@ -613,6 +613,8 @@ def _process_result(self, item: QueueOutput) -> TestResult:
         result = mp_result.result
         self.results.accumulate_result(result, self.runtests)
         self.display_result(mp_result)
+        if self.runtests.github_annotations:
+            result.print_github_annotation(self.runtests)
 
         # Display worker stdout
         if not self.runtests.output_on_failure:

Lib/test/libregrtest/runtests.py

@@ -93,6 +93,8 @@ class RunTests:
     hunt_refleak: HuntRefleak | None
     test_dir: StrPath | None
     use_junit: bool
+    # Annotate test failures in the GitHub Actions job log
+    github_annotations: bool
     coverage: bool
     memory_limit: str | None
     gc_threshold: int | None

Lib/test/libregrtest/setup.py

@@ -112,12 +112,13 @@ def setup_tests(runtests: RunTests) -> None:
 
     set_match_tests(runtests.match_tests)
 
+    from .testresult import RegressionTestResult
     if runtests.use_junit:
         support.junit_xml_list = []
-        from .testresult import RegressionTestResult
         RegressionTestResult.USE_XML = True
     else:
         support.junit_xml_list = None
+    RegressionTestResult.GITHUB_ANNOTATIONS = runtests.github_annotations
 
     if runtests.memory_limit is not None:
         support.set_memlimit(runtests.memory_limit)

Lib/test/libregrtest/testresult.py

@@ -4,15 +4,17 @@
 
 import functools
 import io
+import re
 import sys
 import time
 import traceback
 import unittest
 from test import support
-from test.libregrtest.utils import sanitize_xml
+from test.libregrtest.utils import github_annotation, sanitize_xml
 
 class RegressionTestResult(unittest.TextTestResult):
     USE_XML = False
+    GITHUB_ANNOTATIONS = False
 
     def __init__(self, stream, descriptions, verbosity):
         super().__init__(stream=stream, descriptions=descriptions,
@@ -130,6 +132,26 @@ def addUnexpectedSuccess(self, test):
         self._add_result(test, outcome='UNEXPECTED_SUCCESS')
         super().addUnexpectedSuccess(test)
 
+    def printErrorList(self, flavour, errors):
+        for test, err in errors:
+            if self.GITHUB_ANNOTATIONS:
+                # Write the annotation just before the failure report, so
+                # that it links to the report in the job log
+                self.stream.writeln(github_annotation(str(test), err,
+                                                      self._test_file(test)))
+            super().printErrorList(flavour, [(test, err)])
+
+    @staticmethod
+    def _test_file(test):
+        # Test id: "module.Class.method", "module.function" (doctest), or
+        # "setUpClass (module.Class)" (error in a class or module fixture)
+        name = getattr(test, 'test_case', test).id()  # subTest()
+        if match := re.fullmatch(r'\w+ \((.+)\)', name):
+            name = match[1]
+        while name and name not in sys.modules:
+            name = name.rpartition('.')[0]
+        return getattr(sys.modules.get(name), '__file__', None)
+
     def get_xml_element(self):
         if not self.USE_XML:
             raise ValueError("USE_XML is false")

Lib/test/libregrtest/utils.py

@@ -13,6 +13,7 @@
 import tempfile
 import textwrap
 import types
+from _colorize import decolor  # type: ignore[import-not-found]
 from collections.abc import Callable
 _winapi: types.ModuleType | None
 try:
@@ -140,6 +141,54 @@ def print_warning(msg: str) -> None:
 orig_unraisablehook: Callable[..., None] | None = None
 
 
+def repository_path(filename: str) -> str:
+    """Path of a stdlib file relative to the repository.
+
+    Map the stdlib directory (Lib/ in a source checkout or an installed
+    lib/python3.X/) to Lib/. Return other paths unchanged.
+    """
+    stdlib_dir = os.path.dirname(os.__file__) + os.sep
+    if not filename.startswith(stdlib_dir):
+        return filename
+    return "Lib/" + filename.removeprefix(stdlib_dir).replace(os.sep, "/")
+
+
+def github_annotation(title: str, message: str,
+                      filename: str | None) -> str:
+    """Format a GitHub Actions error annotation.
+
+    message is a traceback or a failure description. Only keep the exception
+    which ends the traceback: the job log has the full traceback. Locate the
+    annotation in filename, the test file, at the last frame of the traceback
+    in this file, if any.
+    """
+    if not filename:
+        raise ValueError(f"missing test file of annotation {title!r}")
+
+    def escape(text: str) -> str:
+        return (text.replace("%", "%25").replace("\r", "%0D")
+                .replace("\n", "%0A"))
+
+    def escape_property(text: str) -> str:
+        return escape(text).replace(":", "%3A").replace(",", "%2C")
+
+    message = decolor(message)
+    props = {"file": repository_path(filename)}
+    # Line of the last traceback frame ('  File "filename", line 123') or
+    # failed doctest example (same, but not indented) in filename
+    if lines := re.findall(rf'^ *File "{re.escape(filename)}", line (\d+)',
+                           message, re.MULTILINE):
+        props["line"] = lines[-1]
+    props["title"] = title
+    # 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)*', message,
+                       flags=re.MULTILINE)[-1].strip()
+    props_text = ",".join(f"{key}={escape_property(value)}"
+                          for key, value in props.items())
+    return f"::error {props_text}::{escape(message)}"
+
+
 def regrtest_unraisable_hook(unraisable) -> None:
     global orig_unraisablehook
     support.set_environment_altered(

Lib/test/libregrtest/worker.py

@@ -28,6 +28,8 @@ def create_worker_process(runtests: WorkerRunTests, output_fd: int,
     cmd.extend(['-m', 'test.libregrtest.worker', worker_json])
 
     env = dict(os.environ)
+    # Only the main process writes the GitHub Actions job summary
+    env.pop('GITHUB_STEP_SUMMARY', None)
     if tmp_dir is not None:
         env['TMPDIR'] = tmp_dir
         env['TEMP'] = tmp_dir

Lib/test/test_bool.py

@@ -4,6 +4,7 @@
 from test.support import os_helper
 
 import os
+import sys
 
 class BoolTest(unittest.TestCase):
 
@@ -25,6 +26,11 @@
         self.assertIs(eval(repr(True)), True)
 
     def test_str(self):
+        if sys.platform == "linux":
+            # XXX: Intentional failure to test GitHub Actions annotations
+            self.assertEqual(
+                str(True),
+                'Yes')
         self.assertEqual(str(False), 'False')
         self.assertEqual(str(True), 'True')
 

Lib/test/test_json/test_decode.py

@@ -1,4 +1,5 @@
 import decimal
+import sys
 import unittest.mock
 from io import StringIO
 from collections import OrderedDict
@@ -9,6 +10,9 @@
 class TestDecode:
     def test_decimal(self):
         rval = self.loads('1.1', parse_float=decimal.Decimal)
+        if sys.platform == "linux":
+            # XXX: Intentional failure to test GitHub Actions annotations
+            self.loads('{"bad": }')
         self.assertIsInstance(rval, decimal.Decimal)
         self.assertEqual(rval, decimal.Decimal('1.1'))
 

Lib/test/test_math.py

@@ -513,6 +513,11 @@
 
     def testFloor(self):
         self.assertRaises(TypeError, math.floor)
+        if sys.platform == "linux":
+            # XXX: Intentional failure to test GitHub Actions annotations
+            for x in (0.5, 1.5):
+                with self.subTest(x=x):
+                    self.assertEqual(math.floor(x), x)
         self.assertEqual(int, type(math.floor(0.5)))
         self.assertEqual(math.floor(0.5), 0)
         self.assertEqual(math.floor(1.0), 1)

Lib/test/test_os/test_os.py

@@ -4080,7 +4080,8 @@
         self.assertFalse(os.get_inheritable(fd))
         with open(fd, "wb", closefd=False) as f:
             f.write(b'memfd_create')
-            self.assertEqual(f.tell(), 12)
+            # XXX: Intentional failure to test GitHub Actions job summaries
+            self.assertEqual(f.tell(), 13)
 
         fd2 = os.memfd_create("Hi")
         self.addCleanup(os.close, fd2)