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

30.2

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

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

from __future__ import annotations

import dataclasses
import itertools
import operator
from collections import abc, defaultdict
from collections.abc import Iterator, Mapping
from dataclasses import dataclass
from enum import Enum
from functools import reduce
from typing import Any, Self, cast

from pants.build_graph.address import BANNED_CHARS_IN_PARAMETERS
from pants.engine.addresses import Address
from pants.engine.collection import Collection
from pants.engine.engine_aware import EngineAwareParameter
from pants.engine.target import (
    Field,
    FieldDefaults,
    ImmutableValue,
    Target,
    TargetTypesToGenerateTargetsRequests,
)
from pants.util.frozendict import FrozenDict
from pants.util.strutil import bullet_list, pluralize, softwrap


def _named_args_explanation(arg: str) -> str:
    return (
        f"To use `{arg}` as a parameter, you can pass it as a keyword argument to "
        f"give it an alias. For example: `parametrize(short_memorable_name={arg})`"
    )


@dataclass(frozen=True)
class Parametrize:
    """A builtin function/dataclass that can be used to parametrize Targets.

    Parametrization is applied between TargetAdaptor construction and Target instantiation, which
    means that individual Field instances need not be aware of it.
    """

    class _MergeBehaviour(Enum):
        # Do not merge this parametrization.
        never = "never"
        # Discard this parametrization with `other`.
        replace = "replace"

    args: tuple[str, ...]
    kwargs: FrozenDict[str, ImmutableValue]
    is_group: bool
    merge_behaviour: _MergeBehaviour = dataclasses.field(compare=False)

    def __init__(self, *args: str, **kwargs: Any) -> None:
        object.__setattr__(self, "args", tuple(args))
        object.__setattr__(self, "kwargs", FrozenDict.deep_freeze(kwargs))
        object.__setattr__(self, "is_group", False)
        object.__setattr__(self, "merge_behaviour", Parametrize._MergeBehaviour.never)

    def keys(self) -> tuple[str]:
        return (f"parametrize_{hash(self.args)}:{id(self)}",)

    def __getitem__(self, key) -> Any:
        if isinstance(key, str) and key.startswith("parametrize_"):
            return self.to_group()
        else:
            raise KeyError(key)

    def to_group(self) -> Self:
        object.__setattr__(self, "is_group", True)
        return self

    def to_weak(self) -> Self:
        object.__setattr__(self, "merge_behaviour", Parametrize._MergeBehaviour.replace)
        return self

    def to_parameters(self) -> dict[str, Any]:
        """Validates and returns a mapping from aliases to parameter values.

        This conversion is executed lazily to allow for more context in error messages, such as the
        TargetAdaptor consuming the Parametrize instance.
        """
        parameters = dict(self.kwargs)
        for arg in self.args:
            if not isinstance(arg, str):
                raise Exception(
                    f"In {self}:\n  Positional arguments must be strings, but "
                    f"`{arg!r}` was a `{type(arg).__name__}`.\n\n"
                    + _named_args_explanation(f"{arg!r}")
                )
            previous_arg = parameters.get(arg)
            if previous_arg is not None:
                raise Exception(
                    f"In {self}:\n  Positional arguments cannot have the same name as "
                    f"keyword arguments. `{arg}` was also defined as `{arg}={previous_arg!r}`."
                )
            banned_chars = BANNED_CHARS_IN_PARAMETERS & set(arg)
            if banned_chars:
                raise Exception(
                    f"In {self}:\n  Positional argument `{arg}` contained separator characters "
                    f"(`{'`,`'.join(banned_chars)}`).\n\n" + _named_args_explanation(arg)
                )
            parameters[arg] = arg
        return parameters

    @property
    def group_name(self) -> str:
        assert self.is_group
        if len(self.args) == 1:
            name = self.args[0]
            banned_chars = BANNED_CHARS_IN_PARAMETERS & set(name)
            if banned_chars:
                raise Exception(
                    f"In {self}:\n  Parametrization group name `{name}` contained separator characters "
                    f"(`{'`,`'.join(banned_chars)}`)."
                )
            return name
        else:
            raise ValueError(
                softwrap(
                    f"""
                    A parametrize group must begin with the group name followed by the target field
                    values for the group.

                    Example:

                        target(
                            ...,
                            **parametrize("group-a", field_a=1, field_b=True),
                        )

                    Got: `{self!r}`
                    """
                )
            )

    @classmethod
    def expand(
        cls,
        address: Address,
        fields: Mapping[str, Any | Parametrize],
    ) -> Iterator[tuple[Address, Mapping[str, Any]]]:
        """Produces the cartesian product of fields for the given possibly-Parametrized fields.

        Only one level of expansion is performed: if individual field values might also contain
        Parametrize instances (in particular: an `overrides` field), expanding those will require
        separate calls.

        Parametrized groups are expanded however (that is: any `parametrize` field values in a
        `**parametrize()` group are also expanded).
        """
        try:
            yield from cls._expand(address, fields)
        except Exception as e:
            raise Exception(f"Failed to parametrize `{address}`:\n  {e}") from e

    @classmethod
    def _expand(
        cls,
        address: Address,
        fields: Mapping[str, Any | Parametrize],
        _parametrization_group_prefix: str = "",
    ) -> Iterator[tuple[Address, Mapping[str, Any]]]:
        parametrizations = cls._collect_parametrizations(fields)
        cls._check_parametrizations(parametrizations)
        parametrized: list[list[tuple[str, str, Any]]] = [
            [(field_name, alias, field_value) for alias, field_value in v.to_parameters().items()]
            for field_name, v in parametrizations.get(None, ())
        ]
        parametrized_groups: list[tuple[str, str, Parametrize]] = [
            ("parametrize", (_parametrization_group_prefix + group_name), vs[0][1])
            for group_name, vs in parametrizations.items()
            if group_name is not None
        ]
        parameters = address.parameters
        non_parametrized = tuple(
            (field_name, field_value)
            for field_name, field_value in fields.items()
            if not isinstance(field_value, Parametrize)
        )
        if parametrized_groups:
            # Add the groups as one vector for the cross-product.
            parametrized.append(parametrized_groups)

        unparametrize_keys = {k for k, _ in non_parametrized if k in parameters}

        # Remove non-parametrized fields from the address parameters.
        for k in unparametrize_keys:
            parameters.pop(k, None)

        if not parametrized:
            if unparametrize_keys:
                address = address.parametrize(parameters, replace=True)
            yield (address, fields)
            return

        for parametrized_args in itertools.product(*parametrized):
            expanded_parameters = parameters | {
                field_name: alias for field_name, alias, _ in parametrized_args
            }
            # There will be at most one group per cross product.
            parametrize_group: Parametrize | None = next(
                (
                    field_value
                    for _, _, field_value in parametrized_args
                    if isinstance(field_value, Parametrize) and field_value.is_group
                ),
                None,
            )
            if parametrize_group is not None:
                # Exclude fields from parametrize group from address parameters.
                for k in parametrize_group.kwargs.keys() & parameters.keys():
                    expanded_parameters.pop(k, None)
                expand_recursively = any(
                    isinstance(group_value, Parametrize)
                    for group_value in parametrize_group.kwargs.values()
                )
            else:
                expand_recursively = False

            parametrized_args_fields = tuple(
                (field_name, field_value)
                for field_name, _, field_value in parametrized_args
                # Exclude any parametrize group
                if not (isinstance(field_value, Parametrize) and field_value.is_group)
            )
            expanded_fields: dict[str, Any] = dict(non_parametrized + parametrized_args_fields)
            expanded_address = address.parametrize(expanded_parameters, replace=True)

            if expand_recursively:
                assert parametrize_group is not None  # Type narrowing to satisfy mypy.
                # Expand nested parametrize within a parametrized group.
                for grouped_address, grouped_fields in cls._expand(
                    expanded_address,
                    parametrize_group.kwargs,
                    _parametrization_group_prefix=_parametrization_group_prefix
                    + parametrize_group.group_name
                    + "-",
                ):
                    cls._check_conflicting(
                        {
                            name
                            for name in grouped_fields.keys()
                            if isinstance(fields.get(name), Parametrize)
                        }
                    )
                    yield (
                        expanded_address.parametrize(grouped_address.parameters),
                        expanded_fields | dict(grouped_fields),
                    )
            else:
                if parametrize_group is not None:
                    expanded_fields |= dict(parametrize_group.kwargs)
                yield expanded_address, expanded_fields

    @staticmethod
    def _collect_parametrizations(
        fields: Mapping[str, Any | Parametrize],
    ) -> Mapping[str | None, list[tuple[str, Parametrize]]]:
        parametrizations = defaultdict(list)
        for field_name, v in fields.items():
            if not isinstance(v, Parametrize):
                continue
            group_name = None if not v.is_group else v.group_name
            parametrizations[group_name].append((field_name, v))
        return parametrizations

    @staticmethod
    def _check_parametrizations(
        parametrizations: Mapping[str | None, list[tuple[str, Parametrize]]],
    ) -> None:
        for group_name, groups in parametrizations.items():
            if group_name is not None and len(groups) > 1:
                group = Parametrize._combine(*(group for _, group in groups))
                groups.clear()
                groups.append(("combined", group))

        parametrize_field_names = {field_name for field_name, v in parametrizations.get(None, ())}
        parametrize_field_names_from_groups = {
            field_name
            for group_name, groups in parametrizations.items()
            if group_name is not None
            for field_name in groups[0][1].kwargs.keys()
        }
        Parametrize._check_conflicting(
            parametrize_field_names.intersection(parametrize_field_names_from_groups)
        )

    @staticmethod
    def _check_conflicting(conflicting: abc.Collection[str]) -> None:
        if conflicting:
            raise ValueError(
                softwrap(
                    f"""
                    Conflicting parametrizations for {pluralize(len(conflicting), "field", include_count=False)}:
                    {", ".join(map(repr, sorted(conflicting)))}
                    """
                )
            )

    @staticmethod
    def _combine(head: Parametrize, *tail: Parametrize) -> Parametrize:
        return reduce(operator.add, tail, head)

    def __add__(self, other: Parametrize) -> Parametrize:
        if not isinstance(other, Parametrize):
            raise TypeError(f"Can not combine {self} with {other!r}")
        if self.merge_behaviour is Parametrize._MergeBehaviour.replace:
            return other
        if other.merge_behaviour is Parametrize._MergeBehaviour.replace:
            return self
        if self.is_group and other.is_group:
            raise ValueError(f"Parametrization group name is not unique: {self.group_name!r}")
        raise ValueError(f"Can not combine parametrizations: {self} | {other}")

    def __repr__(self) -> str:
        strs = [repr(s) for s in self.args]
        strs.extend(f"{alias}={value!r}" for alias, value in self.kwargs.items())
        return f"parametrize({', '.join(strs)})"


