18517631058

Committed 15 Oct 2025 04:18AM UTC coverage: 69.207% (-11.1%) from 80.267%

Build # 18517631058

Build Type

Pull #22745

github

Committed by

web-flow

Commit Message

Merge 642a76ca1 into 99919310e

Pull Request Pull Request #22745: [windows] Add windows support in the stdio crate.

Run Details

53815 of 77759 relevant lines covered (69.21%)

2.42 hits per line

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

46.91

/src/python/pants/engine/internals/defaults.py

# Copyright 2022 Pants project contributors (see CONTRIBUTORS.md).
# Licensed under the Apache License, Version 2.0 (see LICENSE).
"""The `BuildFileDefaultsParserState.set_defaults` is used by the pants.engine.internals.Parser,
exposed as the `__defaults__` BUILD file symbol.

When parsing a BUILD (from the rule `pants.engine.internals.build_files.parse_address_family`) the
defaults from the closest parent BUILD file is passed as input to the parser, and the new defaults
resulting after the BUILD file have been parsed is returned in the `AddressFamily`.

These defaults are then applied when creating the `TargetAdaptor` targets by the `Registrar` in the
parser.
"""

from __future__ import annotations

from collections.abc import Callable, Iterable, Mapping
from dataclasses import dataclass
from typing import Any, Union

from pants.engine.addresses import Address
from pants.engine.internals.parametrize import Parametrize
from pants.engine.target import (
    Field,
    ImmutableValue,
    InvalidFieldException,
    RegisteredTargetTypes,
    Target,
    TargetGenerator,
)
from pants.engine.unions import UnionMembership
from pants.util.frozendict import FrozenDict

SetDefaultsValueT = Mapping[str, Any]
SetDefaultsKeyT = Union[str, tuple[str, ...]]
SetDefaultsT = Mapping[SetDefaultsKeyT, SetDefaultsValueT]


class BuildFileDefaults(FrozenDict[str, FrozenDict[str, ImmutableValue]]):
    """Map target types to default field values."""


class ParametrizeDefault(Parametrize):
    """Parametrize for default field values.

    This is to have eager validation on the field values rather than erroring first when applied on
    an actual target.
    """

    @classmethod
    def create(
        cls, freeze: Callable[[Any], ImmutableValue], parametrize: Parametrize
    ) -> ParametrizeDefault:
        return cls(
            *map(freeze, parametrize.args),
            **{kw: freeze(arg) for kw, arg in parametrize.kwargs.items()},
        ).to_weak()


