26380816428

Committed 25 May 2026 02:57AM UTC coverage: 52.312% (-40.6%) from 92.89%

Build # 26380816428

Build Type

Pull #23368

github

Committed by

web-flow

Commit Message

Merge 7410b48e1 into 7b1060c81

Pull Request Pull Request #23368: Run Linux ARM CI on Depot runners (Cherry-pick of #23363)

Coverage Stats

31807 of 60802 relevant lines covered (52.31%)

1.05 hits per line

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

94.5

/src/python/pants/backend/python/util_rules/pex_cli.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
from collections.abc import Iterable, Mapping
from dataclasses import dataclass

from pants.backend.python.subsystems.python_native_code import PythonNativeCodeSubsystem
from pants.backend.python.subsystems.setup import PythonSetup
from pants.backend.python.util_rules import pex_environment
from pants.backend.python.util_rules.pex_environment import (
    PexEnvironment,
    PexSubsystem,
    PythonExecutable,
)
from pants.core.goals.resolves import ExportableTool
from pants.core.util_rules import adhoc_binaries, external_tool
from pants.core.util_rules.adhoc_binaries import PythonBuildStandaloneBinary
from pants.core.util_rules.external_tool import (
    DownloadedExternalTool,
    TemplatedExternalTool,
    download_external_tool,
)
from pants.engine.fs import CreateDigest, Digest, Directory, MergeDigests
from pants.engine.internals.selectors import concurrently
from pants.engine.intrinsics import create_digest, merge_digests
from pants.engine.platform import Platform
from pants.engine.process import Process, ProcessCacheScope
from pants.engine.rules import collect_rules, rule
from pants.engine.unions import UnionRule
from pants.option.global_options import GlobalOptions, ca_certs_path_to_file_content
from pants.option.option_types import ArgsListOption
from pants.util.frozendict import FrozenDict
from pants.util.logging import LogLevel
from pants.util.strutil import softwrap

logger = logging.getLogger(__name__)


_PEX_VERSION = "v2.95.1"
_PEX_BINARY_HASH = "2cc018cf1a112ae7f94e78456f90da2d15c1a3c1327da4c73c98701ab1f8f732"
_PEX_BINARY_SIZE = 5124990


class PexCli(TemplatedExternalTool):
    options_scope = "pex-cli"
    name = "pex"
    help = "The PEX (Python EXecutable) tool (https://github.com/pex-tool/pex)."

    default_version = _PEX_VERSION
    default_url_template = "https://github.com/pex-tool/pex/releases/download/{version}/pex"
    version_constraints = ">=2.76.0,<3.0"

    # extra args to be passed to the pex tool; note that they
    # are going to apply to all invocations of the pex tool.
    global_args = ArgsListOption(
        example="--check=error --no-compile",
        extra_help=softwrap(
            """
            Note that these apply to all invocations of the pex tool, including building `pex_binary`
            targets, preparing `python_test` targets to run, and generating lockfiles.
            """
        ),
    )

    default_known_versions = [
        f"{_PEX_VERSION}|{platform}|{_PEX_BINARY_HASH}|{_PEX_BINARY_SIZE}"
        for platform in ["macos_x86_64", "macos_arm64", "linux_x86_64", "linux_arm64"]
    ]


@dataclass(frozen=True)
class PexCliProcess:
    interpreter: PythonExecutable | PythonBuildStandaloneBinary | None
    subcommand: tuple[str, ...]
    extra_args: tuple[str, ...]
    description: str = dataclasses.field(compare=False)
    additional_input_digest: Digest | None
    extra_env: FrozenDict[str, str] | None
    append_only_caches: FrozenDict[str, str] | None
    output_files: tuple[str, ...] | None
    output_directories: tuple[str, ...] | None
    level: LogLevel
    concurrency_available: int
    cache_scope: ProcessCacheScope

    def __init__(
        self,
        *,
        interpreter: PythonExecutable | PythonBuildStandaloneBinary | None = None,
        subcommand: Iterable[str],
        extra_args: Iterable[str],
        description: str,
        additional_input_digest: Digest | None = None,
        extra_env: Mapping[str, str] | None = None,
        append_only_caches: Mapping[str, str] | None = None,
        output_files: Iterable[str] | None = None,
        output_directories: Iterable[str] | None = None,
        level: LogLevel = LogLevel.INFO,
        concurrency_available: int = 0,
        cache_scope: ProcessCacheScope = ProcessCacheScope.SUCCESSFUL,
    ) -> None:
        object.__setattr__(self, "interpreter", interpreter)
        object.__setattr__(self, "subcommand", tuple(subcommand))
        object.__setattr__(self, "extra_args", tuple(extra_args))
        object.__setattr__(self, "description", description)
        object.__setattr__(self, "additional_input_digest", additional_input_digest)
        object.__setattr__(self, "extra_env", FrozenDict(extra_env) if extra_env else None)
        object.__setattr__(
            self,
            "append_only_caches",
            FrozenDict(append_only_caches) if append_only_caches else None,
        )
        object.__setattr__(self, "output_files", tuple(output_files) if output_files else None)
        object.__setattr__(
            self, "output_directories", tuple(output_directories) if output_directories else None
        )
        object.__setattr__(self, "level", level)
        object.__setattr__(self, "concurrency_available", concurrency_available)
        object.__setattr__(self, "cache_scope", cache_scope)

        self.__post_init__()

    def __post_init__(self) -> None:
        if "--pex-root-path" in self.extra_args:
            raise ValueError("`--pex-root` flag not allowed. We set its value for you.")


