python · JukkaL · Nov 3, 2023 · Nov 1, 2023
diff --git a/mypy/build.py b/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()
 

diff --git a/mypy/config_parser.py b/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,

diff --git a/mypy/main.py b/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)
 
 

diff --git a/mypy/options.py b/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

diff --git a/mypy/test/testerrorstream.py b/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)

diff --git a/mypy/test/testgraph.py b/mypy/test/testgraph.py
@@ -50,7 +50,7 @@ def _make_manager(self) -> BuildManager:
             plugin=Plugin(options),

F438
             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,

diff --git a/mypy/test/testutil.py b/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
diff --git a/mypy/util.py b/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))