24055979590

Committed 06 Apr 2026 11:17PM UTC coverage: 52.37% (-40.5%) from 92.908%

Build # 24055979590

Build Type

Pull #23225

github

Committed by

web-flow

Commit Message

Merge 67474653c into 542ca048d

Pull Request Pull Request #23225: Add --test-show-all-batch-targets to expose all targets in batched pytest

Coverage Stats

6 of 17 new or added lines in 2 files covered. (35.29%)

23030 existing lines in 605 files now uncovered.

31643 of 60422 relevant lines covered (52.37%)

1.05 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

80.42

/src/python/pants/core/util_rules/external_tool.py

# Copyright 2020 Pants project contributors (see CONTRIBUTORS.md).
# Licensed under the Apache License, Version 2.0 (see LICENSE).

from __future__ import annotations

import dataclasses
import logging
import os
import textwrap
from abc import ABCMeta, abstractmethod
from dataclasses import dataclass
from enum import Enum

from packaging.requirements import Requirement

from pants.core.goals.export import (
    ExportedBinary,
    ExportRequest,
    ExportResult,
    ExportResults,
    ExportSubsystem,
)
from pants.core.goals.resolves import ExportableTool, ExportMode
from pants.core.util_rules import archive
from pants.core.util_rules.archive import maybe_extract_archive
from pants.engine.download_file import download_file
from pants.engine.engine_aware import EngineAwareParameter
from pants.engine.fs import CreateDigest, Digest, DownloadFile, FileDigest, FileEntry
from pants.engine.internals.native_engine import RemovePrefix
from pants.engine.internals.selectors import concurrently
from pants.engine.intrinsics import create_digest, get_digest_entries, remove_prefix
from pants.engine.platform import Platform
from pants.engine.rules import collect_rules, implicitly, rule
from pants.engine.unions import UnionMembership, UnionRule
from pants.option.option_types import DictOption, EnumOption, StrListOption, StrOption
from pants.option.subsystem import Subsystem, _construct_subsystem
from pants.util.docutil import doc_url
from pants.util.logging import LogLevel
from pants.util.meta import classproperty
from pants.util.strutil import softwrap

logger = logging.getLogger(__name__)


class UnknownVersion(Exception):
    pass


class ExternalToolError(Exception):
    pass


class UnsupportedVersion(ExternalToolError):
    """The specified version of the tool is not supported, according to the given version
    constraints."""


class UnsupportedVersionUsage(Enum):
    """What action to take in case the requested version of the tool is not supported."""

    RaiseError = "error"
    LogWarning = "warning"


@dataclass(frozen=True)
class ExternalToolRequest:
    download_file_request: DownloadFile
    exe: str
    # Some archive files for tools may have a common path prefix, e.g., representing the platform.
    # If this field is set, strip the common path prefix. If the archive contains just one file
    # will strip all dirs from that file.
    strip_common_path_prefix: bool = False


@dataclass(frozen=True)
class DownloadedExternalTool:
    digest: Digest
    exe: str


@dataclass(frozen=True)
class ExternalToolVersion:
    version: str
    platform: str
    sha256: str
    filesize: int
    url_override: str | None = None

    def encode(self) -> str:
        parts = [self.version, self.platform, self.sha256, str(self.filesize)]
        if self.url_override:
            parts.append(self.url_override)
        return "|".join(parts)

    @classmethod
    def decode(cls, version_str: str) -> ExternalToolVersion:
        parts = [x.strip() for x in version_str.split("|")]
        version, platform, sha256, filesize = parts[:4]
        url_override = parts[4] if len(parts) > 4 else None
        return cls(version, platform, sha256, int(filesize), url_override=url_override)