@dataclass
class BuildFileDefaultsParserState:
    address: Address
    defaults: dict[str, Mapping[str, Any]]
    registered_target_types: RegisteredTargetTypes
    union_membership: UnionMembership

    @classmethod
    def create(
        cls,
        path: str,
        defaults: BuildFileDefaults,
        registered_target_types: RegisteredTargetTypes,
        union_membership: UnionMembership,
    ) -> BuildFileDefaultsParserState:
        return cls(
            address=Address(path, generated_name="__defaults__"),
            defaults=dict(defaults),
            registered_target_types=registered_target_types,
            union_membership=union_membership,
        )

    def _freeze_field_value(self, field_type: type[Field], value: Any) -> ImmutableValue:
        if isinstance(value, ParametrizeDefault):
            return value
        elif isinstance(value, Parametrize):

            def freeze(v: Any) -> ImmutableValue:
                return self._freeze_field_value(field_type, v)

            return ParametrizeDefault.create(freeze, value)
        else:
            return field_type.compute_value(raw_value=value, address=self.address)

    def get_frozen_defaults(self) -> BuildFileDefaults:
        types = self.registered_target_types.aliases_to_types
        return BuildFileDefaults(
            {
                target_alias: FrozenDict(
                    {
                        **{
                            field_type.alias: self._freeze_field_value(field_type, default)
                            for field_alias, default in fields.items()
                            for field_type in self._target_type_field_types(types[target_alias])
                            if field_alias in (field_type.alias, field_type.deprecated_alias)
                        },
                        **{
                            key: ParametrizeDefault(
                                parametrize.group_name,
                                **{
                                    field_type.alias: self._freeze_field_value(field_type, default)
                                    for field_alias, default in parametrize.kwargs.items()
                                    for field_type in self._target_type_field_types(
                                        types[target_alias]
                                    )
                                    if field_alias
                                    in (field_type.alias, field_type.deprecated_alias)
                                },
                            )
                            .to_weak()
                            .to_group()
                            for key, parametrize in fields.items()
                            if isinstance(parametrize, Parametrize) and parametrize.is_group
                        },
                    }
                )
                for target_alias, fields in self.defaults.items()
            }
        )

    def get(self, target_alias: str) -> Mapping[str, Any]:
        # Used by `pants.engine.internals.parser.Parser._generate_symbols.Registrar.__call__`
        return self.defaults.get(target_alias, {})

    def set_defaults(
        self,
        *args: SetDefaultsT,
        all: SetDefaultsValueT | None = None,
        extend: bool = False,
        ignore_unknown_fields: bool = False,
        ignore_unknown_targets: bool = False,
    ) -> None:
        defaults: dict[str, dict[str, Any]] = (
            {} if not extend else {k: dict(v) for k, v in self.defaults.items()}
        )

        if all is not None:
            self._process_defaults(
                defaults,
                {tuple(self.registered_target_types.aliases): all},
                ignore_unknown_fields=True,
                ignore_unknown_targets=ignore_unknown_targets,
            )

        for arg in args:
            self._process_defaults(
                defaults,
                arg,
                ignore_unknown_fields=ignore_unknown_fields,
                ignore_unknown_targets=ignore_unknown_targets,
            )

        # Update with new defaults, dropping targets without any default values.
        for tgt, default in defaults.items():
            if not default:
                self.defaults.pop(tgt, None)
            else:
                self.defaults[tgt] = default

    def _target_type_field_types(self, target_type: type[Target]) -> tuple[type[Field], ...]:
        return (
            *target_type.class_field_types(self.union_membership),
            *(target_type.moved_fields if issubclass(target_type, TargetGenerator) else ()),
        )

    def _process_defaults(
        self,
        defaults: dict[str, dict[str, Any]],
        targets_defaults: SetDefaultsT,
        ignore_unknown_fields: bool = False,
        ignore_unknown_targets: bool = False,
    ):
        if not isinstance(targets_defaults, dict):
            raise ValueError(
                f"Expected dictionary mapping targets to default field values for {self.address} "
                f"but got: {type(targets_defaults).__name__}."
            )

        types = self.registered_target_types.aliases_to_types
        for target, default in targets_defaults.items():
            if not isinstance(default, dict):
                raise ValueError(
                    f"Invalid default field values in {self.address} for target type {target}, "
                    f"must be an `dict` but was {default!r} with type `{type(default).__name__}`."
                )

            targets: Iterable[str]
            targets = target if isinstance(target, tuple) else (target,)
            for target_alias in map(str, targets):
                if target_alias in types:
                    target_type = types[target_alias]
                elif ignore_unknown_targets:
                    continue
                else:
                    raise ValueError(f"Unrecognized target type {target_alias} in {self.address}.")

                # Copy default dict if we may mutate it.
                raw_values = dict(default) if ignore_unknown_fields else default

                # Validate that field exists on target
                valid_field_aliases = set(
                    target_type._get_field_aliases_to_field_types(
                        self._target_type_field_types(target_type)
                    ).keys()
                )

                def _check_field_alias(field_alias: str) -> None:
                    if field_alias in valid_field_aliases:
                        return
                    if not ignore_unknown_fields:
                        raise InvalidFieldException(
                            f"Unrecognized field `{field_alias}` for target {target_type.alias}. "
                            f"Valid fields are: {', '.join(sorted(valid_field_aliases))}.",
                        )
                    elif field_alias in raw_values:
                        del raw_values[field_alias]

                for field_alias, field_value in default.items():
                    if isinstance(field_value, Parametrize) and field_value.is_group:
                        field_value.to_weak()
                        for parametrize_field_alias in field_value.kwargs.keys():
                            _check_field_alias(parametrize_field_alias)
                    else:
                        _check_field_alias(field_alias)

                # Merge all provided defaults for this call.
                defaults.setdefault(target_type.alias, {}).update(raw_values)

