26080722777

Committed 19 May 2026 06:37AM UTC coverage: 52.106% (-11.5%) from 63.597%

Build # 26080722777

Build Type

Pull #23250

github

Committed by

web-flow

Commit Message

Merge 63ec06323 into 2693df832

Pull Request Pull Request #23250: Feature: Add generic option to docker image

Coverage Stats

12 of 50 new or added lines in 3 files covered. (24.0%)

5382 existing lines in 201 files now uncovered.

32053 of 61515 relevant lines covered (52.11%)

1.04 hits per line

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

86.36

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

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

from __future__ import annotations

import os.path
from collections.abc import Iterable, Mapping
from dataclasses import dataclass
from typing import TypeVar

from pants.backend.project_info.filter_targets import FilterSubsystem
from pants.base.exceptions import MappingError
from pants.build_graph.address import Address, BuildFileAddress
from pants.engine.addresses import Addresses
from pants.engine.collection import Collection
from pants.engine.env_vars import EnvironmentVars
from pants.engine.internals.defaults import BuildFileDefaults, BuildFileDefaultsParserState
from pants.engine.internals.dep_rules import (
    BuildFileDependencyRules,
    BuildFileDependencyRulesParserState,
)
from pants.engine.internals.parser import BuildFilePreludeSymbols, Parser
from pants.engine.internals.target_adaptor import TargetAdaptor
from pants.engine.target import RegisteredTargetTypes, Tags, Target
from pants.util.filtering import TargetFilter, and_filters, create_filters
from pants.util.frozendict import FrozenDict
from pants.util.memo import memoized_property


class DuplicateNameError(MappingError):
    """Indicates more than one top-level object was found with the same name."""


AddressMapT = TypeVar("AddressMapT", bound="AddressMap")


# A pseudo-spec to refer to deleted paths.
DELETED_SPEC_PATH = "__deleted"
# A pseudo-target type to refer to deleted paths.
DELETED_TARGET_TYPE = "__deleted_target"


@dataclass(frozen=True)
class AddressMap:
    """Maps target adaptors from a byte source."""

    path: str
    name_to_target_adaptor: FrozenDict[str, TargetAdaptor]

    @classmethod
    def parse(
        cls,
        filepath: str,
        build_file_content: str,
        parser: Parser,
        extra_symbols: BuildFilePreludeSymbols,
        env_vars: EnvironmentVars,
        is_bootstrap: bool,
        defaults: BuildFileDefaultsParserState,
        dependents_rules: BuildFileDependencyRulesParserState | None,
        dependencies_rules: BuildFileDependencyRulesParserState | None,
    ) -> AddressMap:
        """Parses a source for targets.

        The target adaptors are all 'thin': any targets they point to in other namespaces or even in
        the same namespace but from a separate source are left as unresolved pointers.
        """
        try:
            target_adaptors = parser.parse(
                filepath,
                build_file_content,
                extra_symbols,
                env_vars,
                is_bootstrap,
                defaults,
                dependents_rules,
                dependencies_rules,
            )
        except Exception as e:
            raise MappingError(f"Failed to parse ./{filepath}:\n{type(e).__name__}: {e}")
        return cls.create(filepath, target_adaptors)

    @classmethod
    def create(
        cls: type[AddressMapT], filepath: str, target_adaptors: Iterable[TargetAdaptor]
    ) -> AddressMapT:
        name_to_target_adaptors: dict[str, TargetAdaptor] = {}
        for target_adaptor in target_adaptors:
            name = target_adaptor.name or os.path.basename(os.path.dirname(filepath))
            if name in name_to_target_adaptors:
                duplicate = name_to_target_adaptors[name]
                raise DuplicateNameError(
                    f"A target already exists at {filepath!r} with name {name!r} and target type "
                    f"{duplicate.type_alias!r}. The {target_adaptor.type_alias!r} target "
                    "cannot use the same name."
                )
            name_to_target_adaptors[name] = target_adaptor
        return cls(filepath, FrozenDict(sorted(name_to_target_adaptors.items())))


class DifferingFamiliesError(MappingError):
    """Indicates an attempt was made to merge address maps from different families together."""


