18812500213

Committed 26 Oct 2025 03:42AM UTC coverage: 80.284% (+0.005%) from 80.279%

Build # 18812500213

Build Type

Pull #22804

github

Committed by

web-flow

Commit Message

Merge 2a56fdb46 into 4834308dc

Pull Request Pull Request #22804: test_shell_command: use correct default cache scope for a test's environment

Run Details

29 of 31 new or added lines in 2 files covered. (93.55%)

1314 existing lines in 64 files now uncovered.

77900 of 97030 relevant lines covered (80.28%)

3.35 hits per line

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

66.99

/src/python/pants/backend/python/util_rules/pex_environment.py

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

from __future__ import annotations

import os
from collections.abc import Mapping
from dataclasses import dataclass
from pathlib import PurePath

from pants.core.subsystems.python_bootstrap import PythonBootstrap
from pants.core.util_rules import subprocess_environment, system_binaries
from pants.core.util_rules.adhoc_binaries import PythonBuildStandaloneBinary
from pants.core.util_rules.subprocess_environment import SubprocessEnvironmentVars
from pants.core.util_rules.system_binaries import BinaryPath
from pants.engine.engine_aware import EngineAwareReturnType
from pants.engine.rules import collect_rules, rule
from pants.option.global_options import NamedCachesDirOption
from pants.option.option_types import BoolOption, IntOption, StrListOption
from pants.option.subsystem import Subsystem
from pants.util.frozendict import FrozenDict
from pants.util.logging import LogLevel
from pants.util.memo import memoized_property
from pants.util.ordered_set import OrderedSet
from pants.util.strutil import softwrap


class PexSubsystem(Subsystem):
    options_scope = "pex"
    help = "How Pants uses Pex to run Python subprocesses."

    class EnvironmentAware(Subsystem.EnvironmentAware):
        # TODO(#9760): We'll want to deprecate this in favor of a global option which allows for a
        #  per-process override.

        env_vars_used_by_options = ("PATH",)

        _executable_search_paths = StrListOption(
            default=["<PATH>"],
            help=softwrap(
                """
                The PATH value that will be used by the PEX subprocess and any subprocesses it
                spawns.

                The special string `"<PATH>"` will expand to the contents of the PATH env var.
                """
            ),
            advanced=True,
            metavar="<binary-paths>",
        )

        @memoized_property
        def path(self) -> tuple[str, ...]:
            def iter_path_entries():
                for entry in self._executable_search_paths:
                    if entry == "<PATH>":
                        path = self._options_env.get("PATH")
                        if path:
                            yield from path.split(os.pathsep)
                    else:
                        yield entry

            return tuple(OrderedSet(iter_path_entries()))

    _verbosity = IntOption(
        default=0,
        help="Set the verbosity level of PEX logging, from 0 (no logging) up to 9 (max logging).",
        advanced=True,
    )
    venv_use_symlinks = BoolOption(
        default=False,
        help=softwrap(
            """
            When possible, use venvs whose site-packages directories are populated with symlinks.

            Enabling this can save space in the `--named-caches-dir` directory
            and lead to slightly faster execution times for Pants Python goals. Some
            distributions do not work with symlinked venvs though, so you may not be able to
            enable this optimization as a result.
            """
        ),
        advanced=True,
    )
    emit_warnings = BoolOption(
        default=False,
        help=softwrap(
            """
            If warnings from Pex should be logged by Pants to the console.

            Note: Pants uses Pex internally in some ways that trigger some warnings at the moment,
            so enabling this may result in warnings not related to your code. See
            [#20577](https://github.com/pantsbuild/pants/issues/20577) and
            [#20586](https://github.com/pantsbuild/pants/issues/20586).
            """
        ),
    )

    @property
    def verbosity(self) -> int:
        level = self._verbosity
        if level < 0 or level > 9:
            raise ValueError("verbosity level must be between 0 and 9")
        return level