class ExternalToolOptionsMixin:
    """Common options for implementing subsystem providing an `ExternalToolRequest`."""

    @classproperty
    def name(cls):
        """The name of the tool, for use in user-facing messages.

        Derived from the classname, but subclasses can override, e.g., with a classproperty.
        """
        return cls.__name__.lower()

    @classproperty
    def binary_name(cls):
        """The name of the binary, as it normally known.

        This allows renaming a built binary to what users expect, even when the name is different.
        For example, the binary might be "taplo-linux-x86_64" and the name "Taplo", but users expect
        just "taplo".
        """
        return cls.name.lower()

    # The default values for --version and --known-versions, and the supported versions.
    # Subclasses must set appropriately.
    default_version: str
    default_known_versions: list[str]
    version_constraints: str | None = None

    version = StrOption(
        default=lambda cls: cls.default_version,
        advanced=True,
        help=lambda cls: f"Use this version of {cls.name}."
        + (
            f"\n\nSupported {cls.name} versions: {cls.version_constraints}"
            if cls.version_constraints
            else ""
        ),
    )

    # Note that you can compute the length and sha256 conveniently with:
    #   `curl -L $URL | tee >(wc -c) >(shasum -a 256) >/dev/null`
    known_versions = StrListOption(
        default=lambda cls: cls.default_known_versions,
        advanced=True,
        help=textwrap.dedent(
            f"""
        Known versions to verify downloads against.

        Each element is a pipe-separated string of `version|platform|sha256|length` or
        `version|platform|sha256|length|url_override`, where:

          - `version` is the version string
          - `platform` is one of `[{",".join(Platform.__members__.keys())}]`
          - `sha256` is the 64-character hex representation of the expected sha256
            digest of the download file, as emitted by `shasum -a 256`
          - `length` is the expected length of the download file in bytes, as emitted by
            `wc -c`
          - (Optional) `url_override` is a specific url to use instead of the normally
            generated url for this version

        E.g., `3.1.2|macos_x86_64|6d0f18cd84b918c7b3edd0203e75569e0c7caecb1367bbbe409b44e28514f5be|42813`.
        and `3.1.2|macos_arm64 |aca5c1da0192e2fd46b7b55ab290a92c5f07309e7b0ebf4e45ba95731ae98291|50926|https://example.mac.org/bin/v3.1.2/mac-aarch64-v3.1.2.tgz`.

        Values are space-stripped, so pipes can be indented for readability if necessary.
        """
        ),
    )


class ExternalTool(Subsystem, ExportableTool, ExternalToolOptionsMixin, metaclass=ABCMeta):
    """Configuration for an invocable tool that we download from an external source.

    Subclass this to configure a specific tool.


    Idiomatic use:

    class MyExternalTool(ExternalTool):
        options_scope = "my-external-tool"
        default_version = "1.2.3"
        default_known_versions = [
          "1.2.3|linux_arm64 |feed6789feed6789feed6789feed6789feed6789feed6789feed6789feed6789|112233",
          "1.2.3|linux_x86_64|cafebabacafebabacafebabacafebabacafebabacafebabacafebabacafebaba|878986",
          "1.2.3|macos_arm64 |deadbeefdeadbeefdeadbeefdeadbeefdeadbeefdeadbeefdeadbeefdeadbeef|222222",
          "1.2.3|macos_x86_64|1234abcd1234abcd1234abcd1234abcd1234abcd1234abcd1234abcd1234abcd|333333",
        ]

        version_constraints = ">=1.2.3, <2.0"

        def generate_url(self, plat: Platform) -> str:
            ...

        def generate_exe(self, plat: Platform) -> str:
            return "./path-to/binary

    @rule
    async def my_rule(my_external_tool: MyExternalTool, platform: Platform) -> Foo:
        downloaded_tool = await download_external_tool(my_external_tool.get_request(platform))
        ...
    """

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.check_version_constraints()

    export_mode = ExportMode.binary

    use_unsupported_version = EnumOption(
        advanced=True,
        help=lambda cls: textwrap.dedent(
            f"""
                What action to take in case the requested version of {cls.name} is not supported.

                Supported {cls.name} versions: {cls.version_constraints if cls.version_constraints else "unspecified"}
                """
        ),
        default=UnsupportedVersionUsage.RaiseError,
    )

    @abstractmethod
    def generate_url(self, plat: Platform) -> str:
        """Returns the URL for the given version of the tool, runnable on the given os+arch.

        Implementations should raise ExternalToolError if they cannot resolve the arguments
        to a URL. The raised exception need not have a message - a sensible one will be generated.
        """

    def generate_exe(self, plat: Platform) -> str:
        """Returns the path to the tool executable.

        If the downloaded artifact is the executable itself, you can leave this unimplemented.

        If the downloaded artifact is an archive, this should be overridden to provide a
        relative path in the downloaded archive, e.g. `./bin/protoc`.
        """
        return f"./{self.generate_url(plat).rsplit('/', 1)[-1]}"

    def known_version(self, plat: Platform) -> ExternalToolVersion | None:
        for known_version in self.known_versions:
            tool_version = self.decode_known_version(known_version)
            if plat.value == tool_version.platform and tool_version.version == self.version:
                return tool_version
        return None

    def get_request(self, plat: Platform) -> ExternalToolRequest:
        """Generate a request for this tool."""

        tool_version = self.known_version(plat)
        if tool_version:
            return self.get_request_for(
                tool_version.platform,
                tool_version.sha256,
                tool_version.filesize,
                url_override=tool_version.url_override,
            )
        raise UnknownVersion(
            softwrap(
                f"""
                No known version of {self.name} {self.version} for {plat.value} found in
                {self.known_versions}
                """
            )
        )

    @classmethod
    def decode_known_version(cls, known_version: str) -> ExternalToolVersion:
        try:
            return ExternalToolVersion.decode(known_version)
        except ValueError:
            raise ExternalToolError(
                f"Bad value for [{cls.options_scope}].known_versions: {known_version}"
            )

    @classmethod
    def split_known_version_str(cls, known_version: str) -> tuple[str, str, str, int]:
        version = cls.decode_known_version(known_version)
        return version.version, version.platform, version.sha256, version.filesize

    def get_request_for(
        self, plat_val: str, sha256: str, length: int, url_override: str | None = None
    ) -> ExternalToolRequest:
        """Generate a request for this tool from the given info."""
        plat = Platform(plat_val)
        digest = FileDigest(fingerprint=sha256, serialized_bytes_length=length)
        try:
            url = url_override or self.generate_url(plat)
            exe = self.generate_exe(plat)
        except ExternalToolError as e:
            raise ExternalToolError(
                f"Couldn't find {self.name} version {self.version} on {plat.value}"
            ) from e
        return ExternalToolRequest(DownloadFile(url=url, expected_digest=digest), exe)

    def check_version_constraints(self) -> None:
        if not self.version_constraints:
            return None
        # Note that this is not a Python requirement. We're just hackily piggybacking off
        # packaging.requirements.Requirement's ability to check version constraints.
        constraints = Requirement(f"{self.name}{self.version_constraints}")
        if constraints.specifier.contains(self.version):
            # all ok
            return None

        msg = [
            f"The option [{self.options_scope}].version is set to {self.version}, which is not "
            f"compatible with what this release of Pants expects: {constraints}.",
            "Please update the version to a supported value, or consider using a different Pants",
            "release if you cannot change the version.",
        ]

        if self.use_unsupported_version is UnsupportedVersionUsage.LogWarning:
            msg.extend(
                [
                    "Alternatively, you can ignore this warning (at your own peril) by adding this",
                    "to the GLOBAL section of pants.toml:",
                    f'ignore_warnings = ["The option [{self.options_scope}].version is set to"].',
                ]
            )
            logger.warning(" ".join(msg))
        elif self.use_unsupported_version is UnsupportedVersionUsage.RaiseError:
            msg.append(
                softwrap(
                    f"""
                Alternatively, update [{self.options_scope}].use_unsupported_version to be
                'warning'.
                """
                )
            )
            raise UnsupportedVersion(" ".join(msg))