@dataclass(frozen=True)
class AddressFamily:
    """Represents the family of target adaptors collected from the BUILD files in one directory.

    To create an AddressFamily, use `create`.

    An address family can be composed of the target adaptors from zero or more underlying address
    sources. An "empty" AddressFamily is legal, and is the result when there are not build files in
    a particular namespace.

    :param namespace: The namespace path of this address family.
    :param name_to_target_adaptors: A dict mapping from name to the target adaptor.
    :param defaults: The default target field values, per target type, applicable for this address family.
    :param dependents_rules: The rules to apply on incoming dependencies to targets in this family.
    :param dependencies_rules: The rules to apply on the outgoing dependencies from targets in this family.
    """

    # The directory from which the adaptors were parsed.
    namespace: str
    name_to_target_adaptors: dict[str, tuple[str, TargetAdaptor]]
    defaults: BuildFileDefaults
    dependents_rules: BuildFileDependencyRules | None
    dependencies_rules: BuildFileDependencyRules | None

    @classmethod
    def create(
        cls,
        spec_path: str,
        address_maps: Iterable[AddressMap],
        defaults: BuildFileDefaults = BuildFileDefaults({}),
        dependents_rules: BuildFileDependencyRules | None = None,
        dependencies_rules: BuildFileDependencyRules | None = None,
    ) -> AddressFamily:
        """Creates an address family from the given set of address maps.

        :param spec_path: The directory prefix shared by all address_maps.
        :param address_maps: The family of maps that form this namespace.
        :raises: :class:`MappingError` if the given address maps do not form a family.
        """
        if spec_path == ".":
            spec_path = ""
        for address_map in address_maps:
            if not address_map.path.startswith(spec_path):
                raise DifferingFamiliesError(
                    f"Expected AddressMaps to share the same parent directory {spec_path!r}, "
                    f"but received: {address_map.path!r}"
                )

        name_to_target_adaptors: dict[str, tuple[str, TargetAdaptor]] = {}
        for address_map in address_maps:
            for name, target_adaptor in address_map.name_to_target_adaptor.items():
                if name in name_to_target_adaptors:
                    previous_path, _ = name_to_target_adaptors[name]
                    raise DuplicateNameError(
                        f"A target with name {name!r} is already defined in {previous_path!r}, but "
                        f"is also defined in {address_map.path!r}. Because both targets share the "
                        f"same namespace of {spec_path!r}, this is not allowed."
                    )
                name_to_target_adaptors[name] = (address_map.path, target_adaptor)
        return AddressFamily(
            namespace=spec_path,
            name_to_target_adaptors=dict(sorted(name_to_target_adaptors.items())),
            defaults=defaults,
            dependents_rules=dependents_rules,
            dependencies_rules=dependencies_rules,
        )

    @memoized_property
    def addresses_to_target_adaptors(self) -> Mapping[Address, TargetAdaptor]:
        return {
            Address(spec_path=self.namespace, target_name=name): target_adaptor
            for name, (_, target_adaptor) in self.name_to_target_adaptors.items()
        }

    @memoized_property
    def build_file_addresses(self) -> tuple[BuildFileAddress, ...]:
        return tuple(
            BuildFileAddress(
                rel_path=path, address=Address(spec_path=self.namespace, target_name=name)
            )
            for name, (path, _) in self.name_to_target_adaptors.items()
        )

    @property
    def target_names(self) -> tuple[str, ...]:
        return tuple(addr.target_name for addr in self.addresses_to_target_adaptors)

    def get_target_adaptor(self, address: Address) -> TargetAdaptor | None:
        if self.namespace == DELETED_SPEC_PATH:
            return TargetAdaptor(DELETED_TARGET_TYPE, "deleted_files", "deleted files")
        assert address.spec_path == self.namespace
        entry = self.name_to_target_adaptors.get(address.target_name)
        if entry is None:
            return None
        _, target_adaptor = entry
        return target_adaptor

    def __hash__(self):
        return hash((self.namespace, self.defaults))

    def __repr__(self) -> str:
        return (
            f"AddressFamily(namespace={self.namespace!r}, "
            f"name_to_target_adaptors={sorted(self.name_to_target_adaptors.keys())})"
        )