1	# Copyright 2022 Pants project contributors (see CONTRIBUTORS.md).
2	# Licensed under the Apache License, Version 2.0 (see LICENSE).
3	"""The `BuildFileDefaultsParserState.set_defaults` is used by the pants.engine.internals.Parser,
4	exposed as the `__defaults__` BUILD file symbol.
5
6	When parsing a BUILD (from the rule `pants.engine.internals.build_files.parse_address_family`) the
7	defaults from the closest parent BUILD file is passed as input to the parser, and the new defaults
8	resulting after the BUILD file have been parsed is returned in the `AddressFamily`.
9
10	These defaults are then applied when creating the `TargetAdaptor` targets by the `Registrar` in the
11	parser.
12	"""
13
14	from __future__ import annotations	7✔
15
16	from collections.abc import Callable, Iterable, Mapping	7✔
17	from dataclasses import dataclass	7✔
18	from typing import Any, Union	7✔
19
20	from pants.engine.addresses import Address	7✔
21	from pants.engine.internals.parametrize import Parametrize	7✔
22	from pants.engine.target import (	7✔
23	Field,
24	ImmutableValue,
25	InvalidFieldException,
26	RegisteredTargetTypes,
27	Target,
28	TargetGenerator,
29	)
30	from pants.engine.unions import UnionMembership	7✔
31	from pants.util.frozendict import FrozenDict	7✔
32
33	SetDefaultsValueT = Mapping[str, Any]	7✔
34	SetDefaultsKeyT = Union[str, tuple[str, ...]]	7✔
35	SetDefaultsT = Mapping[SetDefaultsKeyT, SetDefaultsValueT]	7✔
36
37
38	class BuildFileDefaults(FrozenDict[str, FrozenDict[str, ImmutableValue]]):	7✔
39	"""Map target types to default field values."""
40
41
42	class ParametrizeDefault(Parametrize):	7✔
43	"""Parametrize for default field values.
44
45	This is to have eager validation on the field values rather than erroring first when applied on
46	an actual target.
47	"""
48
49	@classmethod	7✔
50	def create(	7✔
51	cls, freeze: Callable[[Any], ImmutableValue], parametrize: Parametrize
52	) -> ParametrizeDefault:
53	return cls(	×
54	*map(freeze, parametrize.args),
55	**{kw: freeze(arg) for kw, arg in parametrize.kwargs.items()},
56	).to_weak()
57
58
59	@dataclass	7✔
60	class BuildFileDefaultsParserState:	7✔
61	address: Address	7✔
62	defaults: dict[str, Mapping[str, Any]]	7✔
63	registered_target_types: RegisteredTargetTypes	7✔
64	union_membership: UnionMembership	7✔
65
66	@classmethod	7✔
67	def create(	7✔
68	cls,
69	path: str,
70	defaults: BuildFileDefaults,
71	registered_target_types: RegisteredTargetTypes,
72	union_membership: UnionMembership,
73	) -> BuildFileDefaultsParserState:
74	return cls(	1✔
75	address=Address(path, generated_name="__defaults__"),
76	defaults=dict(defaults),
77	registered_target_types=registered_target_types,
78	union_membership=union_membership,
79	)
80
81	def _freeze_field_value(self, field_type: type[Field], value: Any) -> ImmutableValue:	7✔
82	if isinstance(value, ParametrizeDefault):	1✔
83	return value	×
84	elif isinstance(value, Parametrize):	1✔
85
86	def freeze(v: Any) -> ImmutableValue:	×
87	return self._freeze_field_value(field_type, v)	×
88
89	return ParametrizeDefault.create(freeze, value)	×
90	else:
91	return field_type.compute_value(raw_value=value, address=self.address)	1✔
92
93	def get_frozen_defaults(self) -> BuildFileDefaults:	7✔
94	types = self.registered_target_types.aliases_to_types	1✔
95	return BuildFileDefaults(	1✔
96	{
97	target_alias: FrozenDict(
98	{
99	**{
100	field_type.alias: self._freeze_field_value(field_type, default)
101	for field_alias, default in fields.items()
102	for field_type in self._target_type_field_types(types[target_alias])
103	if field_alias in (field_type.alias, field_type.deprecated_alias)
104	},
105	**{
106	key: ParametrizeDefault(
107	parametrize.group_name,
108	**{
109	field_type.alias: self._freeze_field_value(field_type, default)
110	for field_alias, default in parametrize.kwargs.items()
111	for field_type in self._target_type_field_types(
112	types[target_alias]
113	)
114	if field_alias
115	in (field_type.alias, field_type.deprecated_alias)
116	},
117	)
118	.to_weak()
119	.to_group()
120	for key, parametrize in fields.items()
121	if isinstance(parametrize, Parametrize) and parametrize.is_group
122	},
123	}
124	)
125	for target_alias, fields in self.defaults.items()
126	}
127	)
128
129	def get(self, target_alias: str) -> Mapping[str, Any]:	7✔
130	# Used by `pants.engine.internals.parser.Parser._generate_symbols.Registrar.__call__`
131	return self.defaults.get(target_alias, {})	1✔
132
133	def set_defaults(	7✔
134	self,
135	*args: SetDefaultsT,
136	all: SetDefaultsValueT \| None = None,
137	extend: bool = False,
138	ignore_unknown_fields: bool = False,
139	ignore_unknown_targets: bool = False,
140	) -> None:
141	defaults: dict[str, dict[str, Any]] = (	×
142	{} if not extend else {k: dict(v) for k, v in self.defaults.items()}
143	)
144
145	if all is not None:	×
146	self._process_defaults(	×
147	defaults,
148	{tuple(self.registered_target_types.aliases): all},
149	ignore_unknown_fields=True,
150	ignore_unknown_targets=ignore_unknown_targets,
151	)
152
153	for arg in args:	×
154	self._process_defaults(	×
155	defaults,
156	arg,
157	ignore_unknown_fields=ignore_unknown_fields,
158	ignore_unknown_targets=ignore_unknown_targets,
159	)
160
161	# Update with new defaults, dropping targets without any default values.
162	for tgt, default in defaults.items():	×
163	if not default:	×
164	self.defaults.pop(tgt, None)	×
165	else:
166	self.defaults[tgt] = default	×
167
168	def _target_type_field_types(self, target_type: type[Target]) -> tuple[type[Field], ...]:	7✔
169	return (	1✔
170	*target_type.class_field_types(self.union_membership),
171	*(target_type.moved_fields if issubclass(target_type, TargetGenerator) else ()),
172	)
173
174	def _process_defaults(	7✔
175	self,
176	defaults: dict[str, dict[str, Any]],
177	targets_defaults: SetDefaultsT,
178	ignore_unknown_fields: bool = False,
179	ignore_unknown_targets: bool = False,
180	):
181	if not isinstance(targets_defaults, dict):	×
182	raise ValueError(	×
183	f"Expected dictionary mapping targets to default field values for {self.address} "
184	f"but got: {type(targets_defaults).__name__}."
185	)
186
187	types = self.registered_target_types.aliases_to_types	×
188	for target, default in targets_defaults.items():	×
189	if not isinstance(default, dict):	×
190	raise ValueError(	×
191	f"Invalid default field values in {self.address} for target type {target}, "
192	f"must be an `dict` but was {default!r} with type `{type(default).__name__}`."
193	)
194
195	targets: Iterable[str]
196	targets = target if isinstance(target, tuple) else (target,)	×
197	for target_alias in map(str, targets):	×
198	if target_alias in types:	×
199	target_type = types[target_alias]	×
200	elif ignore_unknown_targets:	×
201	continue	×
202	else:
203	raise ValueError(f"Unrecognized target type {target_alias} in {self.address}.")	×
204
205	# Copy default dict if we may mutate it.
206	raw_values = dict(default) if ignore_unknown_fields else default	×
207
208	# Validate that field exists on target
209	valid_field_aliases = set(	×
210	target_type._get_field_aliases_to_field_types(
211	self._target_type_field_types(target_type)
212	).keys()
213	)
214
215	def _check_field_alias(field_alias: str) -> None:	×
216	if field_alias in valid_field_aliases:	×
217	return	×
218	if not ignore_unknown_fields:	×
219	raise InvalidFieldException(	×
220	f"Unrecognized field `{field_alias}` for target {target_type.alias}. "
221	f"Valid fields are: {', '.join(sorted(valid_field_aliases))}.",
222	)
223	elif field_alias in raw_values:	×
224	del raw_values[field_alias]	×
225
226	for field_alias, field_value in default.items():	×
227	if isinstance(field_value, Parametrize) and field_value.is_group:	×
228	field_value.to_weak()	×
229	for parametrize_field_alias in field_value.kwargs.keys():	×
230	_check_field_alias(parametrize_field_alias)	×
231	else:
232	_check_field_alias(field_alias)	×
233
234	# Merge all provided defaults for this call.
235	defaults.setdefault(target_type.alias, {}).update(raw_values)	×

pantsbuild / pants / 18517631058

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