20974506033

Committed 13 Jan 2026 10:14PM UTC coverage: 43.251% (-37.0%) from 80.269%

Build # 20974506033

Build Type

Pull #22976

github

Committed by

web-flow

Commit Message

Merge a16a40040 into c12556724

Pull Request Pull Request #22976: WIP: Add the ability to set stdin for a Process

Run Details

2 of 4 new or added lines in 2 files covered. (50.0%)

17213 existing lines in 540 files now uncovered.

26146 of 60452 relevant lines covered (43.25%)

0.86 hits per line

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

63.11

/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,
        python: PythonExecutable | PythonBuildStandaloneBinary | None = None,
    ) -> tuple[str, ...]:
        python_exe = python or self._pex_environment.bootstrap_python
        pex_relpath = (
            os.path.relpath(pex_filepath, self._working_directory)
            if self._working_directory
            else pex_filepath
        )
        return (python_exe.path, pex_relpath, *args)

    def environment_dict(self, *, python_configured: bool) -> 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,
        )
        # NB: We only set `PEX_PYTHON_PATH` if the Python interpreter has not already been
        # pre-selected by Pants. Otherwise, Pex would inadvertently try to find another interpreter
        # when running PEXes. (Creating a PEX will ignore this env var in favor of `--python-path`.)
        if not python_configured:
            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	2✔
5
6	import os	2✔
7	from collections.abc import Mapping	2✔
8	from dataclasses import dataclass	2✔
9	from pathlib import PurePath	2✔
10
11	from pants.core.subsystems.python_bootstrap import PythonBootstrap	2✔
12	from pants.core.util_rules import subprocess_environment, system_binaries	2✔
13	from pants.core.util_rules.adhoc_binaries import PythonBuildStandaloneBinary	2✔
14	from pants.core.util_rules.subprocess_environment import SubprocessEnvironmentVars	2✔
15	from pants.core.util_rules.system_binaries import BinaryPath	2✔
16	from pants.engine.engine_aware import EngineAwareReturnType	2✔
17	from pants.engine.rules import collect_rules, rule	2✔
18	from pants.option.global_options import NamedCachesDirOption	2✔
19	from pants.option.option_types import BoolOption, IntOption, StrListOption	2✔
20	from pants.option.subsystem import Subsystem	2✔
21	from pants.util.frozendict import FrozenDict	2✔
22	from pants.util.logging import LogLevel	2✔
23	from pants.util.memo import memoized_property	2✔
24	from pants.util.ordered_set import OrderedSet	2✔
25	from pants.util.strutil import softwrap	2✔
26
27
28	class PexSubsystem(Subsystem):	2✔
29	options_scope = "pex"	2✔
30	help = "How Pants uses Pex to run Python subprocesses."	2✔
31
32	class EnvironmentAware(Subsystem.EnvironmentAware):	2✔
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",)	2✔
37
38	_executable_search_paths = StrListOption(	2✔
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	2✔
53	def path(self) -> tuple[str, ...]:	2✔
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(	2✔
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(	2✔
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(	2✔
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	2✔
99	def verbosity(self) -> int:	2✔
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)	2✔
107	class PythonExecutable(BinaryPath, EngineAwareReturnType):	2✔
108	"""The BinaryPath of a Python executable for user code, along with some extras."""
109
110	append_only_caches: FrozenDict[str, str] = FrozenDict({})	2✔
111
112	def __init__(	2✔
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))	×
UNCOV 119	super().__init__(path, fingerprint)	×
UNCOV 120	self.__post_init__()	×
121
122	def __post_init__(self) -> None:	2✔
UNCOV 123	if not PurePath(self.path).is_absolute():	×
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:	2✔
139	return f"Selected {self.path} to run PEXes with."	×
140
141
142	@dataclass(frozen=True)	2✔
143	class PexEnvironment:	2✔
144	path: tuple[str, ...]	2✔
145	interpreter_search_paths: tuple[str, ...]	2✔
146	subprocess_environment_dict: FrozenDict[str, str]	2✔
147	named_caches_dir: PurePath	2✔
148	bootstrap_python: PythonBuildStandaloneBinary	2✔
149	venv_use_symlinks: bool = False	2✔
150
151	_PEX_ROOT_DIRNAME = "pex_root"	2✔
152
153	def in_sandbox(self, *, working_directory: str \| None) -> CompletePexEnvironment:	2✔
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:	2✔
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:	2✔
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)	2✔
186	async def find_pex_python(	2✔
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)	2✔
205	class CompletePexEnvironment:	2✔
206	_pex_environment: PexEnvironment	2✔
207	pex_root: PurePath	2✔
208	_working_directory: PurePath \| None	2✔
209	append_only_caches: FrozenDict[str, str]	2✔
210
211	@property	2✔
212	def interpreter_search_paths(self) -> tuple[str, ...]:	2✔
213	return self._pex_environment.interpreter_search_paths	×
214
215	def create_argv(	2✔
216	self,
217	pex_filepath: str,
218	*args: str,
219	python: PythonExecutable \| PythonBuildStandaloneBinary \| None = None,
220	) -> tuple[str, ...]:
221	python_exe = python or self._pex_environment.bootstrap_python	×
222	pex_relpath = (	×
223	os.path.relpath(pex_filepath, self._working_directory)
224	if self._working_directory
225	else pex_filepath
226	)
227	return (python_exe.path, pex_relpath, *args)	×
228
229	def environment_dict(self, *, python_configured: bool) -> Mapping[str, str]:	2✔
230	"""The environment to use for running anything with PEX.
231
232	If the Process is run with a pre-selected Python interpreter, set `python_configured=True`
233	to avoid PEX from trying to find a new interpreter.
234	"""
235	path = os.pathsep.join(self._pex_environment.path)	×
236	subprocess_env_dict = dict(self._pex_environment.subprocess_environment_dict)	×
237
238	if "PATH" in self._pex_environment.subprocess_environment_dict:	×
239	path = os.pathsep.join([path, subprocess_env_dict.pop("PATH")])	×
240
241	d = dict(	×
242	PATH=path,
243	PEX_IGNORE_RCFILES="true",
244	PEX_ROOT=(
245	os.path.relpath(self.pex_root, self._working_directory)
246	if self._working_directory
247	else str(self.pex_root)
248	),
249	**subprocess_env_dict,
250	)
251	# NB: We only set `PEX_PYTHON_PATH` if the Python interpreter has not already been
252	# pre-selected by Pants. Otherwise, Pex would inadvertently try to find another interpreter
253	# when running PEXes. (Creating a PEX will ignore this env var in favor of `--python-path`.)
254	if not python_configured:	×
255	d["PEX_PYTHON_PATH"] = os.pathsep.join(self.interpreter_search_paths)	×
256	return d	×
257
258
259	def rules():	2✔
260	return [	2✔
261	*collect_rules(),
262	*subprocess_environment.rules(),
263	*system_binaries.rules(),
264	]

pantsbuild / pants / 20974506033

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