@dataclass(frozen=True)
class PythonExecutable(BinaryPath, EngineAwareReturnType):
    """The BinaryPath of a Python executable for user code, along with some extras."""

    append_only_caches: FrozenDict[str, str] = FrozenDict({})

    def __init__(
        self,
        path: str,
        fingerprint: str | None = None,
        append_only_caches: Mapping[str, str] = FrozenDict({}),
    ) -> None:
        object.__setattr__(self, "append_only_caches", FrozenDict(append_only_caches))
        super().__init__(path, fingerprint)
        self.__post_init__()

    def __post_init__(self) -> None:
        if not PurePath(self.path).is_absolute():
            raise ValueError(
                softwrap(
                    f"""
                    PythonExecutable expects the path to be absolute. Tools like Pex internalize the
                    absolute path to Python (since `sys.executable` is expected to be absolute).

                    In order to ensure the cache key for Python process is "correct" (especially in
                    remote cache/execution situations) we require the Python path is absolute.

                    Got: {self.path}
                    """
                )
            )

    def message(self) -> str:
        return f"Selected {self.path} to run PEXes with."


@dataclass(frozen=True)
class PexEnvironment:
    path: tuple[str, ...]
    interpreter_search_paths: tuple[str, ...]
    subprocess_environment_dict: FrozenDict[str, str]
    named_caches_dir: PurePath
    bootstrap_python: PythonBuildStandaloneBinary
    venv_use_symlinks: bool = False

    _PEX_ROOT_DIRNAME = "pex_root"

    def in_sandbox(self, *, working_directory: str | None) -> CompletePexEnvironment:
        pex_root = PurePath(".cache") / self._PEX_ROOT_DIRNAME
        return CompletePexEnvironment(
            _pex_environment=self,
            pex_root=pex_root,
            _working_directory=PurePath(working_directory) if working_directory else None,
            append_only_caches=FrozenDict(
                **{self._PEX_ROOT_DIRNAME: str(pex_root)},
                **self.bootstrap_python.APPEND_ONLY_CACHES,
            ),
        )

    def in_workspace(self) -> CompletePexEnvironment:
        # N.B.: When running in the workspace the engine doesn't offer an append_only_caches
        # service to setup a symlink to our named cache for us. As such, we point the PEX_ROOT
        # directly at the underlying append only cache in that case to re-use results there and
        # to keep the workspace from being dirtied by the creation of a new Pex cache rooted
        # there.
        pex_root = self.named_caches_dir / self._PEX_ROOT_DIRNAME
        return CompletePexEnvironment(
            _pex_environment=self,
            pex_root=pex_root,
            _working_directory=None,
            append_only_caches=self.bootstrap_python.APPEND_ONLY_CACHES,
        )

    def venv_site_packages_copies_option(self, use_copies: bool) -> str:
        if self.venv_use_symlinks and not use_copies:
            return "--no-venv-site-packages-copies"
        return "--venv-site-packages-copies"


@rule(desc="Prepare environment for running PEXes", level=LogLevel.DEBUG)
async def find_pex_python(
    python_bootstrap: PythonBootstrap,
    python_binary: PythonBuildStandaloneBinary,
    pex_subsystem: PexSubsystem,
    pex_environment_aware: PexSubsystem.EnvironmentAware,
    subprocess_env_vars: SubprocessEnvironmentVars,
    named_caches_dir: NamedCachesDirOption,
) -> PexEnvironment:
    return PexEnvironment(
        path=pex_environment_aware.path,
        interpreter_search_paths=python_bootstrap.interpreter_search_paths,
        subprocess_environment_dict=subprocess_env_vars.vars,
        named_caches_dir=named_caches_dir.val,
        bootstrap_python=python_binary,
        venv_use_symlinks=pex_subsystem.venv_use_symlinks,
    )