@dataclass(frozen=True)
class _TargetParametrization:
    original_target: Target | None
    parametrization: FrozenDict[Address, Target]

    @property
    def all(self) -> Iterator[Target]:
        if self.original_target:
            yield self.original_target
        yield from self.parametrization.values()

    def get(self, address: Address) -> Target | None:
        """Find the Target with an exact Address match, if any."""
        if self.original_target and self.original_target.address == address:
            return self.original_target
        return self.parametrization.get(address)


# TODO: This is not the right name for this class, nor the best place for it to live. But it is
# consumed by both `pants.engine.internals.graph` and `pants.engine.internals.build_files`, and
# shouldn't live in `pants.engine.target` (yet? needs more stabilization).
@dataclass(frozen=True)
class _TargetParametrizationsRequest(EngineAwareParameter):
    address: Address
    description_of_origin: str = dataclasses.field(hash=False, compare=False)

    def __post_init__(self) -> None:
        if self.address.is_parametrized or self.address.is_generated_target:
            raise ValueError(
                softwrap(
                    f"""
                    Cannot create {self.__class__.__name__} on a generated or parametrized target.

                    Self: {self}
                    """
                )
            )

    def debug_hint(self) -> str:
        return self.address.spec


# TODO: See TODO on _TargetParametrizationsRequest about naming this.
class _TargetParametrizations(Collection[_TargetParametrization]):
    """All parametrizations and generated targets for a single input Address.

    If a Target has been parametrized, the original Target might _not_ be present, due to no Target
    being addressable at the un-parameterized Address.
    """

    @property
    def all(self) -> Iterator[Target]:
        """Iterates over all Target instances which are valid after parametrization."""
        for parametrization in self:
            yield from parametrization.all

    @property
    def parametrizations(self) -> dict[Address, Target]:
        """Returns a merged dict of all generated/parametrized instances, excluding originals."""
        return {
            a: t for parametrization in self for a, t in parametrization.parametrization.items()
        }

    def generated_for(self, address: Address) -> FrozenDict[Address, Target]:
        """Find all Targets generated by the given generator Address."""
        assert not address.is_generated_target
        for parametrization in self:
            if (
                parametrization.original_target
                and parametrization.original_target.address == address
            ):
                return parametrization.parametrization

        raise self._bare_address_error(address)

    def get(
        self,
        address: Address,
        target_types_to_generate_requests: TargetTypesToGenerateTargetsRequests | None = None,
    ) -> Target | None:
        """Find the Target with an exact Address match, if any."""
        for parametrization in self:
            instance = parametrization.get(address)
            if instance is not None:
                return instance

        # TODO: This is an accommodation to allow using file/generator Addresses for
        # non-generator atom targets. See https://github.com/pantsbuild/pants/issues/14419.
        if target_types_to_generate_requests and address.is_generated_target:
            base_address = address.maybe_convert_to_target_generator()
            original_target = self.get(base_address, target_types_to_generate_requests)
            if original_target and not target_types_to_generate_requests.is_generator(
                original_target
            ):
                return original_target

        return None

    def get_all_superset_targets(self, address: Address) -> Iterator[Address]:
        """Yield the input address itself, or any parameterized addresses which are a superset of
        the input address.

        For example, an input address `dir:tgt` may yield `(dir:tgt@k=v1, dir:tgt@k=v2)`.

        If no targets are a match, will yield nothing.
        """
        # Check for exact matches.
        if self.get(address) is not None:
            yield address
            return

        for parametrization in self:
            if parametrization.original_target is not None and address.is_parametrized_subset_of(
                parametrization.original_target.address
            ):
                yield parametrization.original_target.address

            for parametrized_tgt in parametrization.parametrization.values():
                if address.is_parametrized_subset_of(parametrized_tgt.address):
                    yield parametrized_tgt.address

    def get_subset(
        self,
        address: Address,
        consumer: Target,
        field_defaults: FieldDefaults,
        target_types_to_generate_requests: TargetTypesToGenerateTargetsRequests,
    ) -> Target:
        """Find the Target with the given Address, or with fields matching the given consumer."""
        # Check for exact matches.
        instance = self.get(address, target_types_to_generate_requests)
        if instance is not None:
            return instance

        def remaining_fields_match(candidate: Target) -> bool:
            """Returns true if all Fields absent from the candidate's Address match the consumer."""
            unspecified_param_field_names = {
                key for key in candidate.address.parameters.keys() if key not in address.parameters
            }
            return all(
                _concrete_fields_are_equivalent(
                    field_defaults,
                    consumer=consumer,
                    candidate_field=field,
                )
                for field_type, field in candidate.field_values.items()
                if field_type.alias in unspecified_param_field_names
            )

        for parametrization in self:
            # If the given Address is a subset-match of the parametrization's original Target
            # (meaning that the user specified an un-parameterized generator Address), then we
            # need to match against one of the generated Targets instead (because a parametrized
            # generator does not keep its Fields).
            if (
                parametrization.original_target
                and address.is_parametrized_subset_of(parametrization.original_target.address)
                and parametrization.parametrization
                and remaining_fields_match(next(iter(parametrization.parametrization.values())))
            ):
                return parametrization.original_target

        consumer_parametrize_group = consumer.address.parameters.get("parametrize")

        def matching_parametrize_group(candidate: Target) -> bool:
            return candidate.address.parameters.get("parametrize") == consumer_parametrize_group

        for candidate in sorted(
            self.parametrizations.values(), key=matching_parametrize_group, reverse=True
        ):
            # Else, see whether any of the generated targets match, preferring a matching
            # parametrize group when available.
            if address.is_parametrized_subset_of(candidate.address) and remaining_fields_match(
                candidate
            ):
                return candidate

        raise ValueError(
            f"The explicit dependency `{address}` of the target at `{consumer.address}` does "
            "not provide enough address parameters to identify which parametrization of the "
            "dependency target should be used.\n"
            f"Target `{address.maybe_convert_to_target_generator()}` can be addressed as:\n"
            f"{bullet_list(str(t.address) for t in self.all)}"
        )

    def generated_or_generator(self, maybe_generator: Address) -> Iterator[Target]:
        """Yield either the Target, or the generated Targets for the given Address."""
        for parametrization in self:
            if (
                not parametrization.original_target
                or parametrization.original_target.address != maybe_generator
            ):
                continue
            if parametrization.parametrization:
                # Generated Targets.
                yield from parametrization.parametrization.values()
            else:
                # Did not generate targets.
                yield parametrization.original_target
            return

        raise self._bare_address_error(maybe_generator)

    def _bare_address_error(self, address) -> ValueError:
        return ValueError(
            "A `parametrized` target cannot be consumed without its parameters specified.\n"
            f"Target `{address}` can be addressed as:\n"
            f"{bullet_list(str(t.address) for t in self.all)}"
        )


def _concrete_fields_are_equivalent(
    field_defaults: FieldDefaults, *, consumer: Target, candidate_field: Field
) -> bool:
    candidate_field_type = type(candidate_field)
    candidate_field_value = field_defaults.value_or_default(candidate_field)

    if consumer.has_field(candidate_field_type):
        return cast(
            bool,
            field_defaults.value_or_default(consumer[candidate_field_type])
            == candidate_field_value,
        )
    # Else, see if the consumer has a field that is a superclass of `candidate_field_value`, to
    # handle https://github.com/pantsbuild/pants/issues/16190. This is only safe because we are
    # confident that both `candidate_field_type` and the fields from `consumer` are _concrete_,
    # meaning they are not abstract templates like `StringField`.
    superclass = next(
        (
            consumer_field
            for consumer_field in consumer.field_types
            if isinstance(candidate_field, consumer_field)
        ),
        None,
    )
    if superclass is None:
        return False
    return cast(
        bool, field_defaults.value_or_default(consumer[superclass]) == candidate_field_value
    )

1	# Copyright 2022 Pants project contributors (see CONTRIBUTORS.md).
2	# Licensed under the Apache License, Version 2.0 (see LICENSE).
3
4	from __future__ import annotations	3✔
5
6	import dataclasses	3✔
7	import itertools	3✔
8	import operator	3✔
9	from collections import abc, defaultdict	3✔
10	from collections.abc import Iterator, Mapping	3✔
11	from dataclasses import dataclass	3✔
12	from enum import Enum	3✔
13	from functools import reduce	3✔
14	from typing import Any, Self, cast	3✔
15
16	from pants.build_graph.address import BANNED_CHARS_IN_PARAMETERS	3✔
17	from pants.engine.addresses import Address	3✔
18	from pants.engine.collection import Collection	3✔
19	from pants.engine.engine_aware import EngineAwareParameter	3✔
20	from pants.engine.target import (	3✔
21	Field,
22	FieldDefaults,
23	ImmutableValue,
24	Target,
25	TargetTypesToGenerateTargetsRequests,
26	)
27	from pants.util.frozendict import FrozenDict	3✔
28	from pants.util.strutil import bullet_list, pluralize, softwrap	3✔
29
30
31	def _named_args_explanation(arg: str) -> str:	3✔
32	return (	×
33	f"To use `{arg}` as a parameter, you can pass it as a keyword argument to "
34	f"give it an alias. For example: `parametrize(short_memorable_name={arg})`"
35	)
36
37
38	@dataclass(frozen=True)	3✔
39	class Parametrize:	3✔
40	"""A builtin function/dataclass that can be used to parametrize Targets.
41
42	Parametrization is applied between TargetAdaptor construction and Target instantiation, which
43	means that individual Field instances need not be aware of it.
44	"""
45
46	class _MergeBehaviour(Enum):	3✔
47	# Do not merge this parametrization.
48	never = "never"	3✔
49	# Discard this parametrization with `other`.
50	replace = "replace"	3✔
51
52	args: tuple[str, ...]	3✔
53	kwargs: FrozenDict[str, ImmutableValue]	3✔
54	is_group: bool	3✔
55	merge_behaviour: _MergeBehaviour = dataclasses.field(compare=False)	3✔
56
57	def __init__(self, args: str, *kwargs: Any) -> None:	3✔
58	object.__setattr__(self, "args", tuple(args))	×
59	object.__setattr__(self, "kwargs", FrozenDict.deep_freeze(kwargs))	×
60	object.__setattr__(self, "is_group", False)	×
61	object.__setattr__(self, "merge_behaviour", Parametrize._MergeBehaviour.never)	×
62
63	def keys(self) -> tuple[str]:	3✔
64	return (f"parametrize_{hash(self.args)}:{id(self)}",)	×
65
66	def __getitem__(self, key) -> Any:	3✔
67	if isinstance(key, str) and key.startswith("parametrize_"):	×
68	return self.to_group()	×
69	else:
70	raise KeyError(key)	×
71
72	def to_group(self) -> Self:	3✔
73	object.__setattr__(self, "is_group", True)	×
74	return self	×
75
76	def to_weak(self) -> Self:	3✔
77	object.__setattr__(self, "merge_behaviour", Parametrize._MergeBehaviour.replace)	×
78	return self	×
79
80	def to_parameters(self) -> dict[str, Any]:	3✔
81	"""Validates and returns a mapping from aliases to parameter values.
82
83	This conversion is executed lazily to allow for more context in error messages, such as the
84	TargetAdaptor consuming the Parametrize instance.
85	"""
86	parameters = dict(self.kwargs)	×
87	for arg in self.args:	×
88	if not isinstance(arg, str):	×
89	raise Exception(	×
90	f"In {self}:\n Positional arguments must be strings, but "
91	f"`{arg!r}` was a `{type(arg).__name__}`.\n\n"
92	+ _named_args_explanation(f"{arg!r}")
93	)
94	previous_arg = parameters.get(arg)	×
95	if previous_arg is not None:	×
96	raise Exception(	×
97	f"In {self}:\n Positional arguments cannot have the same name as "
98	f"keyword arguments. `{arg}` was also defined as `{arg}={previous_arg!r}`."
99	)
100	banned_chars = BANNED_CHARS_IN_PARAMETERS & set(arg)	×
101	if banned_chars:	×
102	raise Exception(	×
103	f"In {self}:\n Positional argument `{arg}` contained separator characters "
104	f"(`{'`,`'.join(banned_chars)}`).\n\n" + _named_args_explanation(arg)
105	)
106	parameters[arg] = arg	×
107	return parameters	×
108
109	@property	3✔
110	def group_name(self) -> str:	3✔
111	assert self.is_group	×
112	if len(self.args) == 1:	×
113	name = self.args[0]	×
114	banned_chars = BANNED_CHARS_IN_PARAMETERS & set(name)	×
115	if banned_chars:	×
116	raise Exception(	×
117	f"In {self}:\n Parametrization group name `{name}` contained separator characters "
118	f"(`{'`,`'.join(banned_chars)}`)."
119	)
120	return name	×
121	else:
122	raise ValueError(	×
123	softwrap(
124	f"""
125	A parametrize group must begin with the group name followed by the target field
126	values for the group.
127
128	Example:
129
130	target(
131	...,
132	**parametrize("group-a", field_a=1, field_b=True),
133	)
134
135	Got: `{self!r}`
136	"""
137	)
138	)
139
140	@classmethod	3✔
141	def expand(	3✔
142	cls,
143	address: Address,
144	fields: Mapping[str, Any \| Parametrize],
145	) -> Iterator[tuple[Address, Mapping[str, Any]]]:
146	"""Produces the cartesian product of fields for the given possibly-Parametrized fields.
147
148	Only one level of expansion is performed: if individual field values might also contain
149	Parametrize instances (in particular: an `overrides` field), expanding those will require
150	separate calls.
151
152	Parametrized groups are expanded however (that is: any `parametrize` field values in a
153	`**parametrize()` group are also expanded).
154	"""
155	try:	×
156	yield from cls._expand(address, fields)	×
157	except Exception as e:	×
158	raise Exception(f"Failed to parametrize `{address}`:\n {e}") from e	×
159
160	@classmethod	3✔
161	def _expand(	3✔
162	cls,
163	address: Address,
164	fields: Mapping[str, Any \| Parametrize],
165	_parametrization_group_prefix: str = "",
166	) -> Iterator[tuple[Address, Mapping[str, Any]]]:
167	parametrizations = cls._collect_parametrizations(fields)	×
168	cls._check_parametrizations(parametrizations)	×
169	parametrized: list[list[tuple[str, str, Any]]] = [	×
170	[(field_name, alias, field_value) for alias, field_value in v.to_parameters().items()]
171	for field_name, v in parametrizations.get(None, ())
172	]
173	parametrized_groups: list[tuple[str, str, Parametrize]] = [	×
174	("parametrize", (_parametrization_group_prefix + group_name), vs[0][1])
175	for group_name, vs in parametrizations.items()
176	if group_name is not None
177	]
178	parameters = address.parameters	×
179	non_parametrized = tuple(	×
180	(field_name, field_value)
181	for field_name, field_value in fields.items()
182	if not isinstance(field_value, Parametrize)
183	)
184	if parametrized_groups:	×
185	# Add the groups as one vector for the cross-product.
186	parametrized.append(parametrized_groups)	×
187
188	unparametrize_keys = {k for k, _ in non_parametrized if k in parameters}	×
189
190	# Remove non-parametrized fields from the address parameters.
191	for k in unparametrize_keys:	×
192	parameters.pop(k, None)	×
193
194	if not parametrized:	×
195	if unparametrize_keys:	×
196	address = address.parametrize(parameters, replace=True)	×
197	yield (address, fields)	×
198	return	×
199
200	for parametrized_args in itertools.product(*parametrized):	×
201	expanded_parameters = parameters \| {	×
202	field_name: alias for field_name, alias, _ in parametrized_args
203	}
204	# There will be at most one group per cross product.
205	parametrize_group: Parametrize \| None = next(	×
206	(
207	field_value
208	for _, _, field_value in parametrized_args
209	if isinstance(field_value, Parametrize) and field_value.is_group
210	),
211	None,
212	)
213	if parametrize_group is not None:	×
214	# Exclude fields from parametrize group from address parameters.
215	for k in parametrize_group.kwargs.keys() & parameters.keys():	×
216	expanded_parameters.pop(k, None)	×
217	expand_recursively = any(	×
218	isinstance(group_value, Parametrize)
219	for group_value in parametrize_group.kwargs.values()
220	)
221	else:
222	expand_recursively = False	×
223
224	parametrized_args_fields = tuple(	×
225	(field_name, field_value)
226	for field_name, _, field_value in parametrized_args
227	# Exclude any parametrize group
228	if not (isinstance(field_value, Parametrize) and field_value.is_group)
229	)
230	expanded_fields: dict[str, Any] = dict(non_parametrized + parametrized_args_fields)	×
231	expanded_address = address.parametrize(expanded_parameters, replace=True)	×
232
233	if expand_recursively:	×
234	assert parametrize_group is not None # Type narrowing to satisfy mypy.	×
235	# Expand nested parametrize within a parametrized group.
236	for grouped_address, grouped_fields in cls._expand(	×
237	expanded_address,
238	parametrize_group.kwargs,
239	_parametrization_group_prefix=_parametrization_group_prefix
240	+ parametrize_group.group_name
241	+ "-",
242	):
243	cls._check_conflicting(	×
244	{
245	name
246	for name in grouped_fields.keys()
247	if isinstance(fields.get(name), Parametrize)
248	}
249	)
250	yield (	×
251	expanded_address.parametrize(grouped_address.parameters),
252	expanded_fields \| dict(grouped_fields),
253	)
254	else:
255	if parametrize_group is not None:	×
256	expanded_fields \|= dict(parametrize_group.kwargs)	×
257	yield expanded_address, expanded_fields	×
258
259	@staticmethod	3✔
260	def _collect_parametrizations(	3✔
261	fields: Mapping[str, Any \| Parametrize],
262	) -> Mapping[str \| None, list[tuple[str, Parametrize]]]:
263	parametrizations = defaultdict(list)	×
264	for field_name, v in fields.items():	×
265	if not isinstance(v, Parametrize):	×
266	continue	×
267	group_name = None if not v.is_group else v.group_name	×
268	parametrizations[group_name].append((field_name, v))	×
269	return parametrizations	×
270
271	@staticmethod	3✔
272	def _check_parametrizations(	3✔
273	parametrizations: Mapping[str \| None, list[tuple[str, Parametrize]]],
274	) -> None:
275	for group_name, groups in parametrizations.items():	×
276	if group_name is not None and len(groups) > 1:	×
277	group = Parametrize._combine(*(group for _, group in groups))	×
278	groups.clear()	×
279	groups.append(("combined", group))	×
280
281	parametrize_field_names = {field_name for field_name, v in parametrizations.get(None, ())}	×
282	parametrize_field_names_from_groups = {	×
283	field_name
284	for group_name, groups in parametrizations.items()
285	if group_name is not None
286	for field_name in groups[0][1].kwargs.keys()
287	}
288	Parametrize._check_conflicting(	×
289	parametrize_field_names.intersection(parametrize_field_names_from_groups)
290	)
291
292	@staticmethod	3✔
293	def _check_conflicting(conflicting: abc.Collection[str]) -> None:	3✔
294	if conflicting:	×
295	raise ValueError(	×
296	softwrap(
297	f"""
298	Conflicting parametrizations for {pluralize(len(conflicting), "field", include_count=False)}:
299	{", ".join(map(repr, sorted(conflicting)))}
300	"""
301	)
302	)
303
304	@staticmethod	3✔
305	def _combine(head: Parametrize, *tail: Parametrize) -> Parametrize:	3✔
306	return reduce(operator.add, tail, head)	×
307
308	def __add__(self, other: Parametrize) -> Parametrize:	3✔
309	if not isinstance(other, Parametrize):	×
310	raise TypeError(f"Can not combine {self} with {other!r}")	×
311	if self.merge_behaviour is Parametrize._MergeBehaviour.replace:	×
312	return other	×
313	if other.merge_behaviour is Parametrize._MergeBehaviour.replace:	×
314	return self	×
315	if self.is_group and other.is_group:	×
316	raise ValueError(f"Parametrization group name is not unique: {self.group_name!r}")	×
317	raise ValueError(f"Can not combine parametrizations: {self} \| {other}")	×
318
319	def __repr__(self) -> str:	3✔
320	strs = [repr(s) for s in self.args]	×
321	strs.extend(f"{alias}={value!r}" for alias, value in self.kwargs.items())	×
322	return f"parametrize({', '.join(strs)})"	×
323
324
325	@dataclass(frozen=True)	3✔
326	class _TargetParametrization:	3✔
327	original_target: Target \| None	3✔
328	parametrization: FrozenDict[Address, Target]	3✔
329
330	@property	3✔
331	def all(self) -> Iterator[Target]:	3✔
332	if self.original_target:	×
333	yield self.original_target	×
334	yield from self.parametrization.values()	×
335
336	def get(self, address: Address) -> Target \| None:	3✔
337	"""Find the Target with an exact Address match, if any."""
338	if self.original_target and self.original_target.address == address:	×
339	return self.original_target	×
340	return self.parametrization.get(address)	×
341
342
343	# TODO: This is not the right name for this class, nor the best place for it to live. But it is
344	# consumed by both `pants.engine.internals.graph` and `pants.engine.internals.build_files`, and
345	# shouldn't live in `pants.engine.target` (yet? needs more stabilization).
346	@dataclass(frozen=True)	3✔
347	class _TargetParametrizationsRequest(EngineAwareParameter):	3✔
348	address: Address	3✔
349	description_of_origin: str = dataclasses.field(hash=False, compare=False)	3✔
350
351	def __post_init__(self) -> None:	3✔
352	if self.address.is_parametrized or self.address.is_generated_target:	×
353	raise ValueError(	×
354	softwrap(
355	f"""
356	Cannot create {self.__class__.__name__} on a generated or parametrized target.
357
358	Self: {self}
359	"""
360	)
361	)
362
363	def debug_hint(self) -> str:	3✔
364	return self.address.spec	×
365
366
367	# TODO: See TODO on _TargetParametrizationsRequest about naming this.
368	class _TargetParametrizations(Collection[_TargetParametrization]):	3✔
369	"""All parametrizations and generated targets for a single input Address.
370
371	If a Target has been parametrized, the original Target might _not_ be present, due to no Target
372	being addressable at the un-parameterized Address.
373	"""
374
375	@property	3✔
376	def all(self) -> Iterator[Target]:	3✔
377	"""Iterates over all Target instances which are valid after parametrization."""
378	for parametrization in self:	×
379	yield from parametrization.all	×
380
381	@property	3✔
382	def parametrizations(self) -> dict[Address, Target]:	3✔
383	"""Returns a merged dict of all generated/parametrized instances, excluding originals."""
384	return {	×
385	a: t for parametrization in self for a, t in parametrization.parametrization.items()
386	}
387
388	def generated_for(self, address: Address) -> FrozenDict[Address, Target]:	3✔
389	"""Find all Targets generated by the given generator Address."""
390	assert not address.is_generated_target	×
391	for parametrization in self:	×
392	if (	×
393	parametrization.original_target
394	and parametrization.original_target.address == address
395	):
396	return parametrization.parametrization	×
397
398	raise self._bare_address_error(address)	×
399
400	def get(	3✔
401	self,
402	address: Address,
403	target_types_to_generate_requests: TargetTypesToGenerateTargetsRequests \| None = None,
404	) -> Target \| None:
405	"""Find the Target with an exact Address match, if any."""
406	for parametrization in self:	×
407	instance = parametrization.get(address)	×
408	if instance is not None:	×
409	return instance	×
410
411	# TODO: This is an accommodation to allow using file/generator Addresses for
412	# non-generator atom targets. See https://github.com/pantsbuild/pants/issues/14419.
413	if target_types_to_generate_requests and address.is_generated_target:	×
414	base_address = address.maybe_convert_to_target_generator()	×
415	original_target = self.get(base_address, target_types_to_generate_requests)	×
416	if original_target and not target_types_to_generate_requests.is_generator(	×
417	original_target
418	):
419	return original_target	×
420
421	return None	×
422
423	def get_all_superset_targets(self, address: Address) -> Iterator[Address]:	3✔
424	"""Yield the input address itself, or any parameterized addresses which are a superset of
425	the input address.
426
427	For example, an input address `dir:tgt` may yield `(dir:tgt@k=v1, dir:tgt@k=v2)`.
428
429	If no targets are a match, will yield nothing.
430	"""
431	# Check for exact matches.
432	if self.get(address) is not None:	×
433	yield address	×
434	return	×
435
436	for parametrization in self:	×
437	if parametrization.original_target is not None and address.is_parametrized_subset_of(	×
438	parametrization.original_target.address
439	):
440	yield parametrization.original_target.address	×
441
442	for parametrized_tgt in parametrization.parametrization.values():	×
443	if address.is_parametrized_subset_of(parametrized_tgt.address):	×
444	yield parametrized_tgt.address	×
445
446	def get_subset(	3✔
447	self,
448	address: Address,
449	consumer: Target,
450	field_defaults: FieldDefaults,
451	target_types_to_generate_requests: TargetTypesToGenerateTargetsRequests,
452	) -> Target:
453	"""Find the Target with the given Address, or with fields matching the given consumer."""
454	# Check for exact matches.
455	instance = self.get(address, target_types_to_generate_requests)	×
456	if instance is not None:	×
457	return instance	×
458
459	def remaining_fields_match(candidate: Target) -> bool:	×
460	"""Returns true if all Fields absent from the candidate's Address match the consumer."""
461	unspecified_param_field_names = {	×
462	key for key in candidate.address.parameters.keys() if key not in address.parameters
463	}
464	return all(	×
465	_concrete_fields_are_equivalent(
466	field_defaults,
467	consumer=consumer,
468	candidate_field=field,
469	)
470	for field_type, field in candidate.field_values.items()
471	if field_type.alias in unspecified_param_field_names
472	)
473
474	for parametrization in self:	×
475	# If the given Address is a subset-match of the parametrization's original Target
476	# (meaning that the user specified an un-parameterized generator Address), then we
477	# need to match against one of the generated Targets instead (because a parametrized
478	# generator does not keep its Fields).
479	if (	×
480	parametrization.original_target
481	and address.is_parametrized_subset_of(parametrization.original_target.address)
482	and parametrization.parametrization
483	and remaining_fields_match(next(iter(parametrization.parametrization.values())))
484	):
485	return parametrization.original_target	×
486
487	consumer_parametrize_group = consumer.address.parameters.get("parametrize")	×
488
489	def matching_parametrize_group(candidate: Target) -> bool:	×
490	return candidate.address.parameters.get("parametrize") == consumer_parametrize_group	×
491
492	for candidate in sorted(	×
493	self.parametrizations.values(), key=matching_parametrize_group, reverse=True
494	):
495	# Else, see whether any of the generated targets match, preferring a matching
496	# parametrize group when available.
497	if address.is_parametrized_subset_of(candidate.address) and remaining_fields_match(	×
498	candidate
499	):
500	return candidate	×
501
502	raise ValueError(	×
503	f"The explicit dependency `{address}` of the target at `{consumer.address}` does "
504	"not provide enough address parameters to identify which parametrization of the "
505	"dependency target should be used.\n"
506	f"Target `{address.maybe_convert_to_target_generator()}` can be addressed as:\n"
507	f"{bullet_list(str(t.address) for t in self.all)}"
508	)
509
510	def generated_or_generator(self, maybe_generator: Address) -> Iterator[Target]:	3✔
511	"""Yield either the Target, or the generated Targets for the given Address."""
512	for parametrization in self:	×
513	if (	×
514	not parametrization.original_target
515	or parametrization.original_target.address != maybe_generator
516	):
517	continue	×
518	if parametrization.parametrization:	×
519	# Generated Targets.
520	yield from parametrization.parametrization.values()	×
521	else:
522	# Did not generate targets.
523	yield parametrization.original_target	×
524	return	×
525
526	raise self._bare_address_error(maybe_generator)	×
527
528	def _bare_address_error(self, address) -> ValueError:	3✔
529	return ValueError(	×
530	"A `parametrized` target cannot be consumed without its parameters specified.\n"
531	f"Target `{address}` can be addressed as:\n"
532	f"{bullet_list(str(t.address) for t in self.all)}"
533	)
534
535
536	def _concrete_fields_are_equivalent(	3✔
537	field_defaults: FieldDefaults, *, consumer: Target, candidate_field: Field
538	) -> bool:
539	candidate_field_type = type(candidate_field)	×
540	candidate_field_value = field_defaults.value_or_default(candidate_field)	×
541
542	if consumer.has_field(candidate_field_type):	×
543	return cast(	×
544	bool,
545	field_defaults.value_or_default(consumer[candidate_field_type])
546	== candidate_field_value,
547	)
548	# Else, see if the consumer has a field that is a superclass of `candidate_field_value`, to
549	# handle https://github.com/pantsbuild/pants/issues/16190. This is only safe because we are
550	# confident that both `candidate_field_type` and the fields from `consumer` are _concrete_,
551	# meaning they are not abstract templates like `StringField`.
552	superclass = next(	×
553	(
554	consumer_field
555	for consumer_field in consumer.field_types
556	if isinstance(candidate_field, consumer_field)
557	),
558	None,
559	)
560	if superclass is None:	×
561	return False	×
562	return cast(	×
563	bool, field_defaults.value_or_default(consumer[superclass]) == candidate_field_value
564	)

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