test(coverage): check the coverage collection with coverage from a pip hub

MODULE.bazel

@@ -401,6 +401,26 @@ 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_override(
+    coverage_tool = "@coverage_pip//coverage:extracted_whl_files",
+    python_version = "3.13.13",
+)
+
 # For testing an arbitrary runtime triggered by a custom flag.
 # See //tests/toolchains:custom_platform_toolchain_test
 dev_python.single_version_platform_override(
@@ -461,7 +481,12 @@ dev_pip.parse(
     python_version = "3.14",
     requirements_lock = "//tests/multi_pypi/beta:requirements.txt",
 )
-use_repo(dev_pip, "dev_pip", "pypi_alpha", "pypi_beta", "pypiserver")
+dev_pip.parse(
+    hub_name = "coverage_pip",
+    python_version = "3.13",
+    uv_lock = "//dev/update_deps/coverage:uv.lock",
+)
+use_repo(dev_pip, "coverage_pip", "dev_pip", "pypi_alpha", "pypi_beta", "pypiserver")
 
 # Bazel integration test setup below
 

dev/update_deps/BUILD.bazel

@@ -35,10 +35,12 @@ py_binary(
     name = "update_coverage_deps",
     srcs = ["update_coverage_deps.py"],
     data = [
+        "//dev/update_deps/coverage:uv.lock",
         "//python/private:coverage_deps_filegroup",
     ],
     env = {
         "UPDATE_FILE": "$(rlocationpath //python/private:coverage_deps_filegroup)",
+        "UV_LOCK": "$(rlocationpath //dev/update_deps/coverage:uv.lock)",
     },
     imports = ["../.."],
     deps = [
@@ -76,3 +78,17 @@ pytest_test(
         ":update_file",
     ],
 )
+
+pytest_test(
+    name = "update_coverage_deps_test",
+    srcs = [
+        "update_coverage_deps.py",
+        "update_coverage_deps_test.py",
+    ],
+    imports = ["../.."],
+    target_compatible_with = SUPPORTS_BZLMOD,
+    deps = [
+        ":args",
+        ":update_file",
+    ],
+)

dev/update_deps/coverage/BUILD.bazel

@@ -0,0 +1,13 @@
+load("//python/uv:lock.bzl", "lock")
+
+package(default_visibility = ["//dev/update_deps:__subpackages__"])
+
+exports_files(["uv.lock"])
+
+# The lock file for the coverage.py versions that are bundled with the hermetic
+# python toolchains. See the `pyproject.toml` file for how to update it.
+lock(
+    name = "uv_lock",
+    srcs = ["pyproject.toml"],
+    out = "uv.lock",
+)

dev/update_deps/coverage/pyproject.toml

@@ -0,0 +1,16 @@
+# The coverage.py versions that are bundled with the hermetic python toolchains.
+#
+# Newer coverage.py versions drop support for older python versions, so each
+# version is used for the python versions given by the marker.
+#
+# To update, edit this file and run:
+#   bazel run //dev/update_deps/coverage:uv_lock.update
+#   bazel run //dev/update_deps:update_coverage_deps
+[project]
+name = "rules-python-coverage"
+version = "0.0.0"
+requires-python = ">=3.9"
+dependencies = [
+    'coverage==7.10.7; python_version < "3.15"',
+    'coverage==7.17.0; python_version >= "3.15"',
+]

dev/update_deps/update_coverage_deps.py

@@ -13,120 +13,90 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-"""A small script to update bazel files within the repo.
+"""Update the bundled coverage wheels from the coverage `uv.lock` file.
 
-We are not running this with 'bazel run' to keep the dependencies minimal
+The coverage versions are defined in `//dev/update_deps/coverage:pyproject.toml`
+and locked with `bazel run //dev/update_deps/coverage:uv_lock.update`.
 """
 
 # NOTE @aignas 2023-01-09: We should only depend on core Python 3 packages.
 import argparse
 import json
 import os
+import pathlib
 import textwrap
-from collections import defaultdict
 from dataclasses import dataclass
 from typing import Any
-from urllib import request
+
+import tomllib
 
 from dev.update_deps.args import path_from_runfiles
 from dev.update_deps.update_file import update_file
 
-# This should be kept in sync with //python:versions.bzl
-_supported_platforms = {
-    # Windows is unsupported right now
-    # "win_amd64": "x86_64-pc-windows-msvc",
-    "manylinux1_x86_64": "x86_64-unknown-linux-gnu",
-    "manylinux2014_x86_64": "x86_64-unknown-linux-gnu",
-    "manylinux2014_aarch64": "aarch64-unknown-linux-gnu",
-    "macosx_11_0_arm64": "aarch64-apple-darwin",
-    "macosx_10_9_x86_64": "x86_64-apple-darwin",
-    "macosx_10_13_x86_64": "x86_64-apple-darwin",
-    "macosx_10_15_x86_64": "x86_64-apple-darwin",
-    ("t", "manylinux1_x86_64"): "x86_64-unknown-linux-gnu-freethreaded",
-    ("t", "manylinux2014_x86_64"): "x86_64-unknown-linux-gnu-freethreaded",
-    ("t", "manylinux2014_aarch64"): "aarch64-unknown-linux-gnu-freethreaded",
-    ("t", "macosx_11_0_arm64"): "aarch64-apple-darwin-freethreaded",
-    ("t", "macosx_10_9_x86_64"): "x86_64-apple-darwin-freethreaded",
-    ("t", "macosx_10_13_x86_64"): "x86_64-apple-darwin-freethreaded",
-    ("t", "macosx_10_15_x86_64"): "x86_64-apple-darwin-freethreaded",
-}
-
 
 @dataclass
-class Dep:
-    name: str
-    platform: str
-    python: str
+class Whl:
+    filename: str
     url: str
-    sha256: str
-
-    @property
-    def repo_name(self):
-        return f"pypi__{self.name}_{self.python}_{self.platform}"
+    hash: str
+    marker: str
 
     def __repr__(self):
-        return "\n".join(
-            [
-                "(",
-                f'    "{self.url}",',
-                f'    "{self.sha256}",',
-                ")",
-            ]
-        )
+        lines = [
+            "struct(",
+            f'    url = "{self.url}",',
+            f'    hash = "{self.hash}",',
+        ]
+        if self.marker:
+            if '"' in self.marker and "'" not in self.marker:
+                marker = f"'{self.marker}'"
+            else:
+                marker = json.dumps(self.marker)
+            lines.append(f"    marker = {marker},")
+        lines.append("),")
+        return "\n".join(lines)
 
 
 @dataclass
-class Deps:
-    deps: list[Dep]
+class Whls:
+    whls: list[Whl]
 
     def __repr__(self):
-        deps = defaultdict(dict)
-        for d in self.deps:
-            deps[d.python][d.platform] = d
-
-        parts = []
-        for python, contents in deps.items():
-            inner = textwrap.indent(
-                "\n".join([f'"{platform}": {d},' for platform, d in contents.items()]),
-                prefix="    ",
+        inner = textwrap.indent(
+            "\n".join([repr(w) for w in self.whls]),
+            prefix="    ",
+        )
+        return "[\n{}\n]".format(inner)
+
+
+def _marker(markers: list[str]) -> str:
+    if len(markers) == 1:
+        return markers[0]
+
+    return " or ".join(f"({m})" for m in markers)
+
+
+def get_whls(uv_lock: dict[str, Any], name: str) -> list[Whl]:
+    """Get all of the wheels for the given package from the parsed uv.lock."""
+    whls = []
+    for pkg in uv_lock.get("package", []):
+        if pkg["name"] != name:
+            continue
+
+        marker = _marker(pkg.get("resolution-markers", []))
+        for wheel in pkg.get("wheels", []):
+            url = wheel["url"]
+            whls.append(
+                Whl(
+                    filename=url.rpartition("/")[2],
+                    url=url,
+                    hash=wheel["hash"],
+                    marker=marker,
+                )
             )
-            parts.append('"{}": {{\n{}\n}},'.format(python, inner))
-        return "{{\n{}\n}}".format(textwrap.indent("\n".join(parts), prefix="    "))
-
-
-def _get_platforms(filename: str, python_version: str):
-    name, _, tail = filename.partition("-")
-    version, _, tail = tail.partition("-")
-    got_python_version, _, tail = tail.partition("-")
-    if python_version != got_python_version:
-        return []
-    abi, _, tail = tail.partition("-")
-
-    platforms, _, tail = tail.rpartition(".")
-    platforms = platforms.split(".")
-
-    return [("t", p) for p in platforms] if abi.endswith("t") else platforms
-
-
-def _map(
-    name: str,
-    filename: str,
-    python_version: str,
-    url: str,
-    digests: dict[str, str],
-    platform: str | tuple[str, str],
-    **kwargs: Any,
-):
-    if platform and platform not in _supported_platforms:
-        return None
-
-    return Dep(
-        name=name,
-        platform=_supported_platforms[platform] if platform else "",
-        python=python_version,
-        url=url,
-        sha256=digests["sha256"],
-    )
+
+    whls.sort(key=lambda x: x.filename)
+    return whls
 
 
 def _parse_args() -> argparse.Namespace:
@@ -138,16 +108,10 @@ def _parse_args() -> argparse.Namespace:
         help="The name of the package",
     )
     parser.add_argument(
-        "version",
-        type=str,
-        help="The version of the package to download",
-    )
-    parser.add_argument(
-        "--py",
-        nargs="+",
-        type=str,
-        default=["cp39", "cp310", "cp311", "cp312", "cp313", "cp314", "cp315"],
-        help="Supported python versions",
+        "--uv-lock",
+        type=path_from_runfiles,
+        default=os.environ.get("UV_LOCK"),
+        help="The path to the uv.lock file, defaults to the value taken from UV_LOCK",
     )
     parser.add_argument(
         "--dry-run",
@@ -166,57 +130,20 @@ def _parse_args() -> argparse.Namespace:
 def main():
     args = _parse_args()
 
-    api_url = f"https://pypi.org/pypi/{args.name}/{args.version}/json"
-    req = request.Request(api_url)
-    with request.urlopen(req) as response:
-        data = json.loads(response.read().decode("utf-8"))
+    uv_lock = tomllib.loads(pathlib.Path(args.uv_lock).read_text())
+    whls = get_whls(uv_lock, args.name)
+    if not whls:
+        raise RuntimeError(f"No wheels for '{args.name}' found in {args.uv_lock}")
 
-    urls = []
-    default_url = None
-    for u in data["urls"]:
-        if u["yanked"]:
-            continue
-
-        if not u["filename"].endswith(".whl"):
-            continue
-
-        if u["filename"].endswith("py3-none-any.whl"):
-            default_url = _map(name=args.name, platform="", **u)
-            continue
-
-        if u["python_version"] not in args.py:
-            continue
-
-        if f"_{u['python_version']}m_" in u["filename"]:
-            continue
-
-        platforms = _get_platforms(
-            u["filename"],
-            u["python_version"],
-        )
-
-        result = [_map(name=args.name, platform=p, **u) for p in platforms]
-        urls.extend(filter(None, result))
-
-    urls.sort(key=lambda x: f"{x.python}_{x.platform}")
-
-    # Update the coverage_deps, which are used to register deps
+    # Update the coverage wheels, the compatible wheel is selected in Starlark
     update_file(
         path=args.update_file,
-        snippet="\n".join(
-            [
-                f"_default = {repr(default_url)}",
-                f"_coverage_deps = {repr(Deps(urls))}",
-                "",
-            ]
-        ),
+        snippet="_coverage_wheels = {}\n".format(repr(Whls(whls))),
         start_marker="# START: maintained by 'bazel run //dev/update_deps:update_coverage_deps <version>'",
         end_marker="# END: maintained by 'bazel run //dev/update_deps:update_coverage_deps <version>'",
         dry_run=args.dry_run,
     )
 
-    return
-
 
 if __name__ == "__main__":
     main()