class TemplatedExternalToolOptionsMixin(ExternalToolOptionsMixin):
    """Common options for implementing a subsystem providing an `ExternalToolRequest` via a URL
    template."""

    default_url_template: str
    default_url_platform_mapping: dict[str, str] | None = None

    url_template = StrOption(
        default=lambda cls: cls.default_url_template,
        advanced=True,
        help=softwrap(
            f"""
            URL to download the tool, either as a single binary file or a compressed file
            (e.g. zip file). You can change this to point to your own hosted file, e.g. to
            work with proxies or for access via the filesystem through a `file:$abspath` URL (e.g.
            `file:/this/is/absolute`, possibly by
            [templating the buildroot in a config file]({doc_url("docs/using-pants/key-concepts/options#config-file-entries")})).

            Use `{{version}}` to have the value from `--version` substituted, and `{{platform}}` to
            have a value from `--url-platform-mapping` substituted in, depending on the
            current platform. For example,
            https://github.com/.../protoc-{{version}}-{{platform}}.zip.
            """
        ),
    )

    url_platform_mapping = DictOption[str](
        "--url-platform-mapping",
        default=lambda cls: cls.default_url_platform_mapping,
        advanced=True,
        help=softwrap(
            """
            A dictionary mapping platforms to strings to be used when generating the URL
            to download the tool.

            In `--url-template`, anytime the `{platform}` string is used, Pants will determine the
            current platform, and substitute `{platform}` with the respective value from your dictionary.

            For example, if you define `{"macos_x86_64": "apple-darwin", "linux_x86_64": "unknown-linux"}`,
            and run Pants on Linux with an intel architecture, then `{platform}` will be substituted
            in the `--url-template` option with `unknown-linux`.
            """
        ),
    )


class TemplatedExternalTool(ExternalTool, TemplatedExternalToolOptionsMixin):
    """Extends the ExternalTool to allow url templating for custom/self-hosted source.

    In addition to ExternalTool functionalities, it is needed to set, e.g.:

    default_url_template = "https://tool.url/{version}/{platform}-mytool.zip"
    default_url_platform_mapping = {
        "macos_x86_64": "osx_intel",
        "macos_arm64": "osx_arm",
        "linux_x86_64": "linux",
    }

    The platform mapping dict is optional.
    """

    def generate_url(self, plat: Platform) -> str:
        platform = self.url_platform_mapping.get(plat.value, "")
        return self.url_template.format(version=self.version, platform=platform)