@dataclass(frozen=True)
class SpecsFilter:
    """Filters targets with the `--tags`, `--exclude-target-regexp`, and `[filter]` subsystem
    options."""

    is_specified: bool
    filter_subsystem_filter: TargetFilter
    tags_filter: TargetFilter

    @classmethod
    def create(
        cls,
        filter_subsystem: FilterSubsystem,
        registered_target_types: RegisteredTargetTypes,
        *,
        tags: Iterable[str],
    ) -> SpecsFilter:
        def tags_outer_filter(tag: str) -> TargetFilter:
            def tags_inner_filter(tgt: Target) -> bool:
                return tag in (tgt.get(Tags).value or [])

            return tags_inner_filter

        tags_filter = and_filters(create_filters(tags, tags_outer_filter))

        return SpecsFilter(
            is_specified=bool(filter_subsystem.is_specified() or tags),
            filter_subsystem_filter=filter_subsystem.all_filters(registered_target_types),
            tags_filter=tags_filter,
        )

    def matches(self, target: Target) -> bool:
        """Check that the target matches the provided `--tag` and `--exclude-target-regexp`
        options."""
        return self.tags_filter(target) and self.filter_subsystem_filter(target)


class AddressFamilies(Collection[AddressFamily]):
    def addresses(self) -> Addresses:
        return Addresses(self._base_addresses())

    def _base_addresses(self) -> Iterable[Address]:
        for family in self:
            yield from family.addresses_to_target_adaptors

