23415513592

Committed 22 Mar 2026 11:51PM UTC coverage: 92.92% (-0.004%) from 92.924%

Build # 23415513592

Build Type

Pull #23192

github

Committed by

web-flow

Commit Message

Merge 1456ad98b into 30dee98e2

Pull Request Pull Request #23192: strip common prefixes off ExternalTool archive paths

Coverage Stats

12 of 17 new or added lines in 2 files covered. (70.59%)

8 existing lines in 2 files now uncovered.

91521 of 98494 relevant lines covered (92.92%)

4.05 hits per line

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

90.37

/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) == 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	12✔
5
6	import dataclasses	12✔
7	import logging	12✔
8	import os	12✔
9	import textwrap	12✔
10	from abc import ABCMeta, abstractmethod	12✔
11	from dataclasses import dataclass	12✔
12	from enum import Enum	12✔
13
14	from packaging.requirements import Requirement	12✔
15
16	from pants.core.goals.export import (	12✔
17	ExportedBinary,
18	ExportRequest,
19	ExportResult,
20	ExportResults,
21	ExportSubsystem,
22	)
23	from pants.core.goals.resolves import ExportableTool, ExportMode	12✔
24	from pants.core.util_rules import archive	12✔
25	from pants.core.util_rules.archive import maybe_extract_archive	12✔
26	from pants.engine.download_file import download_file	12✔
27	from pants.engine.engine_aware import EngineAwareParameter	12✔
28	from pants.engine.fs import CreateDigest, Digest, DownloadFile, FileDigest, FileEntry	12✔
29	from pants.engine.internals.native_engine import RemovePrefix	12✔
30	from pants.engine.internals.selectors import concurrently	12✔
31	from pants.engine.intrinsics import create_digest, get_digest_entries, remove_prefix	12✔
32	from pants.engine.platform import Platform	12✔
33	from pants.engine.rules import collect_rules, implicitly, rule	12✔
34	from pants.engine.unions import UnionMembership, UnionRule	12✔
35	from pants.option.option_types import DictOption, EnumOption, StrListOption, StrOption	12✔
36	from pants.option.subsystem import Subsystem, _construct_subsystem	12✔
37	from pants.util.docutil import doc_url	12✔
38	from pants.util.logging import LogLevel	12✔
39	from pants.util.meta import classproperty	12✔
40	from pants.util.strutil import softwrap	12✔
41
42	logger = logging.getLogger(__name__)	12✔
43
44
45	class UnknownVersion(Exception):	12✔
46	pass	12✔
47
48
49	class ExternalToolError(Exception):	12✔
50	pass	12✔
51
52
53	class UnsupportedVersion(ExternalToolError):	12✔
54	"""The specified version of the tool is not supported, according to the given version
55	constraints."""
56
57
58	class UnsupportedVersionUsage(Enum):	12✔
59	"""What action to take in case the requested version of the tool is not supported."""
60
61	RaiseError = "error"	12✔
62	LogWarning = "warning"	12✔
63
64
65	@dataclass(frozen=True)	12✔
66	class ExternalToolRequest:	12✔
67	download_file_request: DownloadFile	12✔
68	exe: str	12✔
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	12✔
73
74
75	@dataclass(frozen=True)	12✔
76	class DownloadedExternalTool:	12✔
77	digest: Digest	12✔
78	exe: str	12✔
79
80
81	@dataclass(frozen=True)	12✔
82	class ExternalToolVersion:	12✔
83	version: str	12✔
84	platform: str	12✔
85	sha256: str	12✔
86	filesize: int	12✔
87	url_override: str \| None = None	12✔
88
89	def encode(self) -> str:	12✔
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	12✔
96	def decode(cls, version_str: str) -> ExternalToolVersion:	12✔
97	parts = [x.strip() for x in version_str.split("\|")]	12✔
98	version, platform, sha256, filesize = parts[:4]	12✔
99	url_override = parts[4] if len(parts) > 4 else None	12✔
100	return cls(version, platform, sha256, int(filesize), url_override=url_override)	12✔
101
102
103	class ExternalToolOptionsMixin:	12✔
104	"""Common options for implementing subsystem providing an `ExternalToolRequest`."""
105
106	@classproperty	12✔
107	def name(cls):	12✔
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()	12✔
113
114	@classproperty	12✔
115	def binary_name(cls):	12✔
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()	12✔
123
124	# The default values for --version and --known-versions, and the supported versions.
125	# Subclasses must set appropriately.
126	default_version: str	12✔
127	default_known_versions: list[str]	12✔
128	version_constraints: str \| None = None	12✔
129
130	version = StrOption(	12✔
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(	12✔
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):	12✔
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):	12✔
204	super().__init__(args, *kwargs)	12✔
205	self.check_version_constraints()	12✔
206
207	export_mode = ExportMode.binary	12✔
208
209	use_unsupported_version = EnumOption(	12✔
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	12✔
222	def generate_url(self, plat: Platform) -> str:	12✔
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:	12✔
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]}"	12✔
238
239	def known_version(self, plat: Platform) -> ExternalToolVersion \| None:	12✔
240	for known_version in self.known_versions:	12✔
241	tool_version = self.decode_known_version(known_version)	12✔
242	if plat.value == tool_version.platform and tool_version.version == self.version:	12✔
243	return tool_version	12✔
244	return None	1✔
245
246	def get_request(self, plat: Platform) -> ExternalToolRequest:	12✔
247	"""Generate a request for this tool."""
248
249	tool_version = self.known_version(plat)	12✔
250	if tool_version:	12✔
251	return self.get_request_for(	12✔
252	tool_version.platform,
253	tool_version.sha256,
254	tool_version.filesize,
255	url_override=tool_version.url_override,
256	)
257	raise UnknownVersion(	1✔
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	12✔
267	def decode_known_version(cls, known_version: str) -> ExternalToolVersion:	12✔
268	try:	12✔
269	return ExternalToolVersion.decode(known_version)	12✔
270	except ValueError:	×
271	raise ExternalToolError(	×
272	f"Bad value for [{cls.options_scope}].known_versions: {known_version}"
273	)
274
275	@classmethod	12✔
276	def split_known_version_str(cls, known_version: str) -> tuple[str, str, str, int]:	12✔
277	version = cls.decode_known_version(known_version)	×
278	return version.version, version.platform, version.sha256, version.filesize	×
279
280	def get_request_for(	12✔
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)	12✔
285	digest = FileDigest(fingerprint=sha256, serialized_bytes_length=length)	12✔
286	try:	12✔
287	url = url_override or self.generate_url(plat)	12✔
288	exe = self.generate_exe(plat)	12✔
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)	12✔
294
295	def check_version_constraints(self) -> None:	12✔
296	if not self.version_constraints:	12✔
297	return None	12✔
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}")	12✔
301	if constraints.specifier.contains(self.version):	12✔
302	# all ok
303	return None	12✔
304
305	msg = [	1✔
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
312	if self.use_unsupported_version is UnsupportedVersionUsage.LogWarning:	1✔
313	msg.extend(	1✔
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	)
320	logger.warning(" ".join(msg))	1✔
321	elif self.use_unsupported_version is UnsupportedVersionUsage.RaiseError:	1✔
322	msg.append(	1✔
323	softwrap(
324	f"""
325	Alternatively, update [{self.options_scope}].use_unsupported_version to be
326	'warning'.
327	"""
328	)
329	)
330	raise UnsupportedVersion(" ".join(msg))	1✔
331
332
333	class TemplatedExternalToolOptionsMixin(ExternalToolOptionsMixin):	12✔
334	"""Common options for implementing a subsystem providing an `ExternalToolRequest` via a URL
335	template."""
336
337	default_url_template: str	12✔
338	default_url_platform_mapping: dict[str, str] \| None = None	12✔
339
340	url_template = StrOption(	12✔
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](	12✔
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):	12✔
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:	12✔
395	platform = self.url_platform_mapping.get(plat.value, "")	12✔
396	return self.url_template.format(version=self.version, platform=platform)	12✔
397
398
399	@rule(level=LogLevel.DEBUG)	12✔
400	async def download_external_tool(request: ExternalToolRequest) -> DownloadedExternalTool:	12✔
401	# Download and extract.
402	maybe_archive_digest = await download_file(request.download_file_request, **implicitly())	12✔
403	extracted_archive = await maybe_extract_archive(**implicitly(maybe_archive_digest))	12✔
404	digest = extracted_archive.digest	12✔
405	digest_entries = await get_digest_entries(digest)	12✔
406	if request.strip_common_path_prefix:	12✔
NEW 407	paths = tuple(entry.path for entry in digest_entries)	×
NEW 408	if len(paths) == 1:	×
NEW 409	commonpath = os.path.dirname(paths[0])	×
410	else:
NEW 411	commonpath = os.path.commonpath(paths)	×
NEW 412	digest = await remove_prefix(RemovePrefix(extracted_archive.digest, commonpath))	×
413
414	# Confirm executable.
415	exe_path = request.exe.lstrip("./")	12✔
416	is_not_executable = False	12✔
417	updated_digest_entries = []	12✔
418	for entry in digest_entries:	12✔
419	if isinstance(entry, FileEntry) and entry.path == exe_path and not entry.is_executable:	12✔
420	# We should recreate the digest with the executable bit set.
421	is_not_executable = True	12✔
422	entry = dataclasses.replace(entry, is_executable=True)	12✔
423	updated_digest_entries.append(entry)	12✔
424	if is_not_executable:	12✔
425	digest = await create_digest(CreateDigest(updated_digest_entries))	12✔
426
427	return DownloadedExternalTool(digest, request.exe)	12✔
428
429
430	@dataclass(frozen=True)	12✔
431	class ExportExternalToolRequest(ExportRequest):	12✔
432	pass	12✔
433	# tool: type[ExternalTool]
434
435
436	@dataclass(frozen=True)	12✔
437	class _ExportExternalToolForResolveRequest(EngineAwareParameter):	12✔
438	resolve: str	12✔
439
440
441	@dataclass(frozen=True)	12✔
442	class MaybeExportResult:	12✔
443	result: ExportResult \| None	12✔
444
445
446	@rule(level=LogLevel.DEBUG)	12✔
447	async def export_external_tool(	12✔
448	req: _ExportExternalToolForResolveRequest, platform: Platform, union_membership: UnionMembership
449	) -> MaybeExportResult:
450	"""Export a downloadable tool. Downloads all the tools to `bins`, and symlinks the primary exe
451	to the `bin` directory.
452
453	We use the last segment of the exe instead of the resolve because:
454	- it's probably the exe name people expect
455	- avoids clutter from the resolve name (ex "tfsec" instead of "terraform-tfsec")
456	"""
457	exportables = ExportableTool.filter_for_subclasses(	1✔
458	union_membership,
459	ExternalTool, # type:ignore[type-abstract] # ExternalTool is abstract, and mypy doesn't like that we might return it
460	)
461	maybe_exportable = exportables.get(req.resolve)	1✔
462	if not maybe_exportable:	1✔
UNCOV 463	return MaybeExportResult(None)	×
464
465	tool = await _construct_subsystem(maybe_exportable)	1✔
466	downloaded_tool = await download_external_tool(tool.get_request(platform))	1✔
467
468	dest = os.path.join("bins", tool.name)	1✔
469
470	exe = tool.generate_exe(platform)	1✔
471	return MaybeExportResult(	1✔
472	ExportResult(
473	description=f"Export tool {req.resolve}",
474	reldir=dest,
475	digest=downloaded_tool.digest,
476	resolve=req.resolve,
477	exported_binaries=(ExportedBinary(name=tool.binary_name, path_in_export=exe),),
478	)
479	)
480
481
482	@rule	12✔
483	async def export_external_tools(	12✔
484	request: ExportExternalToolRequest, export: ExportSubsystem
485	) -> ExportResults:
UNCOV 486	maybe_tools = await concurrently(	×
487	export_external_tool(_ExportExternalToolForResolveRequest(resolve), **implicitly())
488	for resolve in export.binaries
489	)
490	return ExportResults(tool.result for tool in maybe_tools if tool.result is not None)	×
491
492
493	def rules():	12✔
494	return (	12✔
495	*collect_rules(),
496	*archive.rules(),
497	UnionRule(ExportRequest, ExportExternalToolRequest),
498	)

pantsbuild / pants / 23415513592

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