@rule(level=LogLevel.DEBUG)
async def download_external_tool(request: ExternalToolRequest) -> DownloadedExternalTool:
    # Download and extract.
    maybe_archive_digest = await download_file(request.download_file_request, **implicitly())
    extracted_archive = await maybe_extract_archive(**implicitly(maybe_archive_digest))
    digest = extracted_archive.digest
    digest_entries = await get_digest_entries(digest)
    if request.strip_common_path_prefix:
        paths = tuple(entry.path for entry in digest_entries)
        if len(paths) == 0:
            raise ExternalToolError(
                f"No entries found in archive {request.download_file_request.url}"
            )
        if len(paths) == 1:
            commonpath = os.path.dirname(paths[0])
        else:
            commonpath = os.path.commonpath(paths)
        digest = await remove_prefix(RemovePrefix(extracted_archive.digest, commonpath))

    # Confirm executable.
    exe_path = request.exe.lstrip("./")
    is_not_executable = False
    updated_digest_entries = []
    for entry in digest_entries:
        if isinstance(entry, FileEntry) and entry.path == exe_path and not entry.is_executable:
            # We should recreate the digest with the executable bit set.
            is_not_executable = True
            entry = dataclasses.replace(entry, is_executable=True)
        updated_digest_entries.append(entry)
    if is_not_executable:
        digest = await create_digest(CreateDigest(updated_digest_entries))

    return DownloadedExternalTool(digest, request.exe)


@dataclass(frozen=True)
class ExportExternalToolRequest(ExportRequest):
    pass
    # tool: type[ExternalTool]


@dataclass(frozen=True)
class _ExportExternalToolForResolveRequest(EngineAwareParameter):
    resolve: str


@dataclass(frozen=True)
class MaybeExportResult:
    result: ExportResult | None


@rule(level=LogLevel.DEBUG)
async def export_external_tool(
    req: _ExportExternalToolForResolveRequest, platform: Platform, union_membership: UnionMembership
) -> MaybeExportResult:
    """Export a downloadable tool. Downloads all the tools to `bins`, and symlinks the primary exe
    to the `bin` directory.

    We use the last segment of the exe instead of the resolve because:
    - it's probably the exe name people expect
    - avoids clutter from the resolve name (ex "tfsec" instead of "terraform-tfsec")
    """
    exportables = ExportableTool.filter_for_subclasses(
        union_membership,
        ExternalTool,  # type:ignore[type-abstract]  # ExternalTool is abstract, and mypy doesn't like that we might return it
    )
    maybe_exportable = exportables.get(req.resolve)
    if not maybe_exportable:
        return MaybeExportResult(None)

    tool = await _construct_subsystem(maybe_exportable)
    downloaded_tool = await download_external_tool(tool.get_request(platform))

    dest = os.path.join("bins", tool.name)

    exe = tool.generate_exe(platform)
    return MaybeExportResult(
        ExportResult(
            description=f"Export tool {req.resolve}",
            reldir=dest,
            digest=downloaded_tool.digest,
            resolve=req.resolve,
            exported_binaries=(ExportedBinary(name=tool.binary_name, path_in_export=exe),),
        )
    )


@rule
async def export_external_tools(
    request: ExportExternalToolRequest, export: ExportSubsystem
) -> ExportResults:
    maybe_tools = await concurrently(
        export_external_tool(_ExportExternalToolForResolveRequest(resolve), **implicitly())
        for resolve in export.binaries
    )
    return ExportResults(tool.result for tool in maybe_tools if tool.result is not None)


def rules():
    return (
        *collect_rules(),
        *archive.rules(),
        UnionRule(ExportRequest, ExportExternalToolRequest),
    )

