24144215414

Committed 08 Apr 2026 03:36PM UTC coverage: 90.427% (-2.5%) from 92.908%

Build # 24144215414

Build Type

Pull #23227

github

Committed by

web-flow

Commit Message

Merge 107cd62c9 into 9036734c9

Pull Request Pull Request #23227: Fix uv PEX builder to use pex3 lock export

Coverage Stats

85 of 86 new or added lines in 2 files covered. (98.84%)

1990 existing lines in 131 files now uncovered.

83919 of 92803 relevant lines covered (90.43%)

3.58 hits per line

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

88.0

/src/python/pants/core/util_rules/search_paths.py

# Copyright 2023 Pants project contributors (see CONTRIBUTORS.md).
# Licensed under the Apache License, Version 2.0 (see LICENSE).
from __future__ import annotations

import logging
import os
from collections.abc import Iterable
from dataclasses import dataclass
from pathlib import Path
from typing import Any, ClassVar

from pants.base.build_environment import get_buildroot
from pants.core.environments.target_types import EnvironmentTarget
from pants.engine.collection import DeduplicatedCollection
from pants.engine.env_vars import EnvironmentVars
from pants.engine.rules import Rule, _uncacheable_rule, collect_rules, rule
from pants.option.option_types import StrListOption
from pants.util.logging import LogLevel
from pants.util.memo import memoized_property
from pants.util.ordered_set import FrozenOrderedSet, OrderedSet
from pants.util.strutil import help_text, softwrap

_logger = logging.getLogger(__name__)


@dataclass(frozen=True)
class ValidateSearchPathsRequest:
    env_tgt: EnvironmentTarget
    search_paths: tuple[str, ...]
    option_origin: str
    environment_key: str
    is_default: bool
    local_only: FrozenOrderedSet[str]


@dataclass(frozen=True)
class VersionManagerSearchPathsRequest:
    env_tgt: EnvironmentTarget
    root_dir: str | None
    tool_path: str
    option: str
    version_files: tuple[str, ...] = tuple()
    local_token: str | None = None


class VersionManagerSearchPaths(DeduplicatedCollection[str]):
    pass


@_uncacheable_rule
async def get_un_cachable_version_manager_paths(
    request: VersionManagerSearchPathsRequest,
) -> VersionManagerSearchPaths:
    """Inspects the directory of a version manager tool like pyenv or nvm to find installations."""
    if not request.env_tgt.can_access_local_system_paths:
        return VersionManagerSearchPaths()

    manager_root_dir = request.root_dir
    if not manager_root_dir:
        return VersionManagerSearchPaths()
    root_path = Path(manager_root_dir)
    if not root_path.exists():
        return VersionManagerSearchPaths()

    tool_versions_path = root_path / request.tool_path
    if not tool_versions_path.is_dir():
        return VersionManagerSearchPaths()

    if request.local_token and request.version_files:
        local_version_files = [Path(get_buildroot(), file) for file in request.version_files]
        first_version_file = next((file for file in local_version_files if file.exists()), None)
        if not first_version_file:
            file_string = ", ".join(f"`{file}`" for file in local_version_files)
            no_file = (
                f"No {file_string}" if len(local_version_files) == 1 else f"None of {file_string}"
            )
            _logger.warning(
                softwrap(
                    f"""
                    {no_file} found in the build root,
                    but {request.local_token} was set in `{request.option}`.
                    """
                )
            )
            return VersionManagerSearchPaths()

        _logger.info(
            f"Reading {first_version_file} to determine desired version for {request.option}."
        )
        local_version = first_version_file.read_text().strip()
        path = Path(tool_versions_path, local_version, "bin")
        if path.is_dir():
            return VersionManagerSearchPaths([str(path)])
        return VersionManagerSearchPaths()

    versions_in_dir = (
        tool_versions_path / version / "bin" for version in sorted(tool_versions_path.iterdir())
    )
    return VersionManagerSearchPaths(
        str(version) for version in versions_in_dir if version.is_dir()
    )


class ValidatedSearchPaths(FrozenOrderedSet):
    """Search paths that are valid for the current target environment."""


@rule(level=LogLevel.DEBUG)
async def validate_search_paths(request: ValidateSearchPathsRequest) -> ValidatedSearchPaths:
    """Checks for special search path strings, and errors if any are invalid for the environment.

    This will return:
    * The search paths, unaltered, for local/undefined environments, OR
    * The search paths, with invalid tokens removed, if the provided value was unaltered from the
      default value in the options system.
    * The search paths unaltered, if the search paths are all valid tokens for this environment

    If the environment is non-local and there are invalid tokens for those environments, raise
    `ValueError`.
    """

    env = request.env_tgt.val
    search_paths = request.search_paths

    if request.env_tgt.can_access_local_system_paths:
        return ValidatedSearchPaths(search_paths)
    assert env is not None, "Expected request.env_tgt to be defined"

    if request.is_default:
        # Strip out the not-allowed special strings from search_paths.
        # An error will occur on the off chance the non-local environment expects local_only tokens,
        # but there's nothing we can do here to detect it.
        return ValidatedSearchPaths(path for path in search_paths if path not in request.local_only)

    any_not_allowed = set(search_paths) & request.local_only
    if any_not_allowed:
        env_type = type(env)
        raise ValueError(
            softwrap(
                f"`{request.option_origin}` is configured to use local discovery "
                f"tools, which do not work in {env_type.__name__} runtime environments. To fix "
                f"this, set the value of `{request.environment_key}` in the `{env.alias}` "
                f"defined at `{env.address}` to contain only hardcoded paths or the `<PATH>` "
                "special string."
            )
        )

    return ValidatedSearchPaths(search_paths)