1	# Copyright 2015 Pants project contributors (see CONTRIBUTORS.md).
2	# Licensed under the Apache License, Version 2.0 (see LICENSE).
3
4	from __future__ import annotations	2✔
5
6	import os.path	2✔
7	from collections.abc import Iterable, Mapping	2✔
8	from dataclasses import dataclass	2✔
9	from typing import TypeVar	2✔
10
11	from pants.backend.project_info.filter_targets import FilterSubsystem	2✔
12	from pants.base.exceptions import MappingError	2✔
13	from pants.build_graph.address import Address, BuildFileAddress	2✔
14	from pants.engine.addresses import Addresses	2✔
15	from pants.engine.collection import Collection	2✔
16	from pants.engine.env_vars import EnvironmentVars	2✔
17	from pants.engine.internals.defaults import BuildFileDefaults, BuildFileDefaultsParserState	2✔
18	from pants.engine.internals.dep_rules import (	2✔
19	BuildFileDependencyRules,
20	BuildFileDependencyRulesParserState,
21	)
22	from pants.engine.internals.parser import BuildFilePreludeSymbols, Parser	2✔
23	from pants.engine.internals.target_adaptor import TargetAdaptor	2✔
24	from pants.engine.target import RegisteredTargetTypes, Tags, Target	2✔
25	from pants.util.filtering import TargetFilter, and_filters, create_filters	2✔
26	from pants.util.frozendict import FrozenDict	2✔
27	from pants.util.memo import memoized_property	2✔
28
29
30	class DuplicateNameError(MappingError):	2✔
31	"""Indicates more than one top-level object was found with the same name."""
32
33
34	AddressMapT = TypeVar("AddressMapT", bound="AddressMap")	2✔
35
36
37	# A pseudo-spec to refer to deleted paths.
38	DELETED_SPEC_PATH = "__deleted"	2✔
39	# A pseudo-target type to refer to deleted paths.
40	DELETED_TARGET_TYPE = "__deleted_target"	2✔
41
42
43	@dataclass(frozen=True)	2✔
44	class AddressMap:	2✔
45	"""Maps target adaptors from a byte source."""
46
47	path: str	2✔
48	name_to_target_adaptor: FrozenDict[str, TargetAdaptor]	2✔
49
50	@classmethod	2✔
51	def parse(	2✔
52	cls,
53	filepath: str,
54	build_file_content: str,
55	parser: Parser,
56	extra_symbols: BuildFilePreludeSymbols,
57	env_vars: EnvironmentVars,
58	is_bootstrap: bool,
59	defaults: BuildFileDefaultsParserState,
60	dependents_rules: BuildFileDependencyRulesParserState \| None,
61	dependencies_rules: BuildFileDependencyRulesParserState \| None,
62	) -> AddressMap:
63	"""Parses a source for targets.
64
65	The target adaptors are all 'thin': any targets they point to in other namespaces or even in
66	the same namespace but from a separate source are left as unresolved pointers.
67	"""
68	try:	2✔
69	target_adaptors = parser.parse(	2✔
70	filepath,
71	build_file_content,
72	extra_symbols,
73	env_vars,
74	is_bootstrap,
75	defaults,
76	dependents_rules,
77	dependencies_rules,
78	)
UNCOV 79	except Exception as e:	×
UNCOV 80	raise MappingError(f"Failed to parse ./{filepath}:\n{type(e).__name__}: {e}")	×
81	return cls.create(filepath, target_adaptors)	2✔
82
83	@classmethod	2✔
84	def create(	2✔
85	cls: type[AddressMapT], filepath: str, target_adaptors: Iterable[TargetAdaptor]
86	) -> AddressMapT:
87	name_to_target_adaptors: dict[str, TargetAdaptor] = {}	2✔
88	for target_adaptor in target_adaptors:	2✔
89	name = target_adaptor.name or os.path.basename(os.path.dirname(filepath))	2✔
90	if name in name_to_target_adaptors:	2✔
UNCOV 91	duplicate = name_to_target_adaptors[name]	×
UNCOV 92	raise DuplicateNameError(	×
93	f"A target already exists at {filepath!r} with name {name!r} and target type "
94	f"{duplicate.type_alias!r}. The {target_adaptor.type_alias!r} target "
95	"cannot use the same name."
96	)
97	name_to_target_adaptors[name] = target_adaptor	2✔
98	return cls(filepath, FrozenDict(sorted(name_to_target_adaptors.items())))	2✔
99
100
101	class DifferingFamiliesError(MappingError):	2✔
102	"""Indicates an attempt was made to merge address maps from different families together."""
103
104
105	@dataclass(frozen=True)	2✔
106	class AddressFamily:	2✔
107	"""Represents the family of target adaptors collected from the BUILD files in one directory.
108
109	To create an AddressFamily, use `create`.
110
111	An address family can be composed of the target adaptors from zero or more underlying address
112	sources. An "empty" AddressFamily is legal, and is the result when there are not build files in
113	a particular namespace.
114
115	:param namespace: The namespace path of this address family.
116	:param name_to_target_adaptors: A dict mapping from name to the target adaptor.
117	:param defaults: The default target field values, per target type, applicable for this address family.
118	:param dependents_rules: The rules to apply on incoming dependencies to targets in this family.
119	:param dependencies_rules: The rules to apply on the outgoing dependencies from targets in this family.
120	"""
121
122	# The directory from which the adaptors were parsed.
123	namespace: str	2✔
124	name_to_target_adaptors: dict[str, tuple[str, TargetAdaptor]]	2✔
125	defaults: BuildFileDefaults	2✔
126	dependents_rules: BuildFileDependencyRules \| None	2✔
127	dependencies_rules: BuildFileDependencyRules \| None	2✔
128
129	@classmethod	2✔
130	def create(	2✔
131	cls,
132	spec_path: str,
133	address_maps: Iterable[AddressMap],
134	defaults: BuildFileDefaults = BuildFileDefaults({}),
135	dependents_rules: BuildFileDependencyRules \| None = None,
136	dependencies_rules: BuildFileDependencyRules \| None = None,
137	) -> AddressFamily:
138	"""Creates an address family from the given set of address maps.
139
140	:param spec_path: The directory prefix shared by all address_maps.
141	:param address_maps: The family of maps that form this namespace.
142	:raises: :class:`MappingError` if the given address maps do not form a family.
143	"""
144	if spec_path == ".":	2✔
145	spec_path = ""	2✔
146	for address_map in address_maps:	2✔
147	if not address_map.path.startswith(spec_path):	2✔
UNCOV 148	raise DifferingFamiliesError(	×
149	f"Expected AddressMaps to share the same parent directory {spec_path!r}, "
150	f"but received: {address_map.path!r}"
151	)
152
153	name_to_target_adaptors: dict[str, tuple[str, TargetAdaptor]] = {}	2✔
154	for address_map in address_maps:	2✔
155	for name, target_adaptor in address_map.name_to_target_adaptor.items():	2✔
156	if name in name_to_target_adaptors:	2✔
UNCOV 157	previous_path, _ = name_to_target_adaptors[name]	×
UNCOV 158	raise DuplicateNameError(	×
159	f"A target with name {name!r} is already defined in {previous_path!r}, but "
160	f"is also defined in {address_map.path!r}. Because both targets share the "
161	f"same namespace of {spec_path!r}, this is not allowed."
162	)
163	name_to_target_adaptors[name] = (address_map.path, target_adaptor)	2✔
164	return AddressFamily(	2✔
165	namespace=spec_path,
166	name_to_target_adaptors=dict(sorted(name_to_target_adaptors.items())),
167	defaults=defaults,
168	dependents_rules=dependents_rules,
169	dependencies_rules=dependencies_rules,
170	)
171
172	@memoized_property	2✔
173	def addresses_to_target_adaptors(self) -> Mapping[Address, TargetAdaptor]:	2✔
174	return {	2✔
175	Address(spec_path=self.namespace, target_name=name): target_adaptor
176	for name, (_, target_adaptor) in self.name_to_target_adaptors.items()
177	}
178
179	@memoized_property	2✔
180	def build_file_addresses(self) -> tuple[BuildFileAddress, ...]:	2✔
181	return tuple(	2✔
182	BuildFileAddress(
183	rel_path=path, address=Address(spec_path=self.namespace, target_name=name)
184	)
185	for name, (path, _) in self.name_to_target_adaptors.items()
186	)
187
188	@property	2✔
189	def target_names(self) -> tuple[str, ...]:	2✔
UNCOV 190	return tuple(addr.target_name for addr in self.addresses_to_target_adaptors)	×
191
192	def get_target_adaptor(self, address: Address) -> TargetAdaptor \| None:	2✔
193	if self.namespace == DELETED_SPEC_PATH:	2✔
194	return TargetAdaptor(DELETED_TARGET_TYPE, "deleted_files", "deleted files")	×
195	assert address.spec_path == self.namespace	2✔
196	entry = self.name_to_target_adaptors.get(address.target_name)	2✔
197	if entry is None:	2✔
UNCOV 198	return None	×
199	_, target_adaptor = entry	2✔
200	return target_adaptor	2✔
201
202	def __hash__(self):	2✔
203	return hash((self.namespace, self.defaults))	×
204
205	def __repr__(self) -> str:	2✔
206	return (	×
207	f"AddressFamily(namespace={self.namespace!r}, "
208	f"name_to_target_adaptors={sorted(self.name_to_target_adaptors.keys())})"
209	)
210
211
212	@dataclass(frozen=True)	2✔
213	class SpecsFilter:	2✔
214	"""Filters targets with the `--tags`, `--exclude-target-regexp`, and `[filter]` subsystem
215	options."""
216
217	is_specified: bool	2✔
218	filter_subsystem_filter: TargetFilter	2✔
219	tags_filter: TargetFilter	2✔
220
221	@classmethod	2✔
222	def create(	2✔
223	cls,
224	filter_subsystem: FilterSubsystem,
225	registered_target_types: RegisteredTargetTypes,
226	*,
227	tags: Iterable[str],
228	) -> SpecsFilter:
229	def tags_outer_filter(tag: str) -> TargetFilter:	2✔
UNCOV 230	def tags_inner_filter(tgt: Target) -> bool:	×
UNCOV 231	return tag in (tgt.get(Tags).value or [])	×
232
UNCOV 233	return tags_inner_filter	×
234
235	tags_filter = and_filters(create_filters(tags, tags_outer_filter))	2✔
236
237	return SpecsFilter(	2✔
238	is_specified=bool(filter_subsystem.is_specified() or tags),
239	filter_subsystem_filter=filter_subsystem.all_filters(registered_target_types),
240	tags_filter=tags_filter,
241	)
242
243	def matches(self, target: Target) -> bool:	2✔
244	"""Check that the target matches the provided `--tag` and `--exclude-target-regexp`
245	options."""
246	return self.tags_filter(target) and self.filter_subsystem_filter(target)	2✔
247
248
249	class AddressFamilies(Collection[AddressFamily]):	2✔
250	def addresses(self) -> Addresses:	2✔
251	return Addresses(self._base_addresses())	2✔
252
253	def _base_addresses(self) -> Iterable[Address]:	2✔
254	for family in self:	2✔
255	yield from family.addresses_to_target_adaptors	2✔

pantsbuild / pants / 26080722777

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