feat(toolchain): allow using the extracted coverage wheel as the coverage_tool

MODULE.bazel

@@ -401,6 +401,36 @@ dev_python.override(
     register_all_versions = True,
 )
 
+# For testing the coverage collection with the bundled coverage tool.
+# See //tests/coverage:coverage_collection_test
+#
+# NOTE: A full version is used so that the toolchain is named after it and does
+# not shadow the toolchains that are checked in //tests/toolchains.
+dev_python.toolchain(
+    configure_coverage_tool = True,
+    python_version = "3.12.16",
+)
+
+# For testing the coverage collection with the coverage tool from a PyPI hub.
+# See //tests/coverage:coverage_from_pypi_test
+dev_python.toolchain(
+    python_version = "3.13.13",
+)
+
+[
+    dev_python.single_version_platform_override(
+        coverage_tool = "@dev_pip//coverage:extracted_whl_files",
+        platform = platform,
+        python_version = "3.13.13",
+    )
+    for platform in [
+        "aarch64-apple-darwin",
+        "aarch64-unknown-linux-gnu",
+        "x86_64-apple-darwin",
+        "x86_64-unknown-linux-gnu",
+    ]
+]
+
 # For testing an arbitrary runtime triggered by a custom flag.
 # See //tests/toolchains:custom_platform_toolchain_test
 dev_python.single_version_platform_override(

dev/pyproject.toml

@@ -12,6 +12,8 @@ dependencies = [
     "sphinx_rtd_theme >=2.0", # uv insists on downgrading for some reason
     "readthedocs-sphinx-ext",
     "absl-py",
+    # Used to test the coverage collection with coverage from a PyPI hub.
+    "coverage",
     "typing-extensions",
     "sphinx-reredirects",
     "pefile",

docs/coverage.md

@@ -72,3 +72,30 @@ important behaviors of the target are:
 *   It provides a single output file OR it provides an executable output; this
     output is treated as the coverage entry point.
 *   If it provides runfiles, then `runfiles.files` are included into `py_test`.
+*   Otherwise, if one of its output files is `coverage/__main__.py`, then that
+    file is treated as the coverage entry point. This allows using the files
+    of an extracted `coverage` wheel.
+
+(coverage-from-pypi)=
+### Using `coverage` from PyPI
+
+To use the version of `coverage` that is locked in your requirements instead of
+the one bundled with `rules_python`, point the toolchain to the extracted files
+of the `coverage` package in your {obj}`pip.parse` hub:
+
+```starlark
+python.single_version_platform_override(
+    coverage_tool = "@pypi//coverage:extracted_whl_files",
+    platform = "x86_64-unknown-linux-gnu",
+    python_version = "3.13.2",
+)
+```
+
+Or set {attr}`py_runtime.coverage_tool` to
+`@pypi//coverage:extracted_whl_files` when defining your own runtime.
+
+:::{note}
+Use the `extracted_whl_files` target and not `@pypi//coverage`: the
+`py_library` depends on the Python toolchain itself, so using it as the
+coverage tool of the toolchain creates a dependency cycle.
+:::

news/4433.added.md

@@ -0,0 +1,4 @@
+(toolchains) {attr}`py_runtime.coverage_tool` accepts the extracted files of
+the `coverage` wheel, such as `@pypi//coverage:extracted_whl_files`, so the
+locked version of `coverage` can be used for coverage collection
+([#3902](https://github.com/bazel-contrib/rules_python/issues/3902)).

python/private/py_runtime_rule.bzl

@@ -25,6 +25,24 @@ load(":version.bzl", "version")
 
 _py_builtins = py_internal
 
+_COVERAGE_MAIN = "coverage/__main__.py"
+
+def _coverage_main(files):
+    """Return the `coverage/__main__.py` entry point from the coverage files.
+
+    Args:
+        files: {type}`depset[File]` the files of the coverage tool target, e.g.
+            `@pypi//coverage:extracted_whl_files`.
+
+    Returns:
+        {type}`File | None` the entry point of the coverage tool.
+    """
+    for f in files.to_list():
+        if f.short_path == _COVERAGE_MAIN or f.short_path.endswith("/" + _COVERAGE_MAIN):
+            return f
+
+    return None
+
 def coverage_tool_missing_message(*, coverage_enabled, coverage_tool, label):
     """Build the warning for a selected runtime that cannot produce coverage.
 
@@ -133,7 +151,9 @@ def _py_runtime_impl(ctx):
         elif coverage_di.files_to_run and coverage_di.files_to_run.executable:
             coverage_tool = coverage_di.files_to_run.executable
         else:
-            fail("coverage_tool must be an executable target or must produce exactly one file.")
+            coverage_tool = _coverage_main(coverage_di.files)
+            if not coverage_tool:
+                fail("coverage_tool must be an executable target, must produce exactly one file or must contain '{}'.".format(_COVERAGE_MAIN))
 
         coverage_files = depset(transitive = [
             coverage_di.files,
@@ -311,6 +331,14 @@ The entry point for the tool must be loadable by a Python interpreter (e.g. a
 `.py` or `.pyc` file).  It must accept the command line arguments
 of [`coverage.py`](https://coverage.readthedocs.io), at least including
 the `run` and `lcov` subcommands.
+
+If the target produces multiple files and one of them is
+`coverage/__main__.py`, then it is used as the entry point, e.g.
+`@pypi//coverage:extracted_whl_files`. See {ref}`coverage-from-pypi`.
+
+:::{versionchanged} VERSION_NEXT_FEATURE
+The extracted files of the `coverage` wheel can be used.
+:::
 """,
             ),
             "files": attr.label_list(

tests/coverage/BUILD.bazel

@@ -0,0 +1,62 @@
+load("//python:py_library.bzl", "py_library")
+load("//python:pytest_test.bzl", "pytest_test")
+load("//tests/support:py_reconfig.bzl", "py_reconfig_test")
+load("//tests/support:support.bzl", "NOT_WINDOWS", "SUPPORTS_BZLMOD")
+
+py_library(
+    name = "lib",
+    srcs = ["lib.py"],
+)
+
+# The program that is built with coverage enabled. It is run by the
+# `coverage_collection_test` below, which sets the same environment variables
+# that `bazel coverage` would set.
+py_reconfig_test(
+    name = "instrumented",
+    srcs = ["instrumented.py"],
+    collect_code_coverage = True,
+    main = "instrumented.py",
+    # The bundled coverage tool is configured only for this version, see the
+    # `MODULE.bazel` file.
+    python_version = "3.12.16",
+    tags = ["manual"],
+    # The bundled coverage tool is not supported on Windows and it is only
+    # configured with bzlmod.
+    target_compatible_with = NOT_WINDOWS + SUPPORTS_BZLMOD,
+    deps = [":lib"],
+)
+
+pytest_test(
+    name = "coverage_collection_test",
+    srcs = ["coverage_collection_test.py"],
+    data = [":instrumented"],
+    env = {
+        "INSTRUMENTED": "$(rlocationpath :instrumented)",
+    },
+    target_compatible_with = NOT_WINDOWS + SUPPORTS_BZLMOD,
+    deps = ["//python/runfiles"],
+)
+
+# The same program, but the coverage tool is the extracted `coverage` wheel from
+# the `@dev_pip` hub, see the `MODULE.bazel` file.
+py_reconfig_test(
+    name = "instrumented_pypi",
+    srcs = ["instrumented.py"],
+    collect_code_coverage = True,
+    main = "instrumented.py",
+    python_version = "3.13.13",
+    tags = ["manual"],
+    target_compatible_with = NOT_WINDOWS + SUPPORTS_BZLMOD,
+    deps = [":lib"],
+)
+
+pytest_test(
+    name = "coverage_from_pypi_test",
+    srcs = ["coverage_collection_test.py"],
+    data = [":instrumented_pypi"],
+    env = {
+        "INSTRUMENTED": "$(rlocationpath :instrumented_pypi)",
+    },
+    target_compatible_with = NOT_WINDOWS + SUPPORTS_BZLMOD,
+    deps = ["//python/runfiles"],
+)

tests/coverage/coverage_collection_test.py

@@ -0,0 +1,162 @@
+"""Check that the coverage is collected with the bundled coverage tool.
+
+The tests run the instrumented program with the same environment variables
+that `bazel coverage` sets and check the lcov file that it produces.
+"""
+
+import glob
+import os
+import pathlib
+import subprocess
+import unittest
+
+from python.runfiles import runfiles
+
+_LIB = "tests/coverage/lib.py"
+_MAIN = "tests/coverage/instrumented.py"
+
+
+def _parse_lcov(path):
+    """Parse the line and branch coverage from an lcov file.
+
+    Returns:
+        A dict of the source file to a struct-like dict with the `lines`
+        (line number to hit count) and `branches` (line number to the list of
+        the branch hit counts).
+    """
+    result = {}
+    current = None
+    for line in pathlib.Path(path).read_text().splitlines():
+        if line.startswith("SF:"):
+            current = result.setdefault(
+                line[len("SF:") :],
+                {"lines": {}, "branches": {}},
+            )
+        elif current is None:
+            continue
+        elif line.startswith("DA:"):
+            lineno, hits = line[len("DA:") :].split(",")[:2]
+            current["lines"][int(lineno)] = int(hits)
+        elif line.startswith("BRDA:"):
+            lineno, _, _, taken = line[len("BRDA:") :].split(",")
+            current["branches"].setdefault(int(lineno), []).append(
+                0 if taken == "-" else int(taken)
+            )
+    return result
+
+
+class CoverageCollectionTest(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls):
+        rf = runfiles.Create()
+        assert rf is not None
+        program = rf.Rlocation(os.environ["INSTRUMENTED"])
+        assert program is not None and os.path.exists(program), program
+        lib_path = rf.Rlocation("_main/" + _LIB)
+        assert lib_path is not None
+        cls.lib_lines = pathlib.Path(lib_path).read_text().splitlines()
+
+        tmp = pathlib.Path(os.environ["TEST_TMPDIR"])
+        coverage_dir = tmp / "coverage"
+        coverage_dir.mkdir()
+        manifest = tmp / "coverage_manifest.txt"
+        # Like `bazel coverage`, list all of the instrumented files.
+        manifest.write_text("\n".join([_LIB, _MAIN]) + "\n")
+
+        # Do not leak the coverage settings of this test (e.g. when running
+        # `bazel coverage`) into the instrumented program.
+        env = {
+            k: v
+            for k, v in os.environ.items()
+            if not k.startswith("COVERAGE") and k != "PYTHON_COVERAGE"
+        }
+        env.update(rf.EnvVars())
+        env["COVERAGE"] = "1"
+        env["COVERAGE_DIR"] = str(coverage_dir)
+        env["COVERAGE_MANIFEST"] = str(manifest)
+
+        cls.result = subprocess.run(
+            [program],
+            env=env,
+            capture_output=True,
+            text=True,
+        )
+        cls.lcov_files = glob.glob(str(coverage_dir / "pylcov_*.dat"))
+        cls.lcov = _parse_lcov(cls.lcov_files[0]) if len(cls.lcov_files) == 1 else {}
+
+    def _lineno(self, content):
+        """Return the 1-based line number of the unique line in lib.py."""
+        matches = [
+            i + 1 for i, line in enumerate(self.lib_lines) if line.strip() == content
+        ]
+        self.assertEqual(len(matches), 1, f"'{content}' in {_LIB}: {matches}")
+        return matches[0]
+
+    def _line_hits(self, content):
+        self.assertIn(_LIB, self.lcov, f"lcov: {self.lcov}")
+        lineno = self._lineno(content)
+        lines = self.lcov[_LIB]["lines"]
+        self.assertIn(lineno, lines, f"line {lineno} '{content}' not in lcov: {lines}")
+        return lines[lineno]
+
+    def test_program_succeeds(self):
+        self.assertEqual(
+            self.result.returncode,
+            0,
+            f"stdout:\n{self.result.stdout}\nstderr:\n{self.result.stderr}",
+        )
+        self.assertEqual(self.result.stdout.splitlines(), ["truthy", "hello coverage"])
+
+    def test_single_lcov_file_is_produced(self):
+        self.assertEqual(
+            len(self.lcov_files),
+            1,
+            f"lcov files: {self.lcov_files}\nstderr:\n{self.result.stderr}",
+        )
+
+    def test_instrumented_files_use_relative_paths(self):
+        # NOTE: coverage.py also reports the files from the source directories
+        # that were not imported, so only check the instrumented files.
+        for path in [_LIB, _MAIN]:
+            with self.subTest(path=path):
+                self.assertIn(path, self.lcov, f"lcov files: {sorted(self.lcov)}")
+                self.assertEqual(
+                    [f for f in self.lcov if f != path and f.endswith("/" + path)],
+                    [],
+                )
+
+    def test_executed_lines_are_covered(self):
+        for content in [
+            "if value:",
+            'return "truthy"',
+            'return "hello " + name',
+        ]:
+            with self.subTest(line=content):
+                self.assertGreater(self._line_hits(content), 0)
+
+    def test_definitions_are_covered(self):
+        # Module level statements are executed on import.
+        for content in [
+            "def covered(value):",
+            "def not_covered():",
+            "class Greeter:",
+            "def unused(self):",
+        ]:
+            with self.subTest(line=content):
+                self.assertGreater(self._line_hits(content), 0)
+
+    def test_skipped_branch_is_not_covered(self):
+        self.assertEqual(self._line_hits('return "falsy"'), 0)
+
+    def test_uncalled_function_is_not_covered(self):
+        self.assertEqual(self._line_hits('return "never called"'), 0)
+
+    def test_uncalled_method_is_not_covered(self):
+        self.assertEqual(self._line_hits('return "unused method"'), 0)
+
+    def test_partial_branch_coverage(self):
+        lineno = self._lineno("if value:")
+        branches = self.lcov[_LIB]["branches"]
+        self.assertIn(lineno, branches, f"branches: {branches}")
+        # Only the `True` branch of the `if` was taken.
+        self.assertEqual(sorted(b > 0 for b in branches[lineno]), [False, True])

tests/coverage/instrumented.py

@@ -0,0 +1,7 @@
+"""The program for which the coverage is collected."""
+
+from tests.coverage import lib
+
+if __name__ == "__main__":
+    print(lib.covered(True))
+    print(lib.Greeter().greet("coverage"))

tests/coverage/lib.py

@@ -0,0 +1,22 @@
+"""A small library to check which lines are reported as covered.
+
+The lines are looked up by their content in the tests, so keep them unique.
+"""
+
+
+def covered(value):
+    if value:
+        return "truthy"
+    return "falsy"
+
+
+def not_covered():
+    return "never called"
+
+
+class Greeter:
+    def greet(self, name):
+        return "hello " + name
+
+    def unused(self):
+        return "unused method"