1	# Copyright 2020 Pants project contributors (see CONTRIBUTORS.md).
2	# Licensed under the Apache License, Version 2.0 (see LICENSE).
3
4	from __future__ import annotations	2✔
5
6	import dataclasses	2✔
7	import logging	2✔
8	import os	2✔
9	import textwrap	2✔
10	from abc import ABCMeta, abstractmethod	2✔
11	from dataclasses import dataclass	2✔
12	from enum import Enum	2✔
13
14	from packaging.requirements import Requirement	2✔
15
16	from pants.core.goals.export import (	2✔
17	ExportedBinary,
18	ExportRequest,
19	ExportResult,
20	ExportResults,
21	ExportSubsystem,
22	)
23	from pants.core.goals.resolves import ExportableTool, ExportMode	2✔
24	from pants.core.util_rules import archive	2✔
25	from pants.core.util_rules.archive import maybe_extract_archive	2✔
26	from pants.engine.download_file import download_file	2✔
27	from pants.engine.engine_aware import EngineAwareParameter	2✔
28	from pants.engine.fs import CreateDigest, Digest, DownloadFile, FileDigest, FileEntry	2✔
29	from pants.engine.internals.native_engine import RemovePrefix	2✔
30	from pants.engine.internals.selectors import concurrently	2✔
31	from pants.engine.intrinsics import create_digest, get_digest_entries, remove_prefix	2✔
32	from pants.engine.platform import Platform	2✔
33	from pants.engine.rules import collect_rules, implicitly, rule	2✔
34	from pants.engine.unions import UnionMembership, UnionRule	2✔
35	from pants.option.option_types import DictOption, EnumOption, StrListOption, StrOption	2✔
36	from pants.option.subsystem import Subsystem, _construct_subsystem	2✔
37	from pants.util.docutil import doc_url	2✔
38	from pants.util.logging import LogLevel	2✔
39	from pants.util.meta import classproperty	2✔
40	from pants.util.strutil import softwrap	2✔
41
42	logger = logging.getLogger(__name__)	2✔
43
44
45	class UnknownVersion(Exception):	2✔
46	pass	2✔
47
48
49	class ExternalToolError(Exception):	2✔
50	pass	2✔
51
52
53	class UnsupportedVersion(ExternalToolError):	2✔
54	"""The specified version of the tool is not supported, according to the given version
55	constraints."""
56
57
58	class UnsupportedVersionUsage(Enum):	2✔
59	"""What action to take in case the requested version of the tool is not supported."""
60
61	RaiseError = "error"	2✔
62	LogWarning = "warning"	2✔
63
64
65	@dataclass(frozen=True)	2✔
66	class ExternalToolRequest:	2✔
67	download_file_request: DownloadFile	2✔
68	exe: str	2✔
69	# Some archive files for tools may have a common path prefix, e.g., representing the platform.
70	# If this field is set, strip the common path prefix. If the archive contains just one file
71	# will strip all dirs from that file.
72	strip_common_path_prefix: bool = False	2✔
73
74
75	@dataclass(frozen=True)	2✔
76	class DownloadedExternalTool:	2✔
77	digest: Digest	2✔
78	exe: str	2✔
79
80
81	@dataclass(frozen=True)	2✔
82	class ExternalToolVersion:	2✔
83	version: str	2✔
84	platform: str	2✔
85	sha256: str	2✔
86	filesize: int	2✔
87	url_override: str \| None = None	2✔
88
89	def encode(self) -> str:	2✔
90	parts = [self.version, self.platform, self.sha256, str(self.filesize)]	×
91	if self.url_override:	×
92	parts.append(self.url_override)	×
93	return "\|".join(parts)	×
94
95	@classmethod	2✔
96	def decode(cls, version_str: str) -> ExternalToolVersion:	2✔
97	parts = [x.strip() for x in version_str.split("\|")]	2✔
98	version, platform, sha256, filesize = parts[:4]	2✔
99	url_override = parts[4] if len(parts) > 4 else None	2✔
100	return cls(version, platform, sha256, int(filesize), url_override=url_override)	2✔
101
102
103	class ExternalToolOptionsMixin:	2✔
104	"""Common options for implementing subsystem providing an `ExternalToolRequest`."""
105
106	@classproperty	2✔
107	def name(cls):	2✔
108	"""The name of the tool, for use in user-facing messages.
109
110	Derived from the classname, but subclasses can override, e.g., with a classproperty.
111	"""
112	return cls.__name__.lower()	2✔
113
114	@classproperty	2✔
115	def binary_name(cls):	2✔
116	"""The name of the binary, as it normally known.
117
118	This allows renaming a built binary to what users expect, even when the name is different.
119	For example, the binary might be "taplo-linux-x86_64" and the name "Taplo", but users expect
120	just "taplo".
121	"""
122	return cls.name.lower()	2✔
123
124	# The default values for --version and --known-versions, and the supported versions.
125	# Subclasses must set appropriately.
126	default_version: str	2✔
127	default_known_versions: list[str]	2✔
128	version_constraints: str \| None = None	2✔
129
130	version = StrOption(	2✔
131	default=lambda cls: cls.default_version,
132	advanced=True,
133	help=lambda cls: f"Use this version of {cls.name}."
134	+ (
135	f"\n\nSupported {cls.name} versions: {cls.version_constraints}"
136	if cls.version_constraints
137	else ""
138	),
139	)
140
141	# Note that you can compute the length and sha256 conveniently with:
142	# `curl -L $URL \| tee >(wc -c) >(shasum -a 256) >/dev/null`
143	known_versions = StrListOption(	2✔
144	default=lambda cls: cls.default_known_versions,
145	advanced=True,
146	help=textwrap.dedent(
147	f"""
148	Known versions to verify downloads against.
149
150	Each element is a pipe-separated string of `version\|platform\|sha256\|length` or
151	`version\|platform\|sha256\|length\|url_override`, where:
152
153	- `version` is the version string
154	- `platform` is one of `[{",".join(Platform.__members__.keys())}]`
155	- `sha256` is the 64-character hex representation of the expected sha256
156	digest of the download file, as emitted by `shasum -a 256`
157	- `length` is the expected length of the download file in bytes, as emitted by
158	`wc -c`
159	- (Optional) `url_override` is a specific url to use instead of the normally
160	generated url for this version
161
162	E.g., `3.1.2\|macos_x86_64\|6d0f18cd84b918c7b3edd0203e75569e0c7caecb1367bbbe409b44e28514f5be\|42813`.
163	and `3.1.2\|macos_arm64 \|aca5c1da0192e2fd46b7b55ab290a92c5f07309e7b0ebf4e45ba95731ae98291\|50926\|https://example.mac.org/bin/v3.1.2/mac-aarch64-v3.1.2.tgz`.
164
165	Values are space-stripped, so pipes can be indented for readability if necessary.
166	"""
167	),
168	)
169
170
171	class ExternalTool(Subsystem, ExportableTool, ExternalToolOptionsMixin, metaclass=ABCMeta):	2✔
172	"""Configuration for an invocable tool that we download from an external source.
173
174	Subclass this to configure a specific tool.
175
176
177	Idiomatic use:
178
179	class MyExternalTool(ExternalTool):
180	options_scope = "my-external-tool"
181	default_version = "1.2.3"
182	default_known_versions = [
183	"1.2.3\|linux_arm64 \|feed6789feed6789feed6789feed6789feed6789feed6789feed6789feed6789\|112233",
184	"1.2.3\|linux_x86_64\|cafebabacafebabacafebabacafebabacafebabacafebabacafebabacafebaba\|878986",
185	"1.2.3\|macos_arm64 \|deadbeefdeadbeefdeadbeefdeadbeefdeadbeefdeadbeefdeadbeefdeadbeef\|222222",
186	"1.2.3\|macos_x86_64\|1234abcd1234abcd1234abcd1234abcd1234abcd1234abcd1234abcd1234abcd\|333333",
187	]
188
189	version_constraints = ">=1.2.3, <2.0"
190
191	def generate_url(self, plat: Platform) -> str:
192	...
193
194	def generate_exe(self, plat: Platform) -> str:
195	return "./path-to/binary
196
197	@rule
198	async def my_rule(my_external_tool: MyExternalTool, platform: Platform) -> Foo:
199	downloaded_tool = await download_external_tool(my_external_tool.get_request(platform))
200	...
201	"""
202
203	def __init__(self, args, *kwargs):	2✔
204	super().__init__(args, *kwargs)	2✔
205	self.check_version_constraints()	2✔
206
207	export_mode = ExportMode.binary	2✔
208
209	use_unsupported_version = EnumOption(	2✔
210	advanced=True,
211	help=lambda cls: textwrap.dedent(
212	f"""
213	What action to take in case the requested version of {cls.name} is not supported.
214
215	Supported {cls.name} versions: {cls.version_constraints if cls.version_constraints else "unspecified"}
216	"""
217	),
218	default=UnsupportedVersionUsage.RaiseError,
219	)
220
221	@abstractmethod	2✔
222	def generate_url(self, plat: Platform) -> str:	2✔
223	"""Returns the URL for the given version of the tool, runnable on the given os+arch.
224
225	Implementations should raise ExternalToolError if they cannot resolve the arguments
226	to a URL. The raised exception need not have a message - a sensible one will be generated.
227	"""
228
229	def generate_exe(self, plat: Platform) -> str:	2✔
230	"""Returns the path to the tool executable.
231
232	If the downloaded artifact is the executable itself, you can leave this unimplemented.
233
234	If the downloaded artifact is an archive, this should be overridden to provide a
235	relative path in the downloaded archive, e.g. `./bin/protoc`.
236	"""
237	return f"./{self.generate_url(plat).rsplit('/', 1)[-1]}"	2✔
238
239	def known_version(self, plat: Platform) -> ExternalToolVersion \| None:	2✔
240	for known_version in self.known_versions:	2✔
241	tool_version = self.decode_known_version(known_version)	2✔
242	if plat.value == tool_version.platform and tool_version.version == self.version:	2✔
243	return tool_version	2✔
UNCOV 244	return None	×
245
246	def get_request(self, plat: Platform) -> ExternalToolRequest:	2✔
247	"""Generate a request for this tool."""
248
249	tool_version = self.known_version(plat)	2✔
250	if tool_version:	2✔
251	return self.get_request_for(	2✔
252	tool_version.platform,
253	tool_version.sha256,
254	tool_version.filesize,
255	url_override=tool_version.url_override,
256	)
UNCOV 257	raise UnknownVersion(	×
258	softwrap(
259	f"""
260	No known version of {self.name} {self.version} for {plat.value} found in
261	{self.known_versions}
262	"""
263	)
264	)
265
266	@classmethod	2✔
267	def decode_known_version(cls, known_version: str) -> ExternalToolVersion:	2✔
268	try:	2✔
269	return ExternalToolVersion.decode(known_version)	2✔
270	except ValueError:	×
271	raise ExternalToolError(	×
272	f"Bad value for [{cls.options_scope}].known_versions: {known_version}"
273	)
274
275	@classmethod	2✔
276	def split_known_version_str(cls, known_version: str) -> tuple[str, str, str, int]:	2✔
277	version = cls.decode_known_version(known_version)	×
278	return version.version, version.platform, version.sha256, version.filesize	×
279
280	def get_request_for(	2✔
281	self, plat_val: str, sha256: str, length: int, url_override: str \| None = None
282	) -> ExternalToolRequest:
283	"""Generate a request for this tool from the given info."""
284	plat = Platform(plat_val)	2✔
285	digest = FileDigest(fingerprint=sha256, serialized_bytes_length=length)	2✔
286	try:	2✔
287	url = url_override or self.generate_url(plat)	2✔
288	exe = self.generate_exe(plat)	2✔
289	except ExternalToolError as e:	×
290	raise ExternalToolError(	×
291	f"Couldn't find {self.name} version {self.version} on {plat.value}"
292	) from e
293	return ExternalToolRequest(DownloadFile(url=url, expected_digest=digest), exe)	2✔
294
295	def check_version_constraints(self) -> None:	2✔
296	if not self.version_constraints:	2✔
297	return None	2✔
298	# Note that this is not a Python requirement. We're just hackily piggybacking off
299	# packaging.requirements.Requirement's ability to check version constraints.
300	constraints = Requirement(f"{self.name}{self.version_constraints}")	2✔
301	if constraints.specifier.contains(self.version):	2✔
302	# all ok
303	return None	2✔
304
UNCOV 305	msg = [	×
306	f"The option [{self.options_scope}].version is set to {self.version}, which is not "
307	f"compatible with what this release of Pants expects: {constraints}.",
308	"Please update the version to a supported value, or consider using a different Pants",
309	"release if you cannot change the version.",
310	]
311
UNCOV 312	if self.use_unsupported_version is UnsupportedVersionUsage.LogWarning:	×
UNCOV 313	msg.extend(	×
314	[
315	"Alternatively, you can ignore this warning (at your own peril) by adding this",
316	"to the GLOBAL section of pants.toml:",
317	f'ignore_warnings = ["The option [{self.options_scope}].version is set to"].',
318	]
319	)
UNCOV 320	logger.warning(" ".join(msg))	×
UNCOV 321	elif self.use_unsupported_version is UnsupportedVersionUsage.RaiseError:	×
UNCOV 322	msg.append(	×
323	softwrap(
324	f"""
325	Alternatively, update [{self.options_scope}].use_unsupported_version to be
326	'warning'.
327	"""
328	)
329	)
UNCOV 330	raise UnsupportedVersion(" ".join(msg))	×
331
332
333	class TemplatedExternalToolOptionsMixin(ExternalToolOptionsMixin):	2✔
334	"""Common options for implementing a subsystem providing an `ExternalToolRequest` via a URL
335	template."""
336
337	default_url_template: str	2✔
338	default_url_platform_mapping: dict[str, str] \| None = None	2✔
339
340	url_template = StrOption(	2✔
341	default=lambda cls: cls.default_url_template,
342	advanced=True,
343	help=softwrap(
344	f"""
345	URL to download the tool, either as a single binary file or a compressed file
346	(e.g. zip file). You can change this to point to your own hosted file, e.g. to
347	work with proxies or for access via the filesystem through a `file:$abspath` URL (e.g.
348	`file:/this/is/absolute`, possibly by
349	[templating the buildroot in a config file]({doc_url("docs/using-pants/key-concepts/options#config-file-entries")})).
350
351	Use `{{version}}` to have the value from `--version` substituted, and `{{platform}}` to
352	have a value from `--url-platform-mapping` substituted in, depending on the
353	current platform. For example,
354	https://github.com/.../protoc-{{version}}-{{platform}}.zip.
355	"""
356	),
357	)
358
359	url_platform_mapping = DictOption[str](	2✔
360	"--url-platform-mapping",
361	default=lambda cls: cls.default_url_platform_mapping,
362	advanced=True,
363	help=softwrap(
364	"""
365	A dictionary mapping platforms to strings to be used when generating the URL
366	to download the tool.
367
368	In `--url-template`, anytime the `{platform}` string is used, Pants will determine the
369	current platform, and substitute `{platform}` with the respective value from your dictionary.
370
371	For example, if you define `{"macos_x86_64": "apple-darwin", "linux_x86_64": "unknown-linux"}`,
372	and run Pants on Linux with an intel architecture, then `{platform}` will be substituted
373	in the `--url-template` option with `unknown-linux`.
374	"""
375	),
376	)
377
378
379	class TemplatedExternalTool(ExternalTool, TemplatedExternalToolOptionsMixin):	2✔
380	"""Extends the ExternalTool to allow url templating for custom/self-hosted source.
381
382	In addition to ExternalTool functionalities, it is needed to set, e.g.:
383
384	default_url_template = "https://tool.url/{version}/{platform}-mytool.zip"
385	default_url_platform_mapping = {
386	"macos_x86_64": "osx_intel",
387	"macos_arm64": "osx_arm",
388	"linux_x86_64": "linux",
389	}
390
391	The platform mapping dict is optional.
392	"""
393
394	def generate_url(self, plat: Platform) -> str:	2✔
395	platform = self.url_platform_mapping.get(plat.value, "")	2✔
396	return self.url_template.format(version=self.version, platform=platform)	2✔
397
398
399	@rule(level=LogLevel.DEBUG)	2✔
400	async def download_external_tool(request: ExternalToolRequest) -> DownloadedExternalTool:	2✔
401	# Download and extract.
402	maybe_archive_digest = await download_file(request.download_file_request, **implicitly())	2✔
403	extracted_archive = await maybe_extract_archive(**implicitly(maybe_archive_digest))	2✔
404	digest = extracted_archive.digest	2✔
405	digest_entries = await get_digest_entries(digest)	2✔
406	if request.strip_common_path_prefix:	2✔
407	paths = tuple(entry.path for entry in digest_entries)	×
408	if len(paths) == 0:	×
409	raise ExternalToolError(	×
410	f"No entries found in archive {request.download_file_request.url}"
411	)
412	if len(paths) == 1:	×
413	commonpath = os.path.dirname(paths[0])	×
414	else:
415	commonpath = os.path.commonpath(paths)	×
416	digest = await remove_prefix(RemovePrefix(extracted_archive.digest, commonpath))	×
417
418	# Confirm executable.
419	exe_path = request.exe.lstrip("./")	2✔
420	is_not_executable = False	2✔
421	updated_digest_entries = []	2✔
422	for entry in digest_entries:	2✔
423	if isinstance(entry, FileEntry) and entry.path == exe_path and not entry.is_executable:	2✔
424	# We should recreate the digest with the executable bit set.
425	is_not_executable = True	2✔
426	entry = dataclasses.replace(entry, is_executable=True)	2✔
427	updated_digest_entries.append(entry)	2✔
428	if is_not_executable:	2✔
429	digest = await create_digest(CreateDigest(updated_digest_entries))	2✔
430
431	return DownloadedExternalTool(digest, request.exe)	2✔
432
433
434	@dataclass(frozen=True)	2✔
435	class ExportExternalToolRequest(ExportRequest):	2✔
436	pass	2✔
437	# tool: type[ExternalTool]
438
439
440	@dataclass(frozen=True)	2✔
441	class _ExportExternalToolForResolveRequest(EngineAwareParameter):	2✔
442	resolve: str	2✔
443
444
445	@dataclass(frozen=True)	2✔
446	class MaybeExportResult:	2✔
447	result: ExportResult \| None	2✔
448
449
450	@rule(level=LogLevel.DEBUG)	2✔
451	async def export_external_tool(	2✔
452	req: _ExportExternalToolForResolveRequest, platform: Platform, union_membership: UnionMembership
453	) -> MaybeExportResult:
454	"""Export a downloadable tool. Downloads all the tools to `bins`, and symlinks the primary exe
455	to the `bin` directory.
456
457	We use the last segment of the exe instead of the resolve because:
458	- it's probably the exe name people expect
459	- avoids clutter from the resolve name (ex "tfsec" instead of "terraform-tfsec")
460	"""
UNCOV 461	exportables = ExportableTool.filter_for_subclasses(	×
462	union_membership,
463	ExternalTool, # type:ignore[type-abstract] # ExternalTool is abstract, and mypy doesn't like that we might return it
464	)
UNCOV 465	maybe_exportable = exportables.get(req.resolve)	×
UNCOV 466	if not maybe_exportable:	×
467	return MaybeExportResult(None)	×
468
UNCOV 469	tool = await _construct_subsystem(maybe_exportable)	×
UNCOV 470	downloaded_tool = await download_external_tool(tool.get_request(platform))	×
471
UNCOV 472	dest = os.path.join("bins", tool.name)	×
473
UNCOV 474	exe = tool.generate_exe(platform)	×
UNCOV 475	return MaybeExportResult(	×
476	ExportResult(
477	description=f"Export tool {req.resolve}",
478	reldir=dest,
479	digest=downloaded_tool.digest,
480	resolve=req.resolve,
481	exported_binaries=(ExportedBinary(name=tool.binary_name, path_in_export=exe),),
482	)
483	)
484
485
486	@rule	2✔
487	async def export_external_tools(	2✔
488	request: ExportExternalToolRequest, export: ExportSubsystem
489	) -> ExportResults:
490	maybe_tools = await concurrently(	×
491	export_external_tool(_ExportExternalToolForResolveRequest(resolve), **implicitly())
492	for resolve in export.binaries
493	)
494	return ExportResults(tool.result for tool in maybe_tools if tool.result is not None)	×
495
496
497	def rules():	2✔
498	return (	2✔
499	*collect_rules(),
500	*archive.rules(),
501	UnionRule(ExportRequest, ExportExternalToolRequest),
502	)

pantsbuild / pants / 24055979590

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous