19015773527

Committed 02 Nov 2025 05:33PM UTC coverage: 17.872% (-62.4%) from 80.3%

Build # 19015773527

Build Type

Pull #22816

github

Committed by

web-flow

Commit Message

Merge a12d75757 into 6c024e162

Pull Request Pull Request #22816: Update Pants internal Python to 3.14

Run Details

4 of 5 new or added lines in 3 files covered. (80.0%)

28452 existing lines in 683 files now uncovered.

9831 of 55007 relevant lines covered (17.87%)

0.18 hits per line

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

55.66

/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")


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

pantsbuild / pants / 19015773527

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