class PexPEX(DownloadedExternalTool):
    """The Pex PEX binary."""


@rule
async def download_pex_pex(pex_cli: PexCli, platform: Platform) -> PexPEX:
    pex_pex = await download_external_tool(pex_cli.get_request(platform))
    return PexPEX(digest=pex_pex.digest, exe=pex_pex.exe)


@rule
async def setup_pex_cli_process(
    request: PexCliProcess,
    pex_pex: PexPEX,
    pex_env: PexEnvironment,
    bootstrap_python: PythonBuildStandaloneBinary,
    python_native_code: PythonNativeCodeSubsystem.EnvironmentAware,
    global_options: GlobalOptions,
    pex_subsystem: PexSubsystem,
    pex_cli_subsystem: PexCli,
    python_setup: PythonSetup,
) -> Process:
    tmpdir = ".tmp"
    gets = [create_digest(CreateDigest([Directory(tmpdir)]))]

    cert_args = []
    if global_options.ca_certs_path:
        ca_certs_fc = ca_certs_path_to_file_content(global_options.ca_certs_path)
        gets.append(create_digest(CreateDigest((ca_certs_fc,))))
        cert_args = ["--cert", ca_certs_fc.path]

    digests_to_merge = [pex_pex.digest]
    digests_to_merge.extend(await concurrently(gets))
    if request.additional_input_digest:
        digests_to_merge.append(request.additional_input_digest)
    input_digest = await merge_digests(MergeDigests(digests_to_merge))

    global_args = [
        # Ensure Pex and its subprocesses create temporary files in the process execution
        # sandbox. It may make sense to do this generally for Processes, but in the short term we
        # have known use cases where /tmp is too small to hold large wheel downloads Pex is asked to
        # perform. Making the TMPDIR local to the sandbox allows control via
        # --local-execution-root-dir for the local case and should work well with remote cases where
        # a remoting implementation has to allow for processes producing large binaries in a
        # sandbox to support reasonable workloads. Communicating TMPDIR via --tmpdir instead of via
        # environment variable allows Pex to absolutize the path ensuring subprocesses that change
        # CWD can find the TMPDIR.
        "--tmpdir",
        tmpdir,
    ]

    if request.concurrency_available > 0:
        global_args.extend(["--jobs", "{pants_concurrency}"])

    verbosity_args = [f"-{'v' * pex_subsystem.verbosity}"] if pex_subsystem.verbosity > 0 else []

    warnings_args = [] if pex_subsystem.emit_warnings else ["--no-emit-warnings"]

    # NB: We should always pass `--python-path`, as that tells Pex where to look for interpreters
    # when `--python` isn't an absolute path.
    resolve_args = [
        *cert_args,
        "--python-path",
        os.pathsep.join(pex_env.interpreter_search_paths),
    ]
    # All old-style pex runs take the --pip-version flag, but only certain subcommands of the
    # `pex3` console script do. So if invoked with a subcommand, the caller must selectively
    # set --pip-version only on subcommands that take it.
    pip_version_args = [] if request.subcommand else ["--pip-version", python_setup.pip_version]
    args = [
        *request.subcommand,
        *global_args,
        *verbosity_args,
        *warnings_args,
        *pip_version_args,
        *resolve_args,
        *pex_cli_subsystem.global_args,
        # NB: This comes at the end because it may use `--` passthrough args, # which must come at
        # the end.
        *request.extra_args,
    ]

    complete_pex_env = pex_env.in_sandbox(working_directory=None)
    normalized_argv = complete_pex_env.create_argv(
        pex_pex.exe, *args, python=request.interpreter or bootstrap_python
    )
    env = {
        **complete_pex_env.environment_dict(python_configured=True),
        **python_native_code.subprocess_env_vars,
        **(request.extra_env or {}),
        # If a subcommand is used, we need to use the `pex3` console script.
        **({"PEX_SCRIPT": "pex3"} if request.subcommand else {}),
    }

    return Process(
        normalized_argv,
        description=request.description,
        input_digest=input_digest,
        env=env,
        output_files=request.output_files,
        output_directories=request.output_directories or tuple(),
        append_only_caches={
            **complete_pex_env.append_only_caches,
            **(request.append_only_caches or FrozenDict({})),
        },
        level=request.level,
        concurrency_available=request.concurrency_available,
        cache_scope=request.cache_scope,
    )


def maybe_log_pex_stderr(stderr: bytes, pex_verbosity: int) -> None:
    """Forward Pex's stderr to a Pants logger if conditions are met."""
    log_output = stderr.decode()
    if log_output and "PEXWarning:" in log_output:
        logger.warning("%s", log_output)
    elif log_output and pex_verbosity > 0:
        logger.info("%s", log_output)


def rules():
    return [
        *collect_rules(),
        *external_tool.rules(),
        *pex_environment.rules(),
        *adhoc_binaries.rules(),
        UnionRule(ExportableTool, PexCli),
    ]

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	from collections.abc import Iterable, Mapping	2✔
10	from dataclasses import dataclass	2✔
11
12	from pants.backend.python.subsystems.python_native_code import PythonNativeCodeSubsystem	2✔
13	from pants.backend.python.subsystems.setup import PythonSetup	2✔
14	from pants.backend.python.util_rules import pex_environment	2✔
15	from pants.backend.python.util_rules.pex_environment import (	2✔
16	PexEnvironment,
17	PexSubsystem,
18	PythonExecutable,
19	)
20	from pants.core.goals.resolves import ExportableTool	2✔
21	from pants.core.util_rules import adhoc_binaries, external_tool	2✔
22	from pants.core.util_rules.adhoc_binaries import PythonBuildStandaloneBinary	2✔
23	from pants.core.util_rules.external_tool import (	2✔
24	DownloadedExternalTool,
25	TemplatedExternalTool,
26	download_external_tool,
27	)
28	from pants.engine.fs import CreateDigest, Digest, Directory, MergeDigests	2✔
29	from pants.engine.internals.selectors import concurrently	2✔
30	from pants.engine.intrinsics import create_digest, merge_digests	2✔
31	from pants.engine.platform import Platform	2✔
32	from pants.engine.process import Process, ProcessCacheScope	2✔
33	from pants.engine.rules import collect_rules, rule	2✔
34	from pants.engine.unions import UnionRule	2✔
35	from pants.option.global_options import GlobalOptions, ca_certs_path_to_file_content	2✔
36	from pants.option.option_types import ArgsListOption	2✔
37	from pants.util.frozendict import FrozenDict	2✔
38	from pants.util.logging import LogLevel	2✔
39	from pants.util.strutil import softwrap	2✔
40
41	logger = logging.getLogger(__name__)	2✔
42
43
44	_PEX_VERSION = "v2.95.1"	2✔
45	_PEX_BINARY_HASH = "2cc018cf1a112ae7f94e78456f90da2d15c1a3c1327da4c73c98701ab1f8f732"	2✔
46	_PEX_BINARY_SIZE = 5124990	2✔
47
48
49	class PexCli(TemplatedExternalTool):	2✔
50	options_scope = "pex-cli"	2✔
51	name = "pex"	2✔
52	help = "The PEX (Python EXecutable) tool (https://github.com/pex-tool/pex)."	2✔
53
54	default_version = _PEX_VERSION	2✔
55	default_url_template = "https://github.com/pex-tool/pex/releases/download/{version}/pex"	2✔
56	version_constraints = ">=2.76.0,<3.0"	2✔
57
58	# extra args to be passed to the pex tool; note that they
59	# are going to apply to all invocations of the pex tool.
60	global_args = ArgsListOption(	2✔
61	example="--check=error --no-compile",
62	extra_help=softwrap(
63	"""
64	Note that these apply to all invocations of the pex tool, including building `pex_binary`
65	targets, preparing `python_test` targets to run, and generating lockfiles.
66	"""
67	),
68	)
69
70	default_known_versions = [	2✔
71	f"{_PEX_VERSION}\|{platform}\|{_PEX_BINARY_HASH}\|{_PEX_BINARY_SIZE}"
72	for platform in ["macos_x86_64", "macos_arm64", "linux_x86_64", "linux_arm64"]
73	]
74
75
76	@dataclass(frozen=True)	2✔
77	class PexCliProcess:	2✔
78	interpreter: PythonExecutable \| PythonBuildStandaloneBinary \| None	2✔
79	subcommand: tuple[str, ...]	2✔
80	extra_args: tuple[str, ...]	2✔
81	description: str = dataclasses.field(compare=False)	2✔
82	additional_input_digest: Digest \| None	2✔
83	extra_env: FrozenDict[str, str] \| None	2✔
84	append_only_caches: FrozenDict[str, str] \| None	2✔
85	output_files: tuple[str, ...] \| None	2✔
86	output_directories: tuple[str, ...] \| None	2✔
87	level: LogLevel	2✔
88	concurrency_available: int	2✔
89	cache_scope: ProcessCacheScope	2✔
90
91	def __init__(	2✔
92	self,
93	*,
94	interpreter: PythonExecutable \| PythonBuildStandaloneBinary \| None = None,
95	subcommand: Iterable[str],
96	extra_args: Iterable[str],
97	description: str,
98	additional_input_digest: Digest \| None = None,
99	extra_env: Mapping[str, str] \| None = None,
100	append_only_caches: Mapping[str, str] \| None = None,
101	output_files: Iterable[str] \| None = None,
102	output_directories: Iterable[str] \| None = None,
103	level: LogLevel = LogLevel.INFO,
104	concurrency_available: int = 0,
105	cache_scope: ProcessCacheScope = ProcessCacheScope.SUCCESSFUL,
106	) -> None:
107	object.__setattr__(self, "interpreter", interpreter)	2✔
108	object.__setattr__(self, "subcommand", tuple(subcommand))	2✔
109	object.__setattr__(self, "extra_args", tuple(extra_args))	2✔
110	object.__setattr__(self, "description", description)	2✔
111	object.__setattr__(self, "additional_input_digest", additional_input_digest)	2✔
112	object.__setattr__(self, "extra_env", FrozenDict(extra_env) if extra_env else None)	2✔
113	object.__setattr__(	2✔
114	self,
115	"append_only_caches",
116	FrozenDict(append_only_caches) if append_only_caches else None,
117	)
118	object.__setattr__(self, "output_files", tuple(output_files) if output_files else None)	2✔
119	object.__setattr__(	2✔
120	self, "output_directories", tuple(output_directories) if output_directories else None
121	)
122	object.__setattr__(self, "level", level)	2✔
123	object.__setattr__(self, "concurrency_available", concurrency_available)	2✔
124	object.__setattr__(self, "cache_scope", cache_scope)	2✔
125
126	self.__post_init__()	2✔
127
128	def __post_init__(self) -> None:	2✔
129	if "--pex-root-path" in self.extra_args:	2✔
130	raise ValueError("`--pex-root` flag not allowed. We set its value for you.")	×
131
132
133	class PexPEX(DownloadedExternalTool):	2✔
134	"""The Pex PEX binary."""
135
136
137	@rule	2✔
138	async def download_pex_pex(pex_cli: PexCli, platform: Platform) -> PexPEX:	2✔
139	pex_pex = await download_external_tool(pex_cli.get_request(platform))	2✔
140	return PexPEX(digest=pex_pex.digest, exe=pex_pex.exe)	2✔
141
142
143	@rule	2✔
144	async def setup_pex_cli_process(	2✔
145	request: PexCliProcess,
146	pex_pex: PexPEX,
147	pex_env: PexEnvironment,
148	bootstrap_python: PythonBuildStandaloneBinary,
149	python_native_code: PythonNativeCodeSubsystem.EnvironmentAware,
150	global_options: GlobalOptions,
151	pex_subsystem: PexSubsystem,
152	pex_cli_subsystem: PexCli,
153	python_setup: PythonSetup,
154	) -> Process:
155	tmpdir = ".tmp"	2✔
156	gets = [create_digest(CreateDigest([Directory(tmpdir)]))]	2✔
157
158	cert_args = []	2✔
159	if global_options.ca_certs_path:	2✔
160	ca_certs_fc = ca_certs_path_to_file_content(global_options.ca_certs_path)	×
161	gets.append(create_digest(CreateDigest((ca_certs_fc,))))	×
162	cert_args = ["--cert", ca_certs_fc.path]	×
163
164	digests_to_merge = [pex_pex.digest]	2✔
165	digests_to_merge.extend(await concurrently(gets))	2✔
166	if request.additional_input_digest:	2✔
167	digests_to_merge.append(request.additional_input_digest)	2✔
168	input_digest = await merge_digests(MergeDigests(digests_to_merge))	2✔
169
170	global_args = [	2✔
171	# Ensure Pex and its subprocesses create temporary files in the process execution
172	# sandbox. It may make sense to do this generally for Processes, but in the short term we
173	# have known use cases where /tmp is too small to hold large wheel downloads Pex is asked to
174	# perform. Making the TMPDIR local to the sandbox allows control via
175	# --local-execution-root-dir for the local case and should work well with remote cases where
176	# a remoting implementation has to allow for processes producing large binaries in a
177	# sandbox to support reasonable workloads. Communicating TMPDIR via --tmpdir instead of via
178	# environment variable allows Pex to absolutize the path ensuring subprocesses that change
179	# CWD can find the TMPDIR.
180	"--tmpdir",
181	tmpdir,
182	]
183
184	if request.concurrency_available > 0:	2✔
185	global_args.extend(["--jobs", "{pants_concurrency}"])	2✔
186
187	verbosity_args = [f"-{'v' * pex_subsystem.verbosity}"] if pex_subsystem.verbosity > 0 else []	2✔
188
189	warnings_args = [] if pex_subsystem.emit_warnings else ["--no-emit-warnings"]	2✔
190
191	# NB: We should always pass `--python-path`, as that tells Pex where to look for interpreters
192	# when `--python` isn't an absolute path.
193	resolve_args = [	2✔
194	*cert_args,
195	"--python-path",
196	os.pathsep.join(pex_env.interpreter_search_paths),
197	]
198	# All old-style pex runs take the --pip-version flag, but only certain subcommands of the
199	# `pex3` console script do. So if invoked with a subcommand, the caller must selectively
200	# set --pip-version only on subcommands that take it.
201	pip_version_args = [] if request.subcommand else ["--pip-version", python_setup.pip_version]	2✔
202	args = [	2✔
203	*request.subcommand,
204	*global_args,
205	*verbosity_args,
206	*warnings_args,
207	*pip_version_args,
208	*resolve_args,
209	*pex_cli_subsystem.global_args,
210	# NB: This comes at the end because it may use `--` passthrough args, # which must come at
211	# the end.
212	*request.extra_args,
213	]
214
215	complete_pex_env = pex_env.in_sandbox(working_directory=None)	2✔
216	normalized_argv = complete_pex_env.create_argv(	2✔
217	pex_pex.exe, *args, python=request.interpreter or bootstrap_python
218	)
219	env = {	2✔
220	**complete_pex_env.environment_dict(python_configured=True),
221	**python_native_code.subprocess_env_vars,
222	**(request.extra_env or {}),
223	# If a subcommand is used, we need to use the `pex3` console script.
224	**({"PEX_SCRIPT": "pex3"} if request.subcommand else {}),
225	}
226
227	return Process(	2✔
228	normalized_argv,
229	description=request.description,
230	input_digest=input_digest,
231	env=env,
232	output_files=request.output_files,
233	output_directories=request.output_directories or tuple(),
234	append_only_caches={
235	**complete_pex_env.append_only_caches,
236	**(request.append_only_caches or FrozenDict({})),
237	},
238	level=request.level,
239	concurrency_available=request.concurrency_available,
240	cache_scope=request.cache_scope,
241	)
242
243
244	def maybe_log_pex_stderr(stderr: bytes, pex_verbosity: int) -> None:	2✔
245	"""Forward Pex's stderr to a Pants logger if conditions are met."""
246	log_output = stderr.decode()	2✔
247	if log_output and "PEXWarning:" in log_output:	2✔
248	logger.warning("%s", log_output)	×
249	elif log_output and pex_verbosity > 0:	2✔
250	logger.info("%s", log_output)	×
251
252
253	def rules():	2✔
254	return [	2✔
255	*collect_rules(),
256	*external_tool.rules(),
257	*pex_environment.rules(),
258	*adhoc_binaries.rules(),
259	UnionRule(ExportableTool, PexCli),
260	]

pantsbuild / pants / 26380816428

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