Allow mypy to output a junit file with per-file results (#16388)

Adds a new `--junit-format` flag to MyPy, which affects the structure of
the junit file written when `--junit-xml` is specified (it has no effect
when not writing a junit file). This flag can take `global` or
`per_file` as values:
* `--junit-format=global` (the default) preserves the existing junit
structure, creating a junit file specifying a single "test" for the
entire mypy run.
* `--junit-format=per_file` will cause the junit file to have one test
entry per file with failures (or a single entry, as in the existing
behavior, in the case when typechecking passes).

In some settings it can be useful to know which files had typechecking
failures (for example, a CI system might want to display failures by
file); while that information can be parsed out of the error messages in
the existing junit files, it's much more convenient to have that
represented in the junit structure.

Tests for the old and new junit structure have been added.

Author: mrwright

mypy/build.py

@@ -145,7 +145,7 @@ def build(
     sources: list[BuildSource],
     options: Options,
     alt_lib_path: str | None = None,
-    flush_errors: Callable[[list[str], bool], None] | None = None,
+    flush_errors: Callable[[str | None, list[str], bool], None] | None = None,
     fscache: FileSystemCache | None = None,
     stdout: TextIO | None = None,
     stderr: TextIO | None = None,
@@ -177,7 +177,9 @@ def build(
     # fields for callers that want the traditional API.
     messages = []
 
-    def default_flush_errors(new_messages: list[str], is_serious: bool) -> None:
+    def default_flush_errors(
+        filename: str | None, new_messages: list[str], is_serious: bool
+    ) -> None:
         messages.extend(new_messages)
 
     flush_errors = flush_errors or default_flush_errors
@@ -197,7 +199,7 @@ def default_flush_errors(new_messages: list[str], is_serious: bool) -> None:
         # Patch it up to contain either none or all none of the messages,
         # depending on whether we are flushing errors.
         serious = not e.use_stdout
-        flush_errors(e.messages, serious)
+        flush_errors(None, e.messages, serious)
         e.messages = messages
         raise
 
@@ -206,7 +208,7 @@ def _build(
     sources: list[BuildSource],
     options: Options,
     alt_lib_path: str | None,
-    flush_errors: Callable[[list[str], bool], None],
+    flush_errors: Callable[[str | None, list[str], bool], None],
     fscache: FileSystemCache | None,
     stdout: TextIO,
     stderr: TextIO,
@@ -600,7 +602,7 @@ def __init__(
         plugin: Plugin,
         plugins_snapshot: dict[str, str],
         errors: Errors,
-        flush_errors: Callable[[list[str], bool], None],
+        flush_errors: Callable[[str | None, list[str], bool], None],
         fscache: FileSystemCache,
         stdout: TextIO,
         stderr: TextIO,
@@ -3458,7 +3460,11 @@ def process_stale_scc(graph: Graph, scc: list[str], manager: BuildManager) -> No
         for id in stale:
             graph[id].transitive_error = True
     for id in stale:
-        manager.flush_errors(manager.errors.file_messages(graph[id].xpath), False)
+        manager.flush_errors(
+            manager.errors.simplify_path(graph[id].xpath),
+            manager.errors.file_messages(graph[id].xpath),
+            False,
+        )
         graph[id].write_cache()
         graph[id].mark_as_rechecked()
 

mypy/config_parser.py

@@ -152,6 +152,17 @@ def check_follow_imports(choice: str) -> str:
     return choice
 
 
+def check_junit_format(choice: str) -> str:
+    choices = ["global", "per_file"]
+    if choice not in choices:
+        raise argparse.ArgumentTypeError(
+            "invalid choice '{}' (choose from {})".format(
+                choice, ", ".join(f"'{x}'" for x in choices)
+            )
+        )
+    return choice
+
+
 def split_commas(value: str) -> list[str]:
     # Uses a bit smarter technique to allow last trailing comma
     # and to remove last `""` item from the split.
@@ -173,6 +184,7 @@ def split_commas(value: str) -> list[str]:
     "files": split_and_match_files,
     "quickstart_file": expand_path,
     "junit_xml": expand_path,
+    "junit_format": check_junit_format,
     "follow_imports": check_follow_imports,
     "no_site_packages": bool,
     "plugins": lambda s: [p.strip() for p in split_commas(s)],
@@ -200,6 +212,7 @@ def split_commas(value: str) -> list[str]:
         "python_version": parse_version,
         "mypy_path": lambda s: [expand_path(p) for p in try_split(s, "[,:]")],
         "files": lambda s: split_and_match_files_list(try_split(s)),
+        "junit_format": lambda s: check_junit_format(str(s)),
         "follow_imports": lambda s: check_follow_imports(str(s)),
         "plugins": try_split,
         "always_true": try_split,

mypy/main.py

@@ -7,6 +7,7 @@
 import subprocess
 import sys
 import time
+from collections import defaultdict
 from gettext import gettext
 from typing import IO, Any, Final, NoReturn, Sequence, TextIO
 
@@ -158,11 +159,14 @@ def run_build(
     formatter = util.FancyFormatter(stdout, stderr, options.hide_error_codes)
 
     messages = []
+    messages_by_file = defaultdict(list)
 
-    def flush_errors(new_messages: list[str], serious: bool) -> None:
+    def flush_errors(filename: str | None, new_messages: list[str], serious: bool) -> None:
         if options.pretty:
             new_messages = formatter.fit_in_terminal(new_messages)
         messages.extend(new_messages)
+        if new_messages:
+            messages_by_file[filename].extend(new_messages)
         if options.non_interactive:
             # Collect messages and possibly show them later.
             return
@@ -200,7 +204,7 @@ def flush_errors(new_messages: list[str], serious: bool) -> None:
             ),
             file=stderr,
         )
-    maybe_write_junit_xml(time.time() - t0, serious, messages, options)
+    maybe_write_junit_xml(time.time() - t0, serious, messages, messages_by_file, options)
     return res, messages, blockers
 
 
@@ -1054,6 +1058,12 @@ def add_invertible_flag(
     other_group = parser.add_argument_group(title="Miscellaneous")
     other_group.add_argument("--quickstart-file", help=argparse.SUPPRESS)
     other_group.add_argument("--junit-xml", help="Write junit.xml to the given file")
+    imports_group.add_argument(
+        "--junit-format",
+        choices=["global", "per_file"],
+        default="global",
+        help="If --junit-xml is set, specifies format. global: single test with all errors; per_file: one test entry per file with failures",
+    )
     other_group.add_argument(
         "--find-occurrences",
         metavar="CLASS.MEMBER",
@@ -1483,18 +1493,32 @@ def process_cache_map(
         options.cache_map[source] = (meta_file, data_file)
 
 
-def maybe_write_junit_xml(td: float, serious: bool, messages: list[str], options: Options) -> None:
+def maybe_write_junit_xml(
+    td: float,
+    serious: bool,
+    all_messages: list[str],
+    messages_by_file: dict[str | None, list[str]],
+    options: Options,
+) -> None:
     if options.junit_xml:
         py_version = f"{options.python_version[0]}_{options.python_version[1]}"
-        util.write_junit_xml(
-            td, serious, messages, options.junit_xml, py_version, options.platform
-        )
+        if options.junit_format == "global":
+            util.write_junit_xml(
+                td, serious, {None: all_messages}, options.junit_xml, py_version, options.platform
+            )
+        else:
+            # per_file
+            util.write_junit_xml(
+                td, serious, messages_by_file, options.junit_xml, py_version, options.platform
+            )
 
 
 def fail(msg: str, stderr: TextIO, options: Options) -> NoReturn:
     """Fail with a serious error."""
     stderr.write(f"{msg}\n")
-    maybe_write_junit_xml(0.0, serious=True, messages=[msg], options=options)
+    maybe_write_junit_xml(
+        0.0, serious=True, all_messages=[msg], messages_by_file={None: [msg]}, options=options
+    )
     sys.exit(2)
 
 

mypy/options.py

@@ -255,6 +255,8 @@ def __init__(self) -> None:
         # Write junit.xml to given file
         self.junit_xml: str | None = None
 
+        self.junit_format: str = "global"  # global|per_file
+
         # Caching and incremental checking options
         self.incremental = True
         self.cache_dir = defaults.CACHE_DIR

mypy/test/testerrorstream.py

@@ -29,7 +29,7 @@ def test_error_stream(testcase: DataDrivenTestCase) -> None:
 
     logged_messages: list[str] = []
 
-    def flush_errors(msgs: list[str], serious: bool) -> None:
+    def flush_errors(filename: str | None, msgs: list[str], serious: bool) -> None:
         if msgs:
             logged_messages.append("==== Errors flushed ====")
             logged_messages.extend(msgs)

mypy/test/testgraph.py

@@ -50,7 +50,7 @@ def _make_manager(self) -> BuildManager:
             plugin=Plugin(options),
             plugins_snapshot={},
             errors=errors,
-            flush_errors=lambda msgs, serious: None,
+            flush_errors=lambda filename, msgs, serious: None,
             fscache=fscache,
             stdout=sys.stdout,
             stderr=sys.stderr,

mypy/test/testutil.py

@@ -4,7 +4,7 @@
 from unittest import TestCase, mock
 
 from mypy.inspections import parse_location
-from mypy.util import get_terminal_width
+from mypy.util import _generate_junit_contents, get_terminal_width
 
 
 class TestGetTerminalSize(TestCase):
@@ -20,3 +20,70 @@ def test_get_terminal_size_in_pty_defaults_to_80(self) -> None:
     def test_parse_location_windows(self) -> None:
         assert parse_location(r"C:\test.py:1:1") == (r"C:\test.py", [1, 1])
         assert parse_location(r"C:\test.py:1:1:1:1") == (r"C:\test.py", [1, 1, 1, 1])
+
+
+class TestWriteJunitXml(TestCase):
+    def test_junit_pass(self) -> None:
+        serious = False
+        messages_by_file: dict[str | None, list[str]] = {}
+        expected = """<?xml version="1.0" encoding="utf-8"?>
+<testsuite errors="0" failures="0" name="mypy" skips="0" tests="1" time="1.230">
+  <testcase classname="mypy" file="mypy" line="1" name="mypy-py3.14-test-plat" time="1.230">
+  </testcase>
+</testsuite>
+"""
+        result = _generate_junit_contents(
+            dt=1.23,
+            serious=serious,
+            messages_by_file=messages_by_file,
+            version="3.14",
+            platform="test-plat",
+        )
+        assert result == expected
+
+    def test_junit_fail_two_files(self) -> None:
+        serious = False
+        messages_by_file: dict[str | None, list[str]] = {
+            "file1.py": ["Test failed", "another line"],
+            "file2.py": ["Another failure", "line 2"],
+        }
+        expected = """<?xml version="1.0" encoding="utf-8"?>
+<testsuite errors="0" failures="2" name="mypy" skips="0" tests="2" time="1.230">
+  <testcase classname="mypy" file="file1.py" line="1" name="mypy-py3.14-test-plat file1.py" time="1.230">
+    <failure message="mypy produced messages">Test failed
+another line</failure>
+  </testcase>
+  <testcase classname="mypy" file="file2.py" line="1" name="mypy-py3.14-test-plat file2.py" time="1.230">
+    <failure message="mypy produced messages">Another failure
+line 2</failure>
+  </testcase>
+</testsuite>
+"""
+        result = _generate_junit_contents(
+            dt=1.23,
+            serious=serious,
+            messages_by_file=messages_by_file,
+            version="3.14",
+            platform="test-plat",
+        )
+        assert result == expected
+
+    def test_serious_error(self) -> None:
+        serious = True
+        messages_by_file: dict[str | None, list[str]] = {None: ["Error line 1", "Error line 2"]}
+        expected = """<?xml version="1.0" encoding="utf-8"?>
+<testsuite errors="1" failures="0" name="mypy" skips="0" tests="1" time="1.230">
+  <testcase classname="mypy" file="mypy" line="1" name="mypy-py3.14-test-plat" time="1.230">
+    <failure message="mypy produced messages">Error line 1
+Error line 2</failure>
+  </testcase>
+</testsuite>
+"""
+        result = _generate_junit_contents(
+            dt=1.23,
+            serious=serious,
+            messages_by_file=messages_by_file,
+            version="3.14",
+            platform="test-plat",
+        )
+        assert result == expected

mypy/util.py

@@ -234,45 +234,85 @@ def get_mypy_comments(source: str) -> list[tuple[int, str]]:
     return results
 
 
-PASS_TEMPLATE: Final = """<?xml version="1.0" encoding="utf-8"?>
-<testsuite errors="0" failures="0" name="mypy" skips="0" tests="1" time="{time:.3f}">
-  <testcase classname="mypy" file="mypy" line="1" name="mypy-py{ver}-{platform}" time="{time:.3f}">
-  </testcase>
-</testsuite>
+JUNIT_HEADER_TEMPLATE: Final = """<?xml version="1.0" encoding="utf-8"?>
+<testsuite errors="{errors}" failures="{failures}" name="mypy" skips="0" tests="{tests}" time="{time:.3f}">
 """
 
-FAIL_TEMPLATE: Final = """<?xml version="1.0" encoding="utf-8"?>
-<testsuite errors="0" failures="1" name="mypy" skips="0" tests="1" time="{time:.3f}">
-  <testcase classname="mypy" file="mypy" line="1" name="mypy-py{ver}-{platform}" time="{time:.3f}">
+JUNIT_TESTCASE_FAIL_TEMPLATE: Final = """  <testcase classname="mypy" file="{filename}" line="1" name="{name}" time="{time:.3f}">
     <failure message="mypy produced messages">{text}</failure>
   </testcase>
-</testsuite>
 """
 
-ERROR_TEMPLATE: Final = """<?xml version="1.0" encoding="utf-8"?>
-<testsuite errors="1" failures="0" name="mypy" skips="0" tests="1" time="{time:.3f}">
-  <testcase classname="mypy" file="mypy" line="1" name="mypy-py{ver}-{platform}" time="{time:.3f}">
+JUNIT_ERROR_TEMPLATE: Final = """  <testcase classname="mypy" file="mypy" line="1" name="mypy-py{ver}-{platform}" time="{time:.3f}">
     <error message="mypy produced errors">{text}</error>
   </testcase>
-</testsuite>
 """
 
+JUNIT_TESTCASE_PASS_TEMPLATE: Final = """  <testcase classname="mypy" file="mypy" line="1" name="mypy-py{ver}-{platform}" time="{time:.3f}">
+  </testcase>
+"""
 
-def write_junit_xml(
-    dt: float, serious: bool, messages: list[str], path: str, version: str, platform: str
-) -> None:
-    from xml.sax.saxutils import escape
+JUNIT_FOOTER: Final = """</testsuite>
+"""
 
-    if not messages and not serious:
-        xml = PASS_TEMPLATE.format(time=dt, ver=version, platform=platform)
-    elif not serious:
-        xml = FAIL_TEMPLATE.format(
-            text=escape("\n".join(messages)), time=dt, ver=version, platform=platform
-        )
+
+def _generate_junit_contents(
+    dt: float,
+    serious: bool,
+    messages_by_file: dict[str | None, list[str]],
+    version: str,
+    platform: str,
+) -> str:
+    if serious:
+        failures = 0
+        errors = len(messages_by_file)
     else:
-        xml = ERROR_TEMPLATE.format(
-            text=escape("\n".join(messages)), time=dt, ver=version, platform=platform
-        )
+        failures = len(messages_by_file)
+        errors = 0
+
+    xml = JUNIT_HEADER_TEMPLATE.format(
+        errors=errors,
+        failures=failures,
+        time=dt,
+        # If there are no messages, we still write one "test" indicating success.
+        tests=len(messages_by_file) or 1,
+    )
+
+    if not messages_by_file:
+        xml += JUNIT_TESTCASE_PASS_TEMPLATE.format(time=dt, ver=version, platform=platform)
+    else:
+        for filename, messages in messages_by_file.items():
+            if filename is not None:
+                xml += JUNIT_TESTCASE_FAIL_TEMPLATE.format(
+                    text="\n".join(messages),
+                    filename=filename,
+                    time=dt,
+                    name="mypy-py{ver}-{platform} {filename}".format(
+                        ver=version, platform=platform, filename=filename
+                    ),
+                )
+            else:
+                xml += JUNIT_TESTCASE_FAIL_TEMPLATE.format(
+                    text="\n".join(messages),
+                    filename="mypy",
+                    time=dt,
+                    name="mypy-py{ver}-{platform}".format(ver=version, platform=platform),
+                )
+
+    xml += JUNIT_FOOTER
+
+    return xml
+
+
+def write_junit_xml(
+    dt: float,
+    serious: bool,
+    messages_by_file: dict[str | None, list[str]],
+    path: str,
+    version: str,
+    platform: str,
+) -> None:
+    xml = _generate_junit_contents(dt, serious, messages_by_file, version, platform)
 
     # checks for a directory structure in path and creates folders if needed
     xml_dirs = os.path.dirname(os.path.abspath(path))