20332790708

Committed 18 Dec 2025 09:48AM UTC coverage: 64.992% (-15.3%) from 80.295%

Build # 20332790708

Build Type

Pull #22949

github

Committed by

web-flow

Commit Message

Merge f730a56cd into 407284c67

Pull Request Pull Request #22949: Add experimental uv resolver for Python lockfiles

Run Details

54 of 97 new or added lines in 5 files covered. (55.67%)

8270 existing lines in 295 files now uncovered.

48990 of 75379 relevant lines covered (64.99%)

1.81 hits per line

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

57.14

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

# Copyright 2021 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, Mapping
from dataclasses import dataclass
from enum import Enum

from pants.base.glob_match_error_behavior import GlobMatchErrorBehavior
from pants.engine.fs import EMPTY_SNAPSHOT, PathGlobs, Snapshot
from pants.engine.intrinsics import digest_to_snapshot, get_digest_contents
from pants.engine.rules import collect_rules, implicitly, rule
from pants.util.collections import ensure_str_list
from pants.util.dirutil import find_nearest_ancestor_file
from pants.util.frozendict import FrozenDict
from pants.util.logging import LogLevel
from pants.util.strutil import softwrap

logger = logging.getLogger(__name__)


@dataclass(frozen=True)
class ConfigFiles:
    """Config files used by a tool run by Pants."""

    snapshot: Snapshot


@dataclass(frozen=True)
class ConfigFilesRequest:
    """Resolve the specified config files if given, else look for candidate config files if
    discovery is enabled.

    Files in `check_existence` only need to exist, whereas files in `check_content` both must exist
    and contain the bytes snippet in the file.
    """

    specified: tuple[str, ...]
    specified_option_name: str | None
    discovery: bool
    check_existence: tuple[str, ...]
    check_content: FrozenDict[str, bytes]

    def __init__(
        self,
        *,
        specified: str | Iterable[str] | None = None,
        specified_option_name: str | None = None,
        discovery: bool = False,
        check_existence: Iterable[str] = (),
        check_content: Mapping[str, bytes] = FrozenDict(),
    ) -> None:
        object.__setattr__(
            self, "specified", tuple(ensure_str_list(specified or (), allow_single_str=True))
        )
        object.__setattr__(self, "specified_option_name", specified_option_name)
        object.__setattr__(self, "discovery", discovery)
        object.__setattr__(self, "check_existence", tuple(sorted(check_existence)))
        object.__setattr__(self, "check_content", FrozenDict(check_content))


@rule(desc="Find config files", level=LogLevel.DEBUG)
async def find_config_file(request: ConfigFilesRequest) -> ConfigFiles:
    config_snapshot = EMPTY_SNAPSHOT
    if request.specified:
        config_snapshot = await digest_to_snapshot(
            **implicitly(
                PathGlobs(
                    globs=request.specified,
                    glob_match_error_behavior=GlobMatchErrorBehavior.error,
                    description_of_origin=f"the option `{request.specified_option_name}`",
                )
            )
        )
        return ConfigFiles(config_snapshot)
    elif request.discovery:
        check_content_digest_contents = await get_digest_contents(
            **implicitly(PathGlobs(request.check_content))
        )
        valid_content_files = tuple(
            file_content.path
            for file_content in check_content_digest_contents
            if request.check_content[file_content.path] in file_content.content
        )
        config_snapshot = await digest_to_snapshot(
            **implicitly(PathGlobs((*request.check_existence, *valid_content_files)))
        )
    return ConfigFiles(config_snapshot)


class OrphanFilepathConfigBehavior(Enum):
    IGNORE = "ignore"
    ERROR = "error"
    WARN = "warn"


@dataclass(frozen=True)
class GatheredConfigFilesByDirectories:
    config_filename: str
    snapshot: Snapshot
    source_dir_to_config_file: FrozenDict[str, str]


@dataclass(frozen=True)
class GatherConfigFilesByDirectoriesRequest:
    tool_name: str
    config_filename: str
    filepaths: tuple[str, ...]
    orphan_filepath_behavior: OrphanFilepathConfigBehavior = OrphanFilepathConfigBehavior.ERROR


@rule
async def gather_config_files_by_workspace_dir(
    request: GatherConfigFilesByDirectoriesRequest,
) -> GatheredConfigFilesByDirectories:
    """Gathers config files from the workspace and indexes them by the directories relative to
    them."""

    source_dirs = frozenset(os.path.dirname(path) for path in request.filepaths)
    source_dirs_with_ancestors = {"", *source_dirs}
    for source_dir in source_dirs:
        source_dir_parts = source_dir.split(os.path.sep)
        source_dir_parts.pop()
        while source_dir_parts:
            source_dirs_with_ancestors.add(os.path.sep.join(source_dir_parts))
            source_dir_parts.pop()

    config_file_globs = [
        os.path.join(dir, request.config_filename) for dir in source_dirs_with_ancestors
    ]
    config_files_snapshot = await digest_to_snapshot(**implicitly(PathGlobs(config_file_globs)))
    config_files_set = set(config_files_snapshot.files)
    source_dir_to_config_file: dict[str, str] = {}
    for source_dir in source_dirs:
        config_file = find_nearest_ancestor_file(
            config_files_set, source_dir, request.config_filename
        )
        if config_file:
            source_dir_to_config_file[source_dir] = config_file
        else:
            msg = softwrap(
                f"""
                No {request.tool_name} file (`{request.config_filename}`) found for
                source directory '{source_dir}'.
                """
            )
            if request.orphan_filepath_behavior == OrphanFilepathConfigBehavior.ERROR:
                raise ValueError(msg)
            elif request.orphan_filepath_behavior == OrphanFilepathConfigBehavior.WARN:
                logger.warning(msg)

    return GatheredConfigFilesByDirectories(
        request.config_filename, config_files_snapshot, FrozenDict(source_dir_to_config_file)
    )


def rules():
    return collect_rules()

1	# Copyright 2021 Pants project contributors (see CONTRIBUTORS.md).
2	# Licensed under the Apache License, Version 2.0 (see LICENSE).
3
4	from __future__ import annotations	5✔
5
6	import logging	5✔
7	import os	5✔
8	from collections.abc import Iterable, Mapping	5✔
9	from dataclasses import dataclass	5✔
10	from enum import Enum	5✔
11
12	from pants.base.glob_match_error_behavior import GlobMatchErrorBehavior	5✔
13	from pants.engine.fs import EMPTY_SNAPSHOT, PathGlobs, Snapshot	5✔
14	from pants.engine.intrinsics import digest_to_snapshot, get_digest_contents	5✔
15	from pants.engine.rules import collect_rules, implicitly, rule	5✔
16	from pants.util.collections import ensure_str_list	5✔
17	from pants.util.dirutil import find_nearest_ancestor_file	5✔
18	from pants.util.frozendict import FrozenDict	5✔
19	from pants.util.logging import LogLevel	5✔
20	from pants.util.strutil import softwrap	5✔
21
22	logger = logging.getLogger(__name__)	5✔
23
24
25	@dataclass(frozen=True)	5✔
26	class ConfigFiles:	5✔
27	"""Config files used by a tool run by Pants."""
28
29	snapshot: Snapshot	5✔
30
31
32	@dataclass(frozen=True)	5✔
33	class ConfigFilesRequest:	5✔
34	"""Resolve the specified config files if given, else look for candidate config files if
35	discovery is enabled.
36
37	Files in `check_existence` only need to exist, whereas files in `check_content` both must exist
38	and contain the bytes snippet in the file.
39	"""
40
41	specified: tuple[str, ...]	5✔
42	specified_option_name: str \| None	5✔
43	discovery: bool	5✔
44	check_existence: tuple[str, ...]	5✔
45	check_content: FrozenDict[str, bytes]	5✔
46
47	def __init__(	5✔
48	self,
49	*,
50	specified: str \| Iterable[str] \| None = None,
51	specified_option_name: str \| None = None,
52	discovery: bool = False,
53	check_existence: Iterable[str] = (),
54	check_content: Mapping[str, bytes] = FrozenDict(),
55	) -> None:
UNCOV 56	object.__setattr__(	×
57	self, "specified", tuple(ensure_str_list(specified or (), allow_single_str=True))
58	)
UNCOV 59	object.__setattr__(self, "specified_option_name", specified_option_name)	×
UNCOV 60	object.__setattr__(self, "discovery", discovery)	×
UNCOV 61	object.__setattr__(self, "check_existence", tuple(sorted(check_existence)))	×
UNCOV 62	object.__setattr__(self, "check_content", FrozenDict(check_content))	×
63
64
65	@rule(desc="Find config files", level=LogLevel.DEBUG)	5✔
66	async def find_config_file(request: ConfigFilesRequest) -> ConfigFiles:	5✔
67	config_snapshot = EMPTY_SNAPSHOT	×
68	if request.specified:	×
69	config_snapshot = await digest_to_snapshot(	×
70	**implicitly(
71	PathGlobs(
72	globs=request.specified,
73	glob_match_error_behavior=GlobMatchErrorBehavior.error,
74	description_of_origin=f"the option `{request.specified_option_name}`",
75	)
76	)
77	)
78	return ConfigFiles(config_snapshot)	×
79	elif request.discovery:	×
80	check_content_digest_contents = await get_digest_contents(	×
81	**implicitly(PathGlobs(request.check_content))
82	)
83	valid_content_files = tuple(	×
84	file_content.path
85	for file_content in check_content_digest_contents
86	if request.check_content[file_content.path] in file_content.content
87	)
88	config_snapshot = await digest_to_snapshot(	×
89	*implicitly(PathGlobs((request.check_existence, *valid_content_files)))
90	)
91	return ConfigFiles(config_snapshot)	×
92
93
94	class OrphanFilepathConfigBehavior(Enum):	5✔
95	IGNORE = "ignore"	5✔
96	ERROR = "error"	5✔
97	WARN = "warn"	5✔
98
99
100	@dataclass(frozen=True)	5✔
101	class GatheredConfigFilesByDirectories:	5✔
102	config_filename: str	5✔
103	snapshot: Snapshot	5✔
104	source_dir_to_config_file: FrozenDict[str, str]	5✔
105
106
107	@dataclass(frozen=True)	5✔
108	class GatherConfigFilesByDirectoriesRequest:	5✔
109	tool_name: str	5✔
110	config_filename: str	5✔
111	filepaths: tuple[str, ...]	5✔
112	orphan_filepath_behavior: OrphanFilepathConfigBehavior = OrphanFilepathConfigBehavior.ERROR	5✔
113
114
115	@rule	5✔
116	async def gather_config_files_by_workspace_dir(	5✔
117	request: GatherConfigFilesByDirectoriesRequest,
118	) -> GatheredConfigFilesByDirectories:
119	"""Gathers config files from the workspace and indexes them by the directories relative to
120	them."""
121
122	source_dirs = frozenset(os.path.dirname(path) for path in request.filepaths)	×
123	source_dirs_with_ancestors = {"", *source_dirs}	×
124	for source_dir in source_dirs:	×
125	source_dir_parts = source_dir.split(os.path.sep)	×
126	source_dir_parts.pop()	×
127	while source_dir_parts:	×
128	source_dirs_with_ancestors.add(os.path.sep.join(source_dir_parts))	×
129	source_dir_parts.pop()	×
130
131	config_file_globs = [	×
132	os.path.join(dir, request.config_filename) for dir in source_dirs_with_ancestors
133	]
134	config_files_snapshot = await digest_to_snapshot(**implicitly(PathGlobs(config_file_globs)))	×
135	config_files_set = set(config_files_snapshot.files)	×
136	source_dir_to_config_file: dict[str, str] = {}	×
137	for source_dir in source_dirs:	×
138	config_file = find_nearest_ancestor_file(	×
139	config_files_set, source_dir, request.config_filename
140	)
141	if config_file:	×
142	source_dir_to_config_file[source_dir] = config_file	×
143	else:
144	msg = softwrap(	×
145	f"""
146	No {request.tool_name} file (`{request.config_filename}`) found for
147	source directory '{source_dir}'.
148	"""
149	)
150	if request.orphan_filepath_behavior == OrphanFilepathConfigBehavior.ERROR:	×
151	raise ValueError(msg)	×
152	elif request.orphan_filepath_behavior == OrphanFilepathConfigBehavior.WARN:	×
153	logger.warning(msg)	×
154
155	return GatheredConfigFilesByDirectories(	×
156	request.config_filename, config_files_snapshot, FrozenDict(source_dir_to_config_file)
157	)
158
159
160	def rules():	5✔
161	return collect_rules()	5✔

pantsbuild / pants / 20332790708

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