22285099215

Committed 22 Feb 2026 08:52PM UTC coverage: 75.854% (-17.1%) from 92.936%

Build # 22285099215

Build Type

Pull #23121

github

Committed by

web-flow

Commit Message

Merge c7299df9c into ba8359840

Pull Request Pull Request #23121: fix issue with optional fields in dependency validator

Run Details

28 of 29 new or added lines in 2 files covered. (96.55%)

11174 existing lines in 400 files now uncovered.

53694 of 70786 relevant lines covered (75.85%)

1.88 hits per line

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

94.05

/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	4✔
5
6	import logging	4✔
7	import os	4✔
8	from collections.abc import Iterable, Mapping	4✔
9	from dataclasses import dataclass	4✔
10	from enum import Enum	4✔
11
12	from pants.base.glob_match_error_behavior import GlobMatchErrorBehavior	4✔
13	from pants.engine.fs import EMPTY_SNAPSHOT, PathGlobs, Snapshot	4✔
14	from pants.engine.intrinsics import digest_to_snapshot, get_digest_contents	4✔
15	from pants.engine.rules import collect_rules, implicitly, rule	4✔
16	from pants.util.collections import ensure_str_list	4✔
17	from pants.util.dirutil import find_nearest_ancestor_file	4✔
18	from pants.util.frozendict import FrozenDict	4✔
19	from pants.util.logging import LogLevel	4✔
20	from pants.util.strutil import softwrap	4✔
21
22	logger = logging.getLogger(__name__)	4✔
23
24
25	@dataclass(frozen=True)	4✔
26	class ConfigFiles:	4✔
27	"""Config files used by a tool run by Pants."""
28
29	snapshot: Snapshot	4✔
30
31
32	@dataclass(frozen=True)	4✔
33	class ConfigFilesRequest:	4✔
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, ...]	4✔
42	specified_option_name: str \| None	4✔
43	discovery: bool	4✔
44	check_existence: tuple[str, ...]	4✔
45	check_content: FrozenDict[str, bytes]	4✔
46
47	def __init__(	4✔
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:
56	object.__setattr__(	4✔
57	self, "specified", tuple(ensure_str_list(specified or (), allow_single_str=True))
58	)
59	object.__setattr__(self, "specified_option_name", specified_option_name)	4✔
60	object.__setattr__(self, "discovery", discovery)	4✔
61	object.__setattr__(self, "check_existence", tuple(sorted(check_existence)))	4✔
62	object.__setattr__(self, "check_content", FrozenDict(check_content))	4✔
63
64
65	@rule(desc="Find config files", level=LogLevel.DEBUG)	4✔
66	async def find_config_file(request: ConfigFilesRequest) -> ConfigFiles:	4✔
67	config_snapshot = EMPTY_SNAPSHOT	4✔
68	if request.specified:	4✔
69	config_snapshot = await digest_to_snapshot(	4✔
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)	4✔
79	elif request.discovery:	4✔
80	check_content_digest_contents = await get_digest_contents(	4✔
81	**implicitly(PathGlobs(request.check_content))
82	)
83	valid_content_files = tuple(	4✔
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(	4✔
89	*implicitly(PathGlobs((request.check_existence, *valid_content_files)))
90	)
91	return ConfigFiles(config_snapshot)	4✔
92
93
94	class OrphanFilepathConfigBehavior(Enum):	4✔
95	IGNORE = "ignore"	4✔
96	ERROR = "error"	4✔
97	WARN = "warn"	4✔
98
99
100	@dataclass(frozen=True)	4✔
101	class GatheredConfigFilesByDirectories:	4✔
102	config_filename: str	4✔
103	snapshot: Snapshot	4✔
104	source_dir_to_config_file: FrozenDict[str, str]	4✔
105
106
107	@dataclass(frozen=True)	4✔
108	class GatherConfigFilesByDirectoriesRequest:	4✔
109	tool_name: str	4✔
110	config_filename: str	4✔
111	filepaths: tuple[str, ...]	4✔
112	orphan_filepath_behavior: OrphanFilepathConfigBehavior = OrphanFilepathConfigBehavior.ERROR	4✔
113
114
115	@rule	4✔
116	async def gather_config_files_by_workspace_dir(	4✔
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)	1✔
123	source_dirs_with_ancestors = {"", *source_dirs}	1✔
124	for source_dir in source_dirs:	1✔
125	source_dir_parts = source_dir.split(os.path.sep)	1✔
126	source_dir_parts.pop()	1✔
127	while source_dir_parts:	1✔
UNCOV 128	source_dirs_with_ancestors.add(os.path.sep.join(source_dir_parts))	×
UNCOV 129	source_dir_parts.pop()	×
130
131	config_file_globs = [	1✔
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)))	1✔
135	config_files_set = set(config_files_snapshot.files)	1✔
136	source_dir_to_config_file: dict[str, str] = {}	1✔
137	for source_dir in source_dirs:	1✔
138	config_file = find_nearest_ancestor_file(	1✔
139	config_files_set, source_dir, request.config_filename
140	)
141	if config_file:	1✔
UNCOV 142	source_dir_to_config_file[source_dir] = config_file	×
143	else:
144	msg = softwrap(	1✔
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:	1✔
151	raise ValueError(msg)	×
152	elif request.orphan_filepath_behavior == OrphanFilepathConfigBehavior.WARN:	1✔
153	logger.warning(msg)	×
154
155	return GatheredConfigFilesByDirectories(	1✔
156	request.config_filename, config_files_snapshot, FrozenDict(source_dir_to_config_file)
157	)
158
159
160	def rules():	4✔
161	return collect_rules()	4✔

pantsbuild / pants / 22285099215

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