18252174847

Committed 05 Oct 2025 01:36AM UTC coverage: 43.382% (-36.9%) from 80.261%

Build # 18252174847

Build Type

push

github

Committed by

web-flow

Commit Message

run tests on mac arm (#22717)

Just doing the minimal to pull forward the x86_64 pattern.

ref #20993

Run Details

25776 of 59416 relevant lines covered (43.38%)

1.3 hits per line

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

35.34

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

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

import logging
import re
from collections.abc import Collection
from dataclasses import dataclass
from enum import Enum
from pathlib import Path, PurePath

from pants.base.build_environment import get_buildroot
from pants.base.build_root import BuildRoot
from pants.core.environments.target_types import EnvironmentTarget
from pants.core.util_rules.env_vars import environment_vars_subset
from pants.engine.env_vars import EnvironmentVars, EnvironmentVarsRequest
from pants.engine.rules import _uncacheable_rule, collect_rules, implicitly
from pants.util.strutil import softwrap

logger = logging.getLogger(__name__)


class AsdfPathString(str, Enum):
    STANDARD = "<ASDF>"
    LOCAL = "<ASDF_LOCAL>"

    @staticmethod
    def contains_strings(search_paths: Collection[str]) -> tuple[bool, bool]:
        return AsdfPathString.STANDARD in search_paths, AsdfPathString.LOCAL in search_paths

    def description(self, tool: str) -> str:
        if self is self.STANDARD:
            return softwrap(
                f"""
                all {tool} versions currently configured by ASDF `(asdf shell, ${{HOME}}/.tool-versions)`,
                with a fallback to all installed versions
                """
            )
        if self is self.LOCAL:
            return f"the ASDF {tool} with the version in `BUILD_ROOT/.tool-versions`"
        raise NotImplementedError(f"{self} has no description.")


@dataclass(frozen=True)
class AsdfToolPathsRequest:
    env_tgt: EnvironmentTarget
    tool_name: str
    tool_description: str
    resolve_standard: bool
    resolve_local: bool
    paths_option_name: str
    bin_relpath: str = "bin"


@dataclass(frozen=True)
class AsdfToolPathsResult:
    tool_name: str
    standard_tool_paths: tuple[str, ...] = ()
    local_tool_paths: tuple[str, ...] = ()

    @classmethod
    async def get_un_cachable_search_paths(
        cls,
        search_paths: Collection[str],
        env_tgt: EnvironmentTarget,
        tool_name: str,
        tool_description: str,
        paths_option_name: str,
        bin_relpath: str = "bin",
    ) -> AsdfToolPathsResult:
        resolve_standard, resolve_local = AsdfPathString.contains_strings(search_paths)

        if resolve_standard or resolve_local:
            # AsdfToolPathsResult is not cacheable, so only request it if absolutely necessary.
            return await resolve_asdf_tool_paths(
                AsdfToolPathsRequest(
                    env_tgt=env_tgt,
                    tool_name=tool_name,
                    tool_description=tool_description,
                    resolve_standard=resolve_standard,
                    resolve_local=resolve_local,
                    paths_option_name=paths_option_name,
                    bin_relpath=bin_relpath,
                ),
                **implicitly(),
            )
        return AsdfToolPathsResult(tool_name)


async def _resolve_asdf_tool_paths(
    env_tgt: EnvironmentTarget,
    tool_name: str,
    paths_option_name: str,
    tool_description: str,
    tool_env_name: str,
    bin_relpath: str,
    env: EnvironmentVars,
    local: bool,
) -> tuple[str, ...]:
    if not env_tgt.can_access_local_system_paths:
        return ()

    asdf_dir = get_asdf_data_dir(env)
    if not asdf_dir:
        return ()

    asdf_dir = Path(asdf_dir)

    # Ignore ASDF if the tool's plugin isn't installed.
    asdf_tool_plugin = asdf_dir / "plugins" / tool_name
    if not asdf_tool_plugin.exists():
        return ()

    # Ignore ASDF if no versions of the tool have ever been installed (the installs folder is
    # missing).
    asdf_installs_dir = asdf_dir / "installs" / tool_name
    if not asdf_installs_dir.exists():
        return ()

    # Find all installed versions.
    asdf_installed_paths: list[str] = []
    for child in asdf_installs_dir.iterdir():
        # Aliases, and non-cpython installs (for Python) may have odd names.
        # Make sure that the entry is a subdirectory of the installs directory.
        if child.is_dir():
            # Make sure that the subdirectory has a bin directory.
            bin_dir = child.joinpath(bin_relpath)
            if bin_dir.exists():
                asdf_installed_paths.append(str(bin_dir))

    # Ignore ASDF if there are no installed versions.
    if not asdf_installed_paths:
        return ()

    asdf_paths: list[str] = []
    asdf_versions: dict[str, str] = {}
    tool_versions_file = None

    # Support "shell" based ASDF configuration
    tool_env_version = env.get(tool_env_name)
    if tool_env_version:
        asdf_versions.update([(v, tool_env_name) for v in re.split(r"\s+", tool_env_version)])

    # Target the local .tool-versions file.
    if local:
        tool_versions_file = Path(get_buildroot(), ".tool-versions")
        if not tool_versions_file.exists():
            logger.warning(
                softwrap(
                    f"""
                    No `.tool-versions` file found in the build root, but <ASDF_LOCAL> was set in
                    `{paths_option_name}`.
                    """
                )
            )
            tool_versions_file = None
    # Target the home directory tool-versions file.
    else:
        home = env.get("HOME")
        if home:
            tool_versions_file = Path(home) / ".tool-versions"
            if not tool_versions_file.exists():
                tool_versions_file = None

    if tool_versions_file:
        # Parse the tool-versions file.
        # A tool-versions file contains multiple lines, one or more per tool.
        # Standardize that the last line for each tool wins.
        #
        # The definition of a tool-versions file can be found here:
        # https://asdf-vm.com/#/core-configuration?id=tool-versions
        tool_versions_lines = tool_versions_file.read_text().splitlines()
        last_line_fields = None
        for line in tool_versions_lines:
            fields = re.split(r"\s+", line.strip())
            if not fields or fields[0] != tool_name:
                continue
            last_line_fields = fields
        if last_line_fields:
            for v in last_line_fields[1:]:
                if ":" in v:
                    key, _, value = v.partition(":")
                    if key.lower() == "path":
                        asdf_paths.append(value)
                    elif key.lower() == "ref":
                        asdf_versions[value] = str(tool_versions_file)
                    else:
                        logger.warning(
                            softwrap(
                                f"""
                                Unknown version format `{v}` from ASDF configured by
                                `{paths_option_name}`, ignoring. This
                                version will not be considered when determining which {tool_description}
                                to use. Please check that `{tool_versions_file}`
                                is accurate.
                            """
                            )
                        )
                elif v == "system":
                    logger.warning(
                        softwrap(
                            f"""
                            System path set by ASDF configured by `{paths_option_name}` is unsupported, ignoring.
                            This version will not be considered when determining which {tool_description} to use.
                            Please remove 'system' from `{tool_versions_file}` to disable this warning.
                            """
                        )
                    )
                else:
                    asdf_versions[v] = str(tool_versions_file)

    for version, source in asdf_versions.items():
        install_dir = asdf_installs_dir / version / bin_relpath
        if install_dir.exists():
            asdf_paths.append(str(install_dir))
        else:
            logger.warning(
                softwrap(
                    f"""
                    Trying to use ASDF version `{version}` configured by
                    `{paths_option_name}` but `{install_dir}` does not
                    exist. This version will not be considered when determining which {tool_description}
                    to use. Please check that `{source}` is accurate.
                    """
                )
            )

    # For non-local, if no paths have been defined, fallback to every version installed
    if not local and len(asdf_paths) == 0:
        # This could be appended to asdf_paths, but there isn't any reason to
        return tuple(asdf_installed_paths)
    else:
        return tuple(asdf_paths)


# TODO: This rule is marked uncacheable because it directly accesses the filesystem to examine ASDF configuration.
# See https://github.com/pantsbuild/pants/issues/10842 for potential future support for capturing from absolute
# paths that could allow this rule to be cached.
@_uncacheable_rule
async def resolve_asdf_tool_paths(
    request: AsdfToolPathsRequest, build_root: BuildRoot
) -> AsdfToolPathsResult:
    tool_env_name = f"ASDF_{request.tool_name.upper()}_VERSION"
    env_vars_to_request = [
        "ASDF_DIR",
        "ASDF_DATA_DIR",
        tool_env_name,
        "HOME",
    ]
    env = await environment_vars_subset(EnvironmentVarsRequest(env_vars_to_request), **implicitly())

    standard_tool_paths: tuple[str, ...] = ()
    if request.resolve_standard:
        standard_tool_paths = await _resolve_asdf_tool_paths(
            env_tgt=request.env_tgt,
            tool_name=request.tool_name,
            paths_option_name=request.paths_option_name,
            tool_description=request.tool_description,
            tool_env_name=tool_env_name,
            bin_relpath=request.bin_relpath,
            env=env,
            local=False,
        )

    local_tool_paths: tuple[str, ...] = ()
    if request.resolve_local:
        local_tool_paths = await _resolve_asdf_tool_paths(
            env_tgt=request.env_tgt,
            tool_name=request.tool_name,
            paths_option_name=request.paths_option_name,
            tool_description=request.tool_description,
            tool_env_name=tool_env_name,
            bin_relpath=request.bin_relpath,
            env=env,
            local=True,
        )

    return AsdfToolPathsResult(
        tool_name=request.tool_name,
        standard_tool_paths=standard_tool_paths,
        local_tool_paths=local_tool_paths,
    )


def get_asdf_data_dir(env: EnvironmentVars) -> PurePath | None:
    """Returns the location of asdf's installed tool versions.

    See https://asdf-vm.com/manage/configuration.html#environment-variables.

    `ASDF_DATA_DIR` is an environment variable that can be set to override the directory
    in which the plugins, installs, and shims are installed.

    `ASDF_DIR` is another environment variable that can be set, but we ignore it since
    that location only specifies where the asdf tool itself is installed, not the managed versions.

    Per the documentation, if `ASDF_DATA_DIR` is not specified, the tool will fall back to
    `$HOME/.asdf`, so we do that as well.

    :param env: The environment to use to look up asdf.
    :return: Path to the data directory, or None if it couldn't be found in the environment.
    """
    asdf_data_dir = env.get("ASDF_DATA_DIR")
    if not asdf_data_dir:
        home = env.get("HOME")
        if home:
            return PurePath(home) / ".asdf"
    return PurePath(asdf_data_dir) if asdf_data_dir else None


def rules():
    return collect_rules()

1	# Copyright 2022 Pants project contributors (see CONTRIBUTORS.md).
2	# Licensed under the Apache License, Version 2.0 (see LICENSE).
3	from __future__ import annotations	3✔
4
5	import logging	3✔
6	import re	3✔
7	from collections.abc import Collection	3✔
8	from dataclasses import dataclass	3✔
9	from enum import Enum	3✔
10	from pathlib import Path, PurePath	3✔
11
12	from pants.base.build_environment import get_buildroot	3✔
13	from pants.base.build_root import BuildRoot	3✔
14	from pants.core.environments.target_types import EnvironmentTarget	3✔
15	from pants.core.util_rules.env_vars import environment_vars_subset	3✔
16	from pants.engine.env_vars import EnvironmentVars, EnvironmentVarsRequest	3✔
17	from pants.engine.rules import _uncacheable_rule, collect_rules, implicitly	3✔
18	from pants.util.strutil import softwrap	3✔
19
20	logger = logging.getLogger(__name__)	3✔
21
22
23	class AsdfPathString(str, Enum):	3✔
24	STANDARD = "<ASDF>"	3✔
25	LOCAL = "<ASDF_LOCAL>"	3✔
26
27	@staticmethod	3✔
28	def contains_strings(search_paths: Collection[str]) -> tuple[bool, bool]:	3✔
29	return AsdfPathString.STANDARD in search_paths, AsdfPathString.LOCAL in search_paths	×
30
31	def description(self, tool: str) -> str:	3✔
32	if self is self.STANDARD:	3✔
33	return softwrap(	3✔
34	f"""
35	all {tool} versions currently configured by ASDF `(asdf shell, ${{HOME}}/.tool-versions)`,
36	with a fallback to all installed versions
37	"""
38	)
39	if self is self.LOCAL:	3✔
40	return f"the ASDF {tool} with the version in `BUILD_ROOT/.tool-versions`"	3✔
41	raise NotImplementedError(f"{self} has no description.")	×
42
43
44	@dataclass(frozen=True)	3✔
45	class AsdfToolPathsRequest:	3✔
46	env_tgt: EnvironmentTarget	3✔
47	tool_name: str	3✔
48	tool_description: str	3✔
49	resolve_standard: bool	3✔
50	resolve_local: bool	3✔
51	paths_option_name: str	3✔
52	bin_relpath: str = "bin"	3✔
53
54
55	@dataclass(frozen=True)	3✔
56	class AsdfToolPathsResult:	3✔
57	tool_name: str	3✔
58	standard_tool_paths: tuple[str, ...] = ()	3✔
59	local_tool_paths: tuple[str, ...] = ()	3✔
60
61	@classmethod	3✔
62	async def get_un_cachable_search_paths(	3✔
63	cls,
64	search_paths: Collection[str],
65	env_tgt: EnvironmentTarget,
66	tool_name: str,
67	tool_description: str,
68	paths_option_name: str,
69	bin_relpath: str = "bin",
70	) -> AsdfToolPathsResult:
71	resolve_standard, resolve_local = AsdfPathString.contains_strings(search_paths)	×
72
73	if resolve_standard or resolve_local:	×
74	# AsdfToolPathsResult is not cacheable, so only request it if absolutely necessary.
75	return await resolve_asdf_tool_paths(	×
76	AsdfToolPathsRequest(
77	env_tgt=env_tgt,
78	tool_name=tool_name,
79	tool_description=tool_description,
80	resolve_standard=resolve_standard,
81	resolve_local=resolve_local,
82	paths_option_name=paths_option_name,
83	bin_relpath=bin_relpath,
84	),
85	**implicitly(),
86	)
87	return AsdfToolPathsResult(tool_name)	×
88
89
90	async def _resolve_asdf_tool_paths(	3✔
91	env_tgt: EnvironmentTarget,
92	tool_name: str,
93	paths_option_name: str,
94	tool_description: str,
95	tool_env_name: str,
96	bin_relpath: str,
97	env: EnvironmentVars,
98	local: bool,
99	) -> tuple[str, ...]:
100	if not env_tgt.can_access_local_system_paths:	×
101	return ()	×
102
103	asdf_dir = get_asdf_data_dir(env)	×
104	if not asdf_dir:	×
105	return ()	×
106
107	asdf_dir = Path(asdf_dir)	×
108
109	# Ignore ASDF if the tool's plugin isn't installed.
110	asdf_tool_plugin = asdf_dir / "plugins" / tool_name	×
111	if not asdf_tool_plugin.exists():	×
112	return ()	×
113
114	# Ignore ASDF if no versions of the tool have ever been installed (the installs folder is
115	# missing).
116	asdf_installs_dir = asdf_dir / "installs" / tool_name	×
117	if not asdf_installs_dir.exists():	×
118	return ()	×
119
120	# Find all installed versions.
121	asdf_installed_paths: list[str] = []	×
122	for child in asdf_installs_dir.iterdir():	×
123	# Aliases, and non-cpython installs (for Python) may have odd names.
124	# Make sure that the entry is a subdirectory of the installs directory.
125	if child.is_dir():	×
126	# Make sure that the subdirectory has a bin directory.
127	bin_dir = child.joinpath(bin_relpath)	×
128	if bin_dir.exists():	×
129	asdf_installed_paths.append(str(bin_dir))	×
130
131	# Ignore ASDF if there are no installed versions.
132	if not asdf_installed_paths:	×
133	return ()	×
134
135	asdf_paths: list[str] = []	×
136	asdf_versions: dict[str, str] = {}	×
137	tool_versions_file = None	×
138
139	# Support "shell" based ASDF configuration
140	tool_env_version = env.get(tool_env_name)	×
141	if tool_env_version:	×
142	asdf_versions.update([(v, tool_env_name) for v in re.split(r"\s+", tool_env_version)])	×
143
144	# Target the local .tool-versions file.
145	if local:	×
146	tool_versions_file = Path(get_buildroot(), ".tool-versions")	×
147	if not tool_versions_file.exists():	×
148	logger.warning(	×
149	softwrap(
150	f"""
151	No `.tool-versions` file found in the build root, but <ASDF_LOCAL> was set in
152	`{paths_option_name}`.
153	"""
154	)
155	)
156	tool_versions_file = None	×
157	# Target the home directory tool-versions file.
158	else:
159	home = env.get("HOME")	×
160	if home:	×
161	tool_versions_file = Path(home) / ".tool-versions"	×
162	if not tool_versions_file.exists():	×
163	tool_versions_file = None	×
164
165	if tool_versions_file:	×
166	# Parse the tool-versions file.
167	# A tool-versions file contains multiple lines, one or more per tool.
168	# Standardize that the last line for each tool wins.
169	#
170	# The definition of a tool-versions file can be found here:
171	# https://asdf-vm.com/#/core-configuration?id=tool-versions
172	tool_versions_lines = tool_versions_file.read_text().splitlines()	×
173	last_line_fields = None	×
174	for line in tool_versions_lines:	×
175	fields = re.split(r"\s+", line.strip())	×
176	if not fields or fields[0] != tool_name:	×
177	continue	×
178	last_line_fields = fields	×
179	if last_line_fields:	×
180	for v in last_line_fields[1:]:	×
181	if ":" in v:	×
182	key, _, value = v.partition(":")	×
183	if key.lower() == "path":	×
184	asdf_paths.append(value)	×
185	elif key.lower() == "ref":	×
186	asdf_versions[value] = str(tool_versions_file)	×
187	else:
188	logger.warning(	×
189	softwrap(
190	f"""
191	Unknown version format `{v}` from ASDF configured by
192	`{paths_option_name}`, ignoring. This
193	version will not be considered when determining which {tool_description}
194	to use. Please check that `{tool_versions_file}`
195	is accurate.
196	"""
197	)
198	)
199	elif v == "system":	×
200	logger.warning(	×
201	softwrap(
202	f"""
203	System path set by ASDF configured by `{paths_option_name}` is unsupported, ignoring.
204	This version will not be considered when determining which {tool_description} to use.
205	Please remove 'system' from `{tool_versions_file}` to disable this warning.
206	"""
207	)
208	)
209	else:
210	asdf_versions[v] = str(tool_versions_file)	×
211
212	for version, source in asdf_versions.items():	×
213	install_dir = asdf_installs_dir / version / bin_relpath	×
214	if install_dir.exists():	×
215	asdf_paths.append(str(install_dir))	×
216	else:
217	logger.warning(	×
218	softwrap(
219	f"""
220	Trying to use ASDF version `{version}` configured by
221	`{paths_option_name}` but `{install_dir}` does not
222	exist. This version will not be considered when determining which {tool_description}
223	to use. Please check that `{source}` is accurate.
224	"""
225	)
226	)
227
228	# For non-local, if no paths have been defined, fallback to every version installed
229	if not local and len(asdf_paths) == 0:	×
230	# This could be appended to asdf_paths, but there isn't any reason to
231	return tuple(asdf_installed_paths)	×
232	else:
233	return tuple(asdf_paths)	×
234
235
236	# TODO: This rule is marked uncacheable because it directly accesses the filesystem to examine ASDF configuration.
237	# See https://github.com/pantsbuild/pants/issues/10842 for potential future support for capturing from absolute
238	# paths that could allow this rule to be cached.
239	@_uncacheable_rule	3✔
240	async def resolve_asdf_tool_paths(	3✔
241	request: AsdfToolPathsRequest, build_root: BuildRoot
242	) -> AsdfToolPathsResult:
243	tool_env_name = f"ASDF_{request.tool_name.upper()}_VERSION"	×
244	env_vars_to_request = [	×
245	"ASDF_DIR",
246	"ASDF_DATA_DIR",
247	tool_env_name,
248	"HOME",
249	]
250	env = await environment_vars_subset(EnvironmentVarsRequest(env_vars_to_request), **implicitly())	×
251
252	standard_tool_paths: tuple[str, ...] = ()	×
253	if request.resolve_standard:	×
254	standard_tool_paths = await _resolve_asdf_tool_paths(	×
255	env_tgt=request.env_tgt,
256	tool_name=request.tool_name,
257	paths_option_name=request.paths_option_name,
258	tool_description=request.tool_description,
259	tool_env_name=tool_env_name,
260	bin_relpath=request.bin_relpath,
261	env=env,
262	local=False,
263	)
264
265	local_tool_paths: tuple[str, ...] = ()	×
266	if request.resolve_local:	×
267	local_tool_paths = await _resolve_asdf_tool_paths(	×
268	env_tgt=request.env_tgt,
269	tool_name=request.tool_name,
270	paths_option_name=request.paths_option_name,
271	tool_description=request.tool_description,
272	tool_env_name=tool_env_name,
273	bin_relpath=request.bin_relpath,
274	env=env,
275	local=True,
276	)
277
278	return AsdfToolPathsResult(	×
279	tool_name=request.tool_name,
280	standard_tool_paths=standard_tool_paths,
281	local_tool_paths=local_tool_paths,
282	)
283
284
285	def get_asdf_data_dir(env: EnvironmentVars) -> PurePath \| None:	3✔
286	"""Returns the location of asdf's installed tool versions.
287
288	See https://asdf-vm.com/manage/configuration.html#environment-variables.
289
290	`ASDF_DATA_DIR` is an environment variable that can be set to override the directory
291	in which the plugins, installs, and shims are installed.
292
293	`ASDF_DIR` is another environment variable that can be set, but we ignore it since
294	that location only specifies where the asdf tool itself is installed, not the managed versions.
295
296	Per the documentation, if `ASDF_DATA_DIR` is not specified, the tool will fall back to
297	`$HOME/.asdf`, so we do that as well.
298
299	:param env: The environment to use to look up asdf.
300	:return: Path to the data directory, or None if it couldn't be found in the environment.
301	"""
302	asdf_data_dir = env.get("ASDF_DATA_DIR")	×
303	if not asdf_data_dir:	×
304	home = env.get("HOME")	×
305	if home:	×
306	return PurePath(home) / ".asdf"	×
307	return PurePath(asdf_data_dir) if asdf_data_dir else None	×
308
309
310	def rules():	3✔
311	return collect_rules()	3✔

pantsbuild / pants / 18252174847

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