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

50.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	1✔
4
5	import logging	1✔
6	import os	1✔
7	from collections.abc import Iterable	1✔
8	from dataclasses import dataclass	1✔
9	from pathlib import Path	1✔
10	from typing import Any, ClassVar	1✔
11
12	from pants.base.build_environment import get_buildroot	1✔
13	from pants.core.environments.target_types import EnvironmentTarget	1✔
14	from pants.engine.collection import DeduplicatedCollection	1✔
15	from pants.engine.env_vars import EnvironmentVars	1✔
16	from pants.engine.rules import Rule, _uncacheable_rule, collect_rules, rule	1✔
17	from pants.option.option_types import StrListOption	1✔
18	from pants.util.logging import LogLevel	1✔
19	from pants.util.memo import memoized_property	1✔
20	from pants.util.ordered_set import FrozenOrderedSet, OrderedSet	1✔
21	from pants.util.strutil import help_text, softwrap	1✔
22
23	_logger = logging.getLogger(__name__)	1✔
24
25
26	@dataclass(frozen=True)	1✔
27	class ValidateSearchPathsRequest:	1✔
28	env_tgt: EnvironmentTarget	1✔
29	search_paths: tuple[str, ...]	1✔
30	option_origin: str	1✔
31	environment_key: str	1✔
32	is_default: bool	1✔
33	local_only: FrozenOrderedSet[str]	1✔
34
35
36	@dataclass(frozen=True)	1✔
37	class VersionManagerSearchPathsRequest:	1✔
38	env_tgt: EnvironmentTarget	1✔
39	root_dir: str \| None	1✔
40	tool_path: str	1✔
41	option: str	1✔
42	version_files: tuple[str, ...] = tuple()	1✔
43	local_token: str \| None = None	1✔
44
45
46	class VersionManagerSearchPaths(DeduplicatedCollection[str]):	1✔
47	pass	1✔
48
49
50	@_uncacheable_rule	1✔
51	async def get_un_cachable_version_manager_paths(	1✔
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:	×
56	return VersionManagerSearchPaths()	×
57
58	manager_root_dir = request.root_dir	×
59	if not manager_root_dir:	×
60	return VersionManagerSearchPaths()	×
61	root_path = Path(manager_root_dir)	×
62	if not root_path.exists():	×
63	return VersionManagerSearchPaths()	×
64
65	tool_versions_path = root_path / request.tool_path	×
66	if not tool_versions_path.is_dir():	×
67	return VersionManagerSearchPaths()	×
68
69	if request.local_token and request.version_files:	×
70	local_version_files = [Path(get_buildroot(), file) for file in request.version_files]	×
71	first_version_file = next((file for file in local_version_files if file.exists()), None)	×
72	if not first_version_file:	×
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(	×
88	f"Reading {first_version_file} to determine desired version for {request.option}."
89	)
90	local_version = first_version_file.read_text().strip()	×
91	path = Path(tool_versions_path, local_version, "bin")	×
92	if path.is_dir():	×
93	return VersionManagerSearchPaths([str(path)])	×
94	return VersionManagerSearchPaths()	×
95
96	versions_in_dir = (	×
97	tool_versions_path / version / "bin" for version in sorted(tool_versions_path.iterdir())
98	)
99	return VersionManagerSearchPaths(	×
100	str(version) for version in versions_in_dir if version.is_dir()
101	)
102
103
104	class ValidatedSearchPaths(FrozenOrderedSet):	1✔
105	"""Search paths that are valid for the current target environment."""
106
107
108	@rule(level=LogLevel.DEBUG)	1✔
109	async def validate_search_paths(request: ValidateSearchPathsRequest) -> ValidatedSearchPaths:	1✔
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
UNCOV 122	env = request.env_tgt.val	×
UNCOV 123	search_paths = request.search_paths	×
124
UNCOV 125	if request.env_tgt.can_access_local_system_paths:	×
UNCOV 126	return ValidatedSearchPaths(search_paths)	×
UNCOV 127	assert env is not None, "Expected request.env_tgt to be defined"	×
128
UNCOV 129	if request.is_default:	×
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
UNCOV 135	any_not_allowed = set(search_paths) & request.local_only	×
UNCOV 136	if any_not_allowed:	×
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
UNCOV 148	return ValidatedSearchPaths(search_paths)	×
149
150
151	class ExecutableSearchPathsOptionMixin:	1✔
152	env_vars_used_by_options: ClassVar[tuple[str, ...]] = ("PATH",)	1✔
153
154	def __init_subclass__(cls, **kwargs: Any) -> None:	1✔
UNCOV 155	if "PATH" not in cls.env_vars_used_by_options:	×
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	1✔
167	_options_env: EnvironmentVars	1✔
168
169	_executable_search_paths = StrListOption(	1✔
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	1✔
182	def executable_search_path(self) -> tuple[str, ...]:	1✔
183	def iter_path_entries():	×
184	for entry in self._executable_search_paths:	×
185	if entry == "<PATH>":	×
186	path = self._options_env.get("PATH")	×
187	if path:	×
188	yield from path.split(os.pathsep)	×
189	else:
190	yield entry	×
191
192	return tuple(OrderedSet(iter_path_entries()))	×
193
194
195	def rules() -> Iterable[Rule]:	1✔
UNCOV 196	return collect_rules()	×

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