@dataclass(frozen=True)
class CompletePexEnvironment:
    _pex_environment: PexEnvironment
    pex_root: PurePath
    _working_directory: PurePath | None
    append_only_caches: FrozenDict[str, str]

    @property
    def interpreter_search_paths(self) -> tuple[str, ...]:
        return self._pex_environment.interpreter_search_paths

    def create_argv(self, pex_filepath: str, *args: str) -> tuple[str, ...]:
        pex_relpath = (
            os.path.relpath(pex_filepath, self._working_directory)
            if self._working_directory
            else pex_filepath
        )
        return (self._pex_environment.bootstrap_python.path, pex_relpath, *args)

    def environment_dict(
        self, *, python: PythonExecutable | PythonBuildStandaloneBinary | None = None
    ) -> Mapping[str, str]:
        """The environment to use for running anything with PEX.

        If the Process is run with a pre-selected Python interpreter, set `python_configured=True`
        to avoid PEX from trying to find a new interpreter.
        """
        path = os.pathsep.join(self._pex_environment.path)
        subprocess_env_dict = dict(self._pex_environment.subprocess_environment_dict)

        if "PATH" in self._pex_environment.subprocess_environment_dict:
            path = os.pathsep.join([path, subprocess_env_dict.pop("PATH")])

        d = dict(
            PATH=path,
            PEX_IGNORE_RCFILES="true",
            PEX_ROOT=(
                os.path.relpath(self.pex_root, self._working_directory)
                if self._working_directory
                else str(self.pex_root)
            ),
            **subprocess_env_dict,
        )
        if python:
            d["PEX_PYTHON"] = python.path
        else:
            d["PEX_PYTHON_PATH"] = os.pathsep.join(self.interpreter_search_paths)
        return d


def rules():
    return [
        *collect_rules(),
        *subprocess_environment.rules(),
        *system_binaries.rules(),
    ]