class ExecutableSearchPathsOptionMixin:
    env_vars_used_by_options: ClassVar[tuple[str, ...]] = ("PATH",)

    def __init_subclass__(cls, **kwargs: Any) -> None:
        if "PATH" not in cls.env_vars_used_by_options:
            raise ValueError(
                softwrap(
                    f"""
                    {ExecutableSearchPathsOptionMixin.__name__} depends on the PATH environment variable.

                    Please add it to the {cls.__name__}.env_vars_used_by_options.
                    """
                )
            )

    executable_search_paths_help: str
    _options_env: EnvironmentVars

    _executable_search_paths = StrListOption(
        default=["<PATH>"],
        help=lambda cls: help_text(
            f"""
            {cls.executable_search_paths_help}
            The special string `"<PATH>"` will expand to the contents of the PATH env var.
            """
        ),
        advanced=True,
        metavar="<binary-paths>",
    )

    @memoized_property
    def executable_search_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()))


def rules() -> Iterable[Rule]:
    return collect_rules()

1	# Copyright 2023 Pants project contributors (see CONTRIBUTORS.md).
2	# Licensed under the Apache License, Version 2.0 (see LICENSE).
3	from __future__ import annotations	10✔
4
5	import logging	10✔
6	import os	10✔
7	from collections.abc import Iterable	10✔
8	from dataclasses import dataclass	10✔
9	from pathlib import Path	10✔
10	from typing import Any, ClassVar	10✔
11
12	from pants.base.build_environment import get_buildroot	10✔
13	from pants.core.environments.target_types import EnvironmentTarget	10✔
14	from pants.engine.collection import DeduplicatedCollection	10✔
15	from pants.engine.env_vars import EnvironmentVars	10✔
16	from pants.engine.rules import Rule, _uncacheable_rule, collect_rules, rule	10✔
17	from pants.option.option_types import StrListOption	10✔
18	from pants.util.logging import LogLevel	10✔
19	from pants.util.memo import memoized_property	10✔
20	from pants.util.ordered_set import FrozenOrderedSet, OrderedSet	10✔
21	from pants.util.strutil import help_text, softwrap	10✔
22
23	_logger = logging.getLogger(__name__)	10✔
24
25
26	@dataclass(frozen=True)	10✔
27	class ValidateSearchPathsRequest:	10✔
28	env_tgt: EnvironmentTarget	10✔
29	search_paths: tuple[str, ...]	10✔
30	option_origin: str	10✔
31	environment_key: str	10✔
32	is_default: bool	10✔
33	local_only: FrozenOrderedSet[str]	10✔
34
35
36	@dataclass(frozen=True)	10✔
37	class VersionManagerSearchPathsRequest:	10✔
38	env_tgt: EnvironmentTarget	10✔
39	root_dir: str \| None	10✔
40	tool_path: str	10✔
41	option: str	10✔
42	version_files: tuple[str, ...] = tuple()	10✔
43	local_token: str \| None = None	10✔
44
45
46	class VersionManagerSearchPaths(DeduplicatedCollection[str]):	10✔
47	pass	10✔
48
49
50	@_uncacheable_rule	10✔
51	async def get_un_cachable_version_manager_paths(	10✔
52	request: VersionManagerSearchPathsRequest,
53	) -> VersionManagerSearchPaths:
54	"""Inspects the directory of a version manager tool like pyenv or nvm to find installations."""
55	if not request.env_tgt.can_access_local_system_paths:	10✔
56	return VersionManagerSearchPaths()	×
57
58	manager_root_dir = request.root_dir	10✔
59	if not manager_root_dir:	10✔
60	return VersionManagerSearchPaths()	8✔
61	root_path = Path(manager_root_dir)	10✔
62	if not root_path.exists():	10✔
63	return VersionManagerSearchPaths()	10✔
64
65	tool_versions_path = root_path / request.tool_path	1✔
66	if not tool_versions_path.is_dir():	1✔
67	return VersionManagerSearchPaths()	×
68
69	if request.local_token and request.version_files:	1✔
70	local_version_files = [Path(get_buildroot(), file) for file in request.version_files]	1✔
71	first_version_file = next((file for file in local_version_files if file.exists()), None)	1✔
72	if not first_version_file:	1✔
73	file_string = ", ".join(f"`{file}`" for file in local_version_files)	×
74	no_file = (	×
75	f"No {file_string}" if len(local_version_files) == 1 else f"None of {file_string}"
76	)
77	_logger.warning(	×
78	softwrap(
79	f"""
80	{no_file} found in the build root,
81	but {request.local_token} was set in `{request.option}`.
82	"""
83	)
84	)
85	return VersionManagerSearchPaths()	×
86
87	_logger.info(	1✔
88	f"Reading {first_version_file} to determine desired version for {request.option}."
89	)
90	local_version = first_version_file.read_text().strip()	1✔
91	path = Path(tool_versions_path, local_version, "bin")	1✔
92	if path.is_dir():	1✔
93	return VersionManagerSearchPaths([str(path)])	1✔
94	return VersionManagerSearchPaths()	×
95
96	versions_in_dir = (	1✔
97	tool_versions_path / version / "bin" for version in sorted(tool_versions_path.iterdir())
98	)
99	return VersionManagerSearchPaths(	1✔
100	str(version) for version in versions_in_dir if version.is_dir()
101	)
102
103
104	class ValidatedSearchPaths(FrozenOrderedSet):	10✔
105	"""Search paths that are valid for the current target environment."""
106
107
108	@rule(level=LogLevel.DEBUG)	10✔
109	async def validate_search_paths(request: ValidateSearchPathsRequest) -> ValidatedSearchPaths:	10✔
110	"""Checks for special search path strings, and errors if any are invalid for the environment.
111
112	This will return:
113	* The search paths, unaltered, for local/undefined environments, OR
114	* The search paths, with invalid tokens removed, if the provided value was unaltered from the
115	default value in the options system.
116	* The search paths unaltered, if the search paths are all valid tokens for this environment
117
118	If the environment is non-local and there are invalid tokens for those environments, raise
119	`ValueError`.
120	"""
121
122	env = request.env_tgt.val	10✔
123	search_paths = request.search_paths	10✔
124
125	if request.env_tgt.can_access_local_system_paths:	10✔
126	return ValidatedSearchPaths(search_paths)	10✔
127	assert env is not None, "Expected request.env_tgt to be defined"	1✔
128
129	if request.is_default:	1✔
130	# Strip out the not-allowed special strings from search_paths.
131	# An error will occur on the off chance the non-local environment expects local_only tokens,
132	# but there's nothing we can do here to detect it.
UNCOV 133	return ValidatedSearchPaths(path for path in search_paths if path not in request.local_only)	×
134
135	any_not_allowed = set(search_paths) & request.local_only	1✔
136	if any_not_allowed:	1✔
UNCOV 137	env_type = type(env)	×
UNCOV 138	raise ValueError(	×
139	softwrap(
140	f"`{request.option_origin}` is configured to use local discovery "
141	f"tools, which do not work in {env_type.__name__} runtime environments. To fix "
142	f"this, set the value of `{request.environment_key}` in the `{env.alias}` "
143	f"defined at `{env.address}` to contain only hardcoded paths or the `<PATH>` "
144	"special string."
145	)
146	)
147
148	return ValidatedSearchPaths(search_paths)	1✔
149
150
151	class ExecutableSearchPathsOptionMixin:	10✔
152	env_vars_used_by_options: ClassVar[tuple[str, ...]] = ("PATH",)	10✔
153
154	def __init_subclass__(cls, **kwargs: Any) -> None:	10✔
155	if "PATH" not in cls.env_vars_used_by_options:	10✔
156	raise ValueError(	×
157	softwrap(
158	f"""
159	{ExecutableSearchPathsOptionMixin.__name__} depends on the PATH environment variable.
160
161	Please add it to the {cls.__name__}.env_vars_used_by_options.
162	"""
163	)
164	)
165
166	executable_search_paths_help: str	10✔
167	_options_env: EnvironmentVars	10✔
168
169	_executable_search_paths = StrListOption(	10✔
170	default=["<PATH>"],
171	help=lambda cls: help_text(
172	f"""
173	{cls.executable_search_paths_help}
174	The special string `"<PATH>"` will expand to the contents of the PATH env var.
175	"""
176	),
177	advanced=True,
178	metavar="<binary-paths>",
179	)
180
181	@memoized_property	10✔
182	def executable_search_path(self) -> tuple[str, ...]:	10✔
183	def iter_path_entries():	9✔
184	for entry in self._executable_search_paths:	9✔
185	if entry == "<PATH>":	9✔
186	path = self._options_env.get("PATH")	9✔
187	if path:	9✔
188	yield from path.split(os.pathsep)	9✔
189	else:
190	yield entry	×
191
192	return tuple(OrderedSet(iter_path_entries()))	9✔
193
194
195	def rules() -> Iterable[Rule]:	10✔
196	return collect_rules()	10✔

pantsbuild / pants / 24144215414

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