25443604553

Committed 06 May 2026 03:05PM UTC coverage: 92.879% (-0.04%) from 92.915%

Build # 25443604553

Build Type

push

github

Committed by

web-flow

Commit Message

[pants_ng] Scaffolding for a pants_ng mode. (#23319)

In this mode the command line is parsed as an
NG invocation, and dispatched appropriately.

Of course at the moment there are no
implementations to dispatch to. That will follow.

This does expose a new option, `pants_ng` to users. 
There is a big warning not to set it, but we're not trying
to hide that we're working on a new thing, so I am
comfortable with this.

Coverage Stats

25 of 76 new or added lines in 9 files covered. (32.89%)

1294 existing lines in 76 files now uncovered.

92234 of 99306 relevant lines covered (92.88%)

4.05 hits per line

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

93.59

/src/python/pants/build_graph/build_configuration.py

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

from __future__ import annotations

import logging
from collections import defaultdict
from collections.abc import Callable, Iterable
from dataclasses import dataclass, field
from enum import Enum
from typing import Any, DefaultDict

from pants.backend.project_info.filter_targets import FilterSubsystem
from pants.build_graph.build_file_aliases import BuildFileAliases
from pants.core.environments.subsystems import EnvironmentsSubsystem
from pants.engine.goal import GoalSubsystem
from pants.engine.rules import Rule, RuleIndex, collect_rules, rule
from pants.engine.target import Target
from pants.engine.unions import UnionRule
from pants.ng.goal import GoalSubsystemNg
from pants.option.alias import CliOptions
from pants.option.global_options import GlobalOptions
from pants.option.scope import OptionsParsingSettings, ScopeInfo, normalize_scope
from pants.option.subsystem import Subsystem
from pants.util.frozendict import FrozenDict
from pants.util.ordered_set import FrozenOrderedSet
from pants.vcs.changed import Changed

logger = logging.getLogger(__name__)


# No goal or target_type can have a name from this set, so that `./pants help <name>`
# is unambiguous.
_RESERVED_NAMES = {
    "api-types",
    "backends",
    "global",
    "goals",
    "subsystems",
    "symbols",
    "targets",
    "tools",
}


# Subsystems used outside of any rule.
_GLOBAL_SUBSYSTEMS: set[type[Subsystem]] = {
    GlobalOptions,
    Changed,
    CliOptions,
    FilterSubsystem,
    EnvironmentsSubsystem,
}


@dataclass(frozen=True)
class BuildConfiguration:
    """Stores the types and helper functions exposed to BUILD files."""

    registered_aliases: BuildFileAliases
    subsystem_to_providers: FrozenDict[type[Subsystem], tuple[str, ...]]
    target_type_to_providers: FrozenDict[type[Target], tuple[str, ...]]
    rule_to_providers: FrozenDict[Rule, tuple[str, ...]]
    union_rule_to_providers: FrozenDict[UnionRule, tuple[str, ...]]
    allow_unknown_options: bool
    remote_auth_plugin_func: Callable | None

    @property
    def all_subsystems(self) -> tuple[type[Subsystem], ...]:
        """Return all subsystems in the system: global and those registered via rule usage."""
        # Sort by options_scope, for consistency.
        return tuple(
            sorted(
                _GLOBAL_SUBSYSTEMS | self.subsystem_to_providers.keys(),
                key=lambda x: x.options_scope,
            )
        )

    @property
    def known_scope_infos(self) -> tuple[ScopeInfo, ...]:
        return tuple(subsystem.get_scope_info() for subsystem in self.all_subsystems)

    @property
    def target_types(self) -> tuple[type[Target], ...]:
        return tuple(sorted(self.target_type_to_providers.keys(), key=lambda x: x.alias))

    @property
    def rules(self) -> FrozenOrderedSet[Rule]:
        return FrozenOrderedSet(self.rule_to_providers.keys())

    @property
    def union_rules(self) -> FrozenOrderedSet[UnionRule]:
        return FrozenOrderedSet(self.union_rule_to_providers.keys())

    def __post_init__(self) -> None:
        class Category(Enum):
            goal = "goal"
            reserved_name = "reserved name"
            subsystem = "subsystem"
            target_type = "target type"

        name_to_categories: DefaultDict[str, set[Category]] = defaultdict(set)
        normalized_to_orig_name: dict[str, str] = {}

        for opt in self.all_subsystems:
            scope = opt.options_scope
            normalized_scope = normalize_scope(scope)
            name_to_categories[normalized_scope].add(
                Category.goal if issubclass(opt, GoalSubsystem) else Category.subsystem
            )
            normalized_to_orig_name[normalized_scope] = scope
        for tgt_type in self.target_types:
            name_to_categories[normalize_scope(tgt_type.alias)].add(Category.target_type)
        for reserved_name in _RESERVED_NAMES:
            name_to_categories[normalize_scope(reserved_name)].add(Category.reserved_name)

        found_collision = False
        for name, cats in name_to_categories.items():
            if len(cats) > 1:
                scats = sorted(cat.value for cat in cats)
                cats_str = ", ".join(f"a {cat}" for cat in scats[:-1]) + f" and a {scats[-1]}."
                colliding_names = "`/`".join(
                    sorted({name, normalized_to_orig_name.get(name, name)})
                )
                logger.error(f"Naming collision: `{colliding_names}` is registered as {cats_str}")
                found_collision = True

        if found_collision:
            raise TypeError("Found naming collisions. See log for details.")

    @dataclass
    class Builder:
        _exposed_object_by_alias: dict[Any, Any] = field(default_factory=dict)
        _exposed_context_aware_object_factory_by_alias: dict[Any, Any] = field(default_factory=dict)
        _subsystem_to_providers: dict[type[Subsystem], list[str]] = field(
            default_factory=lambda: defaultdict(list)
        )
        _target_type_to_providers: dict[type[Target], list[str]] = field(
            default_factory=lambda: defaultdict(list)
        )
        _rule_to_providers: dict[Rule, list[str]] = field(default_factory=lambda: defaultdict(list))
        _union_rule_to_providers: dict[UnionRule, list[str]] = field(
            default_factory=lambda: defaultdict(list)
        )
        _allow_unknown_options: bool = False
        _remote_auth_plugin: Callable | None = None
        _pants_ng: bool = False

        def registered_aliases(self) -> BuildFileAliases:
            """Return the registered aliases exposed in BUILD files.

            These returned aliases aren't so useful for actually parsing BUILD files.
            They are useful for generating online documentation.

            :returns: A new BuildFileAliases instance containing this BuildConfiguration's
                      registered alias mappings.
            """
            return BuildFileAliases(
                objects=self._exposed_object_by_alias.copy(),
                context_aware_object_factories=self._exposed_context_aware_object_factory_by_alias.copy(),
            )

        def register_aliases(self, aliases):
            """Registers the given aliases to be exposed in parsed BUILD files.

            :param aliases: The BuildFileAliases to register.
            :type aliases: :class:`pants.build_graph.build_file_aliases.BuildFileAliases`
            """
            if not isinstance(aliases, BuildFileAliases):
                raise TypeError(f"The aliases must be a BuildFileAliases, given {aliases}")

            for alias, obj in aliases.objects.items():
                self._register_exposed_object(alias, obj)

            for (
                alias,
                context_aware_object_factory,
            ) in aliases.context_aware_object_factories.items():
                self._register_exposed_context_aware_object_factory(
                    alias, context_aware_object_factory
                )

        def _register_exposed_object(self, alias, obj):
            if alias in self._exposed_object_by_alias:
                logger.debug(f"Object alias {alias} has already been registered. Overwriting!")

            self._exposed_object_by_alias[alias] = obj

        def _register_exposed_context_aware_object_factory(
            self, alias, context_aware_object_factory
        ):
            if alias in self._exposed_context_aware_object_factory_by_alias:
                logger.debug(
                    f"This context aware object factory alias {alias} has already been registered. "
                    "Overwriting!"
                )

            self._exposed_context_aware_object_factory_by_alias[alias] = (
                context_aware_object_factory
            )

        def register_subsystems(
            self, plugin_or_backend: str, subsystems: Iterable[type[Subsystem]]
        ):
            """Registers the given subsystem types."""
            if not isinstance(subsystems, Iterable):
                raise TypeError(f"The subsystems must be an iterable, given {subsystems}")
            subsystems = tuple(subsystems)
            if not subsystems:
                return

            invalid_subsystems = [
                s for s in subsystems if not isinstance(s, type) or not issubclass(s, Subsystem)
            ]
            if invalid_subsystems:
                raise TypeError(
                    "The following items from the given subsystems are not Subsystems "
                    "subclasses:\n\t{}".format("\n\t".join(str(i) for i in invalid_subsystems))
                )

            for subsystem in subsystems:
                if (
                    issubclass(subsystem, GoalSubsystem)
                    and issubclass(subsystem, GoalSubsystemNg) != self._pants_ng
                ):
                    # Don't register goal subsystems that aren't pertinent to our og/ng state.
                    # This allows GoalSubsystemNgs to use the same scopes as og GoalSubsystems
                    # without collision. Note that the @rules that consume these subsystems
                    # can still be registered, but the subsystems cannot be configured.
                    continue
                self._subsystem_to_providers[subsystem].append(plugin_or_backend)

        def register_rules(self, plugin_or_backend: str, rules: Iterable[Rule | UnionRule]):
            """Registers the given rules."""
            if not isinstance(rules, Iterable):
                raise TypeError(f"The rules must be an iterable, given {rules!r}")

            # "Index" the rules to normalize them and expand their dependencies.
            rule_index = RuleIndex.create(rules)
            rules_and_queries: tuple[Rule, ...] = (*rule_index.rules, *rule_index.queries)
            for _rule in rules_and_queries:
                self._rule_to_providers[_rule].append(plugin_or_backend)
            for union_rule in rule_index.union_rules:
                self._union_rule_to_providers[union_rule].append(plugin_or_backend)
            self.register_subsystems(
                plugin_or_backend,
                (
                    _rule.output_type
                    for _rule in rules_and_queries
                    if issubclass(_rule.output_type, Subsystem)
                ),
            )

        # NB: We expect the parameter to be Iterable[Type[Target]], but we can't be confident in
        # this because we pass whatever people put in their `register.py`s to this function;
        # I.e., this is an impure function that reads from the outside world. So, we use the type
        # hint `Any` and perform runtime type checking.
        def register_target_types(
            self, plugin_or_backend: str, target_types: Iterable[type[Target]] | Any
        ) -> None:
            """Registers the given target types."""
            if not isinstance(target_types, Iterable):
                raise TypeError(
                    f"The entrypoint `target_types` must return an iterable. "
                    f"Given {repr(target_types)}"
                )
            bad_elements = [
                tgt_type
                for tgt_type in target_types
                if not isinstance(tgt_type, type) or not issubclass(tgt_type, Target)
            ]
            if bad_elements:
                raise TypeError(
                    "Every element of the entrypoint `target_types` must be a subclass of "
                    f"{Target.__name__}. Bad elements: {bad_elements}."
                )
            for target_type in target_types:
                self._target_type_to_providers[target_type].append(plugin_or_backend)
                # Access the Target.PluginField here to ensure the PluginField class is
                # created before the UnionMembership is instantiated, as the class hierarchy is
                # walked during union membership setup.
                _ = target_type.PluginField

        def register_remote_auth_plugin(self, remote_auth_plugin: Callable) -> None:
            self._remote_auth_plugin = remote_auth_plugin

        def register_auxiliary_goals(self, plugin_or_backend: str, auxiliary_goals: Iterable[type]):
            """Registers the given auxiliary goals."""
            if not isinstance(auxiliary_goals, Iterable):
                raise TypeError(
                    f"The entrypoint `auxiliary_goals` must return an iterable. "
                    f"Given {repr(auxiliary_goals)}"
                )
            # Import `AuxiliaryGoal` here to avoid import cycle.
            from pants.goal.auxiliary_goal import AuxiliaryGoal

            bad_elements = [goal for goal in auxiliary_goals if not issubclass(goal, AuxiliaryGoal)]
            if bad_elements:
                raise TypeError(
                    "Every element of the entrypoint `auxiliary_goals` must be a subclass of "
                    f"{AuxiliaryGoal.__name__}. Bad elements: {bad_elements}."
                )
            self.register_subsystems(plugin_or_backend, auxiliary_goals)

        def allow_unknown_options(self, allow: bool = True) -> None:
            """Allows overriding whether Options parsing will fail for unrecognized Options.

            Used to defer options failures while bootstrapping BuildConfiguration until after the
            complete set of plugins is known.
            """
            self._allow_unknown_options = True

        def create(self) -> BuildConfiguration:
            registered_aliases = BuildFileAliases(
                objects=self._exposed_object_by_alias.copy(),
                context_aware_object_factories=self._exposed_context_aware_object_factory_by_alias.copy(),
            )
            return BuildConfiguration(
                registered_aliases=registered_aliases,
                subsystem_to_providers=FrozenDict(
                    (k, tuple(v)) for k, v in self._subsystem_to_providers.items()
                ),
                target_type_to_providers=FrozenDict(
                    (k, tuple(v)) for k, v in self._target_type_to_providers.items()
                ),
                rule_to_providers=FrozenDict(
                    (k, tuple(v)) for k, v in self._rule_to_providers.items()
                ),
                union_rule_to_providers=FrozenDict(
                    (k, tuple(v)) for k, v in self._union_rule_to_providers.items()
                ),
                allow_unknown_options=self._allow_unknown_options,
                remote_auth_plugin_func=self._remote_auth_plugin,
            )


@rule
async def get_full_options_parsing_settings(
    build_config: BuildConfiguration,
) -> OptionsParsingSettings:
    return OptionsParsingSettings(
        build_config.known_scope_infos,
        build_config.allow_unknown_options,
    )


def rules():
    return collect_rules()

1	# Copyright 2014 Pants project contributors (see CONTRIBUTORS.md).
2	# Licensed under the Apache License, Version 2.0 (see LICENSE).
3
4	from __future__ import annotations	12✔
5
6	import logging	12✔
7	from collections import defaultdict	12✔
8	from collections.abc import Callable, Iterable	12✔
9	from dataclasses import dataclass, field	12✔
10	from enum import Enum	12✔
11	from typing import Any, DefaultDict	12✔
12
13	from pants.backend.project_info.filter_targets import FilterSubsystem	12✔
14	from pants.build_graph.build_file_aliases import BuildFileAliases	12✔
15	from pants.core.environments.subsystems import EnvironmentsSubsystem	12✔
16	from pants.engine.goal import GoalSubsystem	12✔
17	from pants.engine.rules import Rule, RuleIndex, collect_rules, rule	12✔
18	from pants.engine.target import Target	12✔
19	from pants.engine.unions import UnionRule	12✔
20	from pants.ng.goal import GoalSubsystemNg	12✔
21	from pants.option.alias import CliOptions	12✔
22	from pants.option.global_options import GlobalOptions	12✔
23	from pants.option.scope import OptionsParsingSettings, ScopeInfo, normalize_scope	12✔
24	from pants.option.subsystem import Subsystem	12✔
25	from pants.util.frozendict import FrozenDict	12✔
26	from pants.util.ordered_set import FrozenOrderedSet	12✔
27	from pants.vcs.changed import Changed	12✔
28
29	logger = logging.getLogger(__name__)	12✔
30
31
32	# No goal or target_type can have a name from this set, so that `./pants help <name>`
33	# is unambiguous.
34	_RESERVED_NAMES = {	12✔
35	"api-types",
36	"backends",
37	"global",
38	"goals",
39	"subsystems",
40	"symbols",
41	"targets",
42	"tools",
43	}
44
45
46	# Subsystems used outside of any rule.
47	_GLOBAL_SUBSYSTEMS: set[type[Subsystem]] = {	12✔
48	GlobalOptions,
49	Changed,
50	CliOptions,
51	FilterSubsystem,
52	EnvironmentsSubsystem,
53	}
54
55
56	@dataclass(frozen=True)	12✔
57	class BuildConfiguration:	12✔
58	"""Stores the types and helper functions exposed to BUILD files."""
59
60	registered_aliases: BuildFileAliases	12✔
61	subsystem_to_providers: FrozenDict[type[Subsystem], tuple[str, ...]]	12✔
62	target_type_to_providers: FrozenDict[type[Target], tuple[str, ...]]	12✔
63	rule_to_providers: FrozenDict[Rule, tuple[str, ...]]	12✔
64	union_rule_to_providers: FrozenDict[UnionRule, tuple[str, ...]]	12✔
65	allow_unknown_options: bool	12✔
66	remote_auth_plugin_func: Callable \| None	12✔
67
68	@property	12✔
69	def all_subsystems(self) -> tuple[type[Subsystem], ...]:	12✔
70	"""Return all subsystems in the system: global and those registered via rule usage."""
71	# Sort by options_scope, for consistency.
72	return tuple(	12✔
73	sorted(
74	_GLOBAL_SUBSYSTEMS \| self.subsystem_to_providers.keys(),
75	key=lambda x: x.options_scope,
76	)
77	)
78
79	@property	12✔
80	def known_scope_infos(self) -> tuple[ScopeInfo, ...]:	12✔
81	return tuple(subsystem.get_scope_info() for subsystem in self.all_subsystems)	12✔
82
83	@property	12✔
84	def target_types(self) -> tuple[type[Target], ...]:	12✔
85	return tuple(sorted(self.target_type_to_providers.keys(), key=lambda x: x.alias))	12✔
86
87	@property	12✔
88	def rules(self) -> FrozenOrderedSet[Rule]:	12✔
89	return FrozenOrderedSet(self.rule_to_providers.keys())	12✔
90
91	@property	12✔
92	def union_rules(self) -> FrozenOrderedSet[UnionRule]:	12✔
93	return FrozenOrderedSet(self.union_rule_to_providers.keys())	12✔
94
95	def __post_init__(self) -> None:	12✔
96	class Category(Enum):	12✔
97	goal = "goal"	12✔
98	reserved_name = "reserved name"	12✔
99	subsystem = "subsystem"	12✔
100	target_type = "target type"	12✔
101
102	name_to_categories: DefaultDict[str, set[Category]] = defaultdict(set)	12✔
103	normalized_to_orig_name: dict[str, str] = {}	12✔
104
105	for opt in self.all_subsystems:	12✔
106	scope = opt.options_scope	12✔
107	normalized_scope = normalize_scope(scope)	12✔
108	name_to_categories[normalized_scope].add(	12✔
109	Category.goal if issubclass(opt, GoalSubsystem) else Category.subsystem
110	)
111	normalized_to_orig_name[normalized_scope] = scope	12✔
112	for tgt_type in self.target_types:	12✔
113	name_to_categories[normalize_scope(tgt_type.alias)].add(Category.target_type)	12✔
114	for reserved_name in _RESERVED_NAMES:	12✔
115	name_to_categories[normalize_scope(reserved_name)].add(Category.reserved_name)	12✔
116
117	found_collision = False	12✔
118	for name, cats in name_to_categories.items():	12✔
119	if len(cats) > 1:	12✔
120	scats = sorted(cat.value for cat in cats)	1✔
121	cats_str = ", ".join(f"a {cat}" for cat in scats[:-1]) + f" and a {scats[-1]}."	1✔
122	colliding_names = "`/`".join(	1✔
123	sorted({name, normalized_to_orig_name.get(name, name)})
124	)
125	logger.error(f"Naming collision: `{colliding_names}` is registered as {cats_str}")	1✔
126	found_collision = True	1✔
127
128	if found_collision:	12✔
129	raise TypeError("Found naming collisions. See log for details.")	1✔
130
131	@dataclass	12✔
132	class Builder:	12✔
133	_exposed_object_by_alias: dict[Any, Any] = field(default_factory=dict)	12✔
134	_exposed_context_aware_object_factory_by_alias: dict[Any, Any] = field(default_factory=dict)	12✔
135	_subsystem_to_providers: dict[type[Subsystem], list[str]] = field(	12✔
136	default_factory=lambda: defaultdict(list)
137	)
138	_target_type_to_providers: dict[type[Target], list[str]] = field(	12✔
139	default_factory=lambda: defaultdict(list)
140	)
141	_rule_to_providers: dict[Rule, list[str]] = field(default_factory=lambda: defaultdict(list))	12✔
142	_union_rule_to_providers: dict[UnionRule, list[str]] = field(	12✔
143	default_factory=lambda: defaultdict(list)
144	)
145	_allow_unknown_options: bool = False	12✔
146	_remote_auth_plugin: Callable \| None = None	12✔
147	_pants_ng: bool = False	12✔
148
149	def registered_aliases(self) -> BuildFileAliases:	12✔
150	"""Return the registered aliases exposed in BUILD files.
151
152	These returned aliases aren't so useful for actually parsing BUILD files.
153	They are useful for generating online documentation.
154
155	:returns: A new BuildFileAliases instance containing this BuildConfiguration's
156	registered alias mappings.
157	"""
158	return BuildFileAliases(	×
159	objects=self._exposed_object_by_alias.copy(),
160	context_aware_object_factories=self._exposed_context_aware_object_factory_by_alias.copy(),
161	)
162
163	def register_aliases(self, aliases):	12✔
164	"""Registers the given aliases to be exposed in parsed BUILD files.
165
166	:param aliases: The BuildFileAliases to register.
167	:type aliases: :class:`pants.build_graph.build_file_aliases.BuildFileAliases`
168	"""
169	if not isinstance(aliases, BuildFileAliases):	12✔
170	raise TypeError(f"The aliases must be a BuildFileAliases, given {aliases}")	1✔
171
172	for alias, obj in aliases.objects.items():	12✔
173	self._register_exposed_object(alias, obj)	12✔
174
175	for (	12✔
176	alias,
177	context_aware_object_factory,
178	) in aliases.context_aware_object_factories.items():
179	self._register_exposed_context_aware_object_factory(	1✔
180	alias, context_aware_object_factory
181	)
182
183	def _register_exposed_object(self, alias, obj):	12✔
184	if alias in self._exposed_object_by_alias:	12✔
UNCOV 185	logger.debug(f"Object alias {alias} has already been registered. Overwriting!")	1✔
186
187	self._exposed_object_by_alias[alias] = obj	12✔
188
189	def _register_exposed_context_aware_object_factory(	12✔
190	self, alias, context_aware_object_factory
191	):
192	if alias in self._exposed_context_aware_object_factory_by_alias:	1✔
193	logger.debug(	×
194	f"This context aware object factory alias {alias} has already been registered. "
195	"Overwriting!"
196	)
197
198	self._exposed_context_aware_object_factory_by_alias[alias] = (	1✔
199	context_aware_object_factory
200	)
201
202	def register_subsystems(	12✔
203	self, plugin_or_backend: str, subsystems: Iterable[type[Subsystem]]
204	):
205	"""Registers the given subsystem types."""
206	if not isinstance(subsystems, Iterable):	12✔
207	raise TypeError(f"The subsystems must be an iterable, given {subsystems}")	×
208	subsystems = tuple(subsystems)	12✔
209	if not subsystems:	12✔
210	return	12✔
211
212	invalid_subsystems = [	12✔
213	s for s in subsystems if not isinstance(s, type) or not issubclass(s, Subsystem)
214	]
215	if invalid_subsystems:	12✔
216	raise TypeError(	×
217	"The following items from the given subsystems are not Subsystems "
218	"subclasses:\n\t{}".format("\n\t".join(str(i) for i in invalid_subsystems))
219	)
220
221	for subsystem in subsystems:	12✔
222	if (	12✔
223	issubclass(subsystem, GoalSubsystem)
224	and issubclass(subsystem, GoalSubsystemNg) != self._pants_ng
225	):
226	# Don't register goal subsystems that aren't pertinent to our og/ng state.
227	# This allows GoalSubsystemNgs to use the same scopes as og GoalSubsystems
228	# without collision. Note that the @rules that consume these subsystems
229	# can still be registered, but the subsystems cannot be configured.
NEW 230	continue	×
231	self._subsystem_to_providers[subsystem].append(plugin_or_backend)	12✔
232
233	def register_rules(self, plugin_or_backend: str, rules: Iterable[Rule \| UnionRule]):	12✔
234	"""Registers the given rules."""
235	if not isinstance(rules, Iterable):	12✔
236	raise TypeError(f"The rules must be an iterable, given {rules!r}")	×
237
238	# "Index" the rules to normalize them and expand their dependencies.
239	rule_index = RuleIndex.create(rules)	12✔
240	rules_and_queries: tuple[Rule, ...] = (rule_index.rules, rule_index.queries)	12✔
241	for _rule in rules_and_queries:	12✔
242	self._rule_to_providers[_rule].append(plugin_or_backend)	12✔
243	for union_rule in rule_index.union_rules:	12✔
244	self._union_rule_to_providers[union_rule].append(plugin_or_backend)	12✔
245	self.register_subsystems(	12✔
246	plugin_or_backend,
247	(
248	_rule.output_type
249	for _rule in rules_and_queries
250	if issubclass(_rule.output_type, Subsystem)
251	),
252	)
253
254	# NB: We expect the parameter to be Iterable[Type[Target]], but we can't be confident in
255	# this because we pass whatever people put in their `register.py`s to this function;
256	# I.e., this is an impure function that reads from the outside world. So, we use the type
257	# hint `Any` and perform runtime type checking.
258	def register_target_types(	12✔
259	self, plugin_or_backend: str, target_types: Iterable[type[Target]] \| Any
260	) -> None:
261	"""Registers the given target types."""
262	if not isinstance(target_types, Iterable):	12✔
263	raise TypeError(	×
264	f"The entrypoint `target_types` must return an iterable. "
265	f"Given {repr(target_types)}"
266	)
267	bad_elements = [	12✔
268	tgt_type
269	for tgt_type in target_types
270	if not isinstance(tgt_type, type) or not issubclass(tgt_type, Target)
271	]
272	if bad_elements:	12✔
273	raise TypeError(	×
274	"Every element of the entrypoint `target_types` must be a subclass of "
275	f"{Target.__name__}. Bad elements: {bad_elements}."
276	)
277	for target_type in target_types:	12✔
278	self._target_type_to_providers[target_type].append(plugin_or_backend)	12✔
279	# Access the Target.PluginField here to ensure the PluginField class is
280	# created before the UnionMembership is instantiated, as the class hierarchy is
281	# walked during union membership setup.
282	_ = target_type.PluginField	12✔
283
284	def register_remote_auth_plugin(self, remote_auth_plugin: Callable) -> None:	12✔
285	self._remote_auth_plugin = remote_auth_plugin	1✔
286
287	def register_auxiliary_goals(self, plugin_or_backend: str, auxiliary_goals: Iterable[type]):	12✔
288	"""Registers the given auxiliary goals."""
289	if not isinstance(auxiliary_goals, Iterable):	12✔
290	raise TypeError(	×
291	f"The entrypoint `auxiliary_goals` must return an iterable. "
292	f"Given {repr(auxiliary_goals)}"
293	)
294	# Import `AuxiliaryGoal` here to avoid import cycle.
295	from pants.goal.auxiliary_goal import AuxiliaryGoal	12✔
296
297	bad_elements = [goal for goal in auxiliary_goals if not issubclass(goal, AuxiliaryGoal)]	12✔
298	if bad_elements:	12✔
299	raise TypeError(	×
300	"Every element of the entrypoint `auxiliary_goals` must be a subclass of "
301	f"{AuxiliaryGoal.__name__}. Bad elements: {bad_elements}."
302	)
303	self.register_subsystems(plugin_or_backend, auxiliary_goals)	12✔
304
305	def allow_unknown_options(self, allow: bool = True) -> None:	12✔
306	"""Allows overriding whether Options parsing will fail for unrecognized Options.
307
308	Used to defer options failures while bootstrapping BuildConfiguration until after the
309	complete set of plugins is known.
310	"""
311	self._allow_unknown_options = True	4✔
312
313	def create(self) -> BuildConfiguration:	12✔
314	registered_aliases = BuildFileAliases(	12✔
315	objects=self._exposed_object_by_alias.copy(),
316	context_aware_object_factories=self._exposed_context_aware_object_factory_by_alias.copy(),
317	)
318	return BuildConfiguration(	12✔
319	registered_aliases=registered_aliases,
320	subsystem_to_providers=FrozenDict(
321	(k, tuple(v)) for k, v in self._subsystem_to_providers.items()
322	),
323	target_type_to_providers=FrozenDict(
324	(k, tuple(v)) for k, v in self._target_type_to_providers.items()
325	),
326	rule_to_providers=FrozenDict(
327	(k, tuple(v)) for k, v in self._rule_to_providers.items()
328	),
329	union_rule_to_providers=FrozenDict(
330	(k, tuple(v)) for k, v in self._union_rule_to_providers.items()
331	),
332	allow_unknown_options=self._allow_unknown_options,
333	remote_auth_plugin_func=self._remote_auth_plugin,
334	)
335
336
337	@rule	12✔
338	async def get_full_options_parsing_settings(	12✔
339	build_config: BuildConfiguration,
340	) -> OptionsParsingSettings:
341	return OptionsParsingSettings(	12✔
342	build_config.known_scope_infos,
343	build_config.allow_unknown_options,
344	)
345
346
347	def rules():	12✔
348	return collect_rules()	12✔

pantsbuild / pants / 25443604553

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