1	# Copyright 2019 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 os	12✔
7	from collections.abc import Mapping	12✔
8	from dataclasses import dataclass	12✔
9	from pathlib import PurePath	12✔
10
11	from pants.core.subsystems.python_bootstrap import PythonBootstrap	12✔
12	from pants.core.util_rules import subprocess_environment, system_binaries	12✔
13	from pants.core.util_rules.adhoc_binaries import PythonBuildStandaloneBinary	12✔
14	from pants.core.util_rules.subprocess_environment import SubprocessEnvironmentVars	12✔
15	from pants.core.util_rules.system_binaries import BinaryPath	12✔
16	from pants.engine.engine_aware import EngineAwareReturnType	12✔
17	from pants.engine.rules import collect_rules, rule	12✔
18	from pants.option.global_options import NamedCachesDirOption	12✔
19	from pants.option.option_types import BoolOption, IntOption, StrListOption	12✔
20	from pants.option.subsystem import Subsystem	12✔
21	from pants.util.frozendict import FrozenDict	12✔
22	from pants.util.logging import LogLevel	12✔
23	from pants.util.memo import memoized_property	12✔
24	from pants.util.ordered_set import OrderedSet	12✔
25	from pants.util.strutil import softwrap	12✔
26
27
28	class PexSubsystem(Subsystem):	12✔
29	options_scope = "pex"	12✔
30	help = "How Pants uses Pex to run Python subprocesses."	12✔
31
32	class EnvironmentAware(Subsystem.EnvironmentAware):	12✔
33	# TODO(#9760): We'll want to deprecate this in favor of a global option which allows for a
34	# per-process override.
35
36	env_vars_used_by_options = ("PATH",)	12✔
37
38	_executable_search_paths = StrListOption(	12✔
39	default=["<PATH>"],
40	help=softwrap(
41	"""
42	The PATH value that will be used by the PEX subprocess and any subprocesses it
43	spawns.
44
45	The special string `"<PATH>"` will expand to the contents of the PATH env var.
46	"""
47	),
48	advanced=True,
49	metavar="<binary-paths>",
50	)
51
52	@memoized_property	12✔
53	def path(self) -> tuple[str, ...]:	12✔
54	def iter_path_entries():	×
55	for entry in self._executable_search_paths:	×
56	if entry == "<PATH>":	×
57	path = self._options_env.get("PATH")	×
58	if path:	×
59	yield from path.split(os.pathsep)	×
60	else:
61	yield entry	×
62
63	return tuple(OrderedSet(iter_path_entries()))	×
64
65	_verbosity = IntOption(	12✔
66	default=0,
67	help="Set the verbosity level of PEX logging, from 0 (no logging) up to 9 (max logging).",
68	advanced=True,
69	)
70	venv_use_symlinks = BoolOption(	12✔
71	default=False,
72	help=softwrap(
73	"""
74	When possible, use venvs whose site-packages directories are populated with symlinks.
75
76	Enabling this can save space in the `--named-caches-dir` directory
77	and lead to slightly faster execution times for Pants Python goals. Some
78	distributions do not work with symlinked venvs though, so you may not be able to
79	enable this optimization as a result.
80	"""
81	),
82	advanced=True,
83	)
84	emit_warnings = BoolOption(	12✔
85	default=False,
86	help=softwrap(
87	"""
88	If warnings from Pex should be logged by Pants to the console.
89
90	Note: Pants uses Pex internally in some ways that trigger some warnings at the moment,
91	so enabling this may result in warnings not related to your code. See
92	[#20577](https://github.com/pantsbuild/pants/issues/20577) and
93	[#20586](https://github.com/pantsbuild/pants/issues/20586).
94	"""
95	),
96	)
97
98	@property	12✔
99	def verbosity(self) -> int:	12✔
100	level = self._verbosity	×
101	if level < 0 or level > 9:	×
102	raise ValueError("verbosity level must be between 0 and 9")	×
103	return level	×
104
105
106	@dataclass(frozen=True)	12✔
107	class PythonExecutable(BinaryPath, EngineAwareReturnType):	12✔
108	"""The BinaryPath of a Python executable for user code, along with some extras."""
109
110	append_only_caches: FrozenDict[str, str] = FrozenDict({})	12✔
111
112	def __init__(	12✔
113	self,
114	path: str,
115	fingerprint: str \| None = None,
116	append_only_caches: Mapping[str, str] = FrozenDict({}),
117	) -> None:
UNCOV 118	object.__setattr__(self, "append_only_caches", FrozenDict(append_only_caches))	1✔
UNCOV 119	super().__init__(path, fingerprint)	1✔
UNCOV 120	self.__post_init__()	1✔
121
122	def __post_init__(self) -> None:	12✔
UNCOV 123	if not PurePath(self.path).is_absolute():	1✔
124	raise ValueError(	×
125	softwrap(
126	f"""
127	PythonExecutable expects the path to be absolute. Tools like Pex internalize the
128	absolute path to Python (since `sys.executable` is expected to be absolute).
129
130	In order to ensure the cache key for Python process is "correct" (especially in
131	remote cache/execution situations) we require the Python path is absolute.
132
133	Got: {self.path}
134	"""
135	)
136	)
137
138	def message(self) -> str:	12✔
139	return f"Selected {self.path} to run PEXes with."	×
140
141
142	@dataclass(frozen=True)	12✔
143	class PexEnvironment:	12✔
144	path: tuple[str, ...]	12✔
145	interpreter_search_paths: tuple[str, ...]	12✔
146	subprocess_environment_dict: FrozenDict[str, str]	12✔
147	named_caches_dir: PurePath	12✔
148	bootstrap_python: PythonBuildStandaloneBinary	12✔
149	venv_use_symlinks: bool = False	12✔
150
151	_PEX_ROOT_DIRNAME = "pex_root"	12✔
152
153	def in_sandbox(self, *, working_directory: str \| None) -> CompletePexEnvironment:	12✔
154	pex_root = PurePath(".cache") / self._PEX_ROOT_DIRNAME	×
155	return CompletePexEnvironment(	×
156	_pex_environment=self,
157	pex_root=pex_root,
158	_working_directory=PurePath(working_directory) if working_directory else None,
159	append_only_caches=FrozenDict(
160	**{self._PEX_ROOT_DIRNAME: str(pex_root)},
161	**self.bootstrap_python.APPEND_ONLY_CACHES,
162	),
163	)
164
165	def in_workspace(self) -> CompletePexEnvironment:	12✔
166	# N.B.: When running in the workspace the engine doesn't offer an append_only_caches
167	# service to setup a symlink to our named cache for us. As such, we point the PEX_ROOT
168	# directly at the underlying append only cache in that case to re-use results there and
169	# to keep the workspace from being dirtied by the creation of a new Pex cache rooted
170	# there.
171	pex_root = self.named_caches_dir / self._PEX_ROOT_DIRNAME	×
172	return CompletePexEnvironment(	×
173	_pex_environment=self,
174	pex_root=pex_root,
175	_working_directory=None,
176	append_only_caches=self.bootstrap_python.APPEND_ONLY_CACHES,
177	)
178
179	def venv_site_packages_copies_option(self, use_copies: bool) -> str:	12✔
180	if self.venv_use_symlinks and not use_copies:	×
181	return "--no-venv-site-packages-copies"	×
182	return "--venv-site-packages-copies"	×
183
184
185	@rule(desc="Prepare environment for running PEXes", level=LogLevel.DEBUG)	12✔
186	async def find_pex_python(	12✔
187	python_bootstrap: PythonBootstrap,
188	python_binary: PythonBuildStandaloneBinary,
189	pex_subsystem: PexSubsystem,
190	pex_environment_aware: PexSubsystem.EnvironmentAware,
191	subprocess_env_vars: SubprocessEnvironmentVars,
192	named_caches_dir: NamedCachesDirOption,
193	) -> PexEnvironment:
194	return PexEnvironment(	×
195	path=pex_environment_aware.path,
196	interpreter_search_paths=python_bootstrap.interpreter_search_paths,
197	subprocess_environment_dict=subprocess_env_vars.vars,
198	named_caches_dir=named_caches_dir.val,
199	bootstrap_python=python_binary,
200	venv_use_symlinks=pex_subsystem.venv_use_symlinks,
201	)
202
203
204	@dataclass(frozen=True)	12✔
205	class CompletePexEnvironment:	12✔
206	_pex_environment: PexEnvironment	12✔
207	pex_root: PurePath	12✔
208	_working_directory: PurePath \| None	12✔
209	append_only_caches: FrozenDict[str, str]	12✔
210
211	@property	12✔
212	def interpreter_search_paths(self) -> tuple[str, ...]:	12✔
213	return self._pex_environment.interpreter_search_paths	×
214
215	def create_argv(self, pex_filepath: str, *args: str) -> tuple[str, ...]:	12✔
216	pex_relpath = (	×
217	os.path.relpath(pex_filepath, self._working_directory)
218	if self._working_directory
219	else pex_filepath
220	)
221	return (self._pex_environment.bootstrap_python.path, pex_relpath, *args)	×
222
223	def environment_dict(	12✔
224	self, *, python: PythonExecutable \| PythonBuildStandaloneBinary \| None = None
225	) -> Mapping[str, str]:
226	"""The environment to use for running anything with PEX.
227
228	If the Process is run with a pre-selected Python interpreter, set `python_configured=True`
229	to avoid PEX from trying to find a new interpreter.
230	"""
231	path = os.pathsep.join(self._pex_environment.path)	×
232	subprocess_env_dict = dict(self._pex_environment.subprocess_environment_dict)	×
233
234	if "PATH" in self._pex_environment.subprocess_environment_dict:	×
235	path = os.pathsep.join([path, subprocess_env_dict.pop("PATH")])	×
236
237	d = dict(	×
238	PATH=path,
239	PEX_IGNORE_RCFILES="true",
240	PEX_ROOT=(
241	os.path.relpath(self.pex_root, self._working_directory)
242	if self._working_directory
243	else str(self.pex_root)
244	),
245	**subprocess_env_dict,
246	)
247	if python:	×
248	d["PEX_PYTHON"] = python.path	×
249	else:
250	d["PEX_PYTHON_PATH"] = os.pathsep.join(self.interpreter_search_paths)	×
251	return d	×
252
253
254	def rules():	12✔
255	return [	12✔
256	*collect_rules(),
257	*subprocess_environment.rules(),
258	*system_binaries.rules(),
259	]

pantsbuild / pants / 18812500213

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