19015773527

Committed 02 Nov 2025 05:33PM UTC coverage: 17.872% (-62.4%) from 80.3%

Build # 19015773527

Build Type

Pull #22816

github

Committed by

web-flow

Commit Message

Merge a12d75757 into 6c024e162

Pull Request Pull Request #22816: Update Pants internal Python to 3.14

Run Details

4 of 5 new or added lines in 3 files covered. (80.0%)

28452 existing lines in 683 files now uncovered.

9831 of 55007 relevant lines covered (17.87%)

0.18 hits per line

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

57.14

/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
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.66.0"
_PEX_BINARY_HASH = "f471dcbffecfdc7d0f9128b5ec839c60fb9e2567ab5c9506405feb5b2fb153fd"
_PEX_BINARY_SIZE = 4927719


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

pantsbuild / pants / 19015773527

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