• Home
  • Features
  • Pricing
  • Docs
  • Announcements
  • Sign In

pantsbuild / pants / 18517631058

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

Pull #22745

github

web-flow
Merge 642a76ca1 into 99919310e
Pull Request #22745: [windows] Add windows support in the stdio crate.

53815 of 77759 relevant lines covered (69.21%)

2.42 hits per line

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

86.18
/src/python/pants/build_graph/build_configuration.py
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
7✔
5

6
import logging
7✔
7
from collections import defaultdict
7✔
8
from collections.abc import Callable, Iterable
7✔
9
from dataclasses import dataclass, field
7✔
10
from enum import Enum
7✔
11
from typing import Any, DefaultDict
7✔
12

13
from pants.backend.project_info.filter_targets import FilterSubsystem
7✔
14
from pants.build_graph.build_file_aliases import BuildFileAliases
7✔
15
from pants.core.environments.subsystems import EnvironmentsSubsystem
7✔
16
from pants.engine.goal import GoalSubsystem
7✔
17
from pants.engine.rules import Rule, RuleIndex, collect_rules, rule
7✔
18
from pants.engine.target import Target
7✔
19
from pants.engine.unions import UnionRule
7✔
20
from pants.option.alias import CliOptions
7✔
21
from pants.option.global_options import GlobalOptions
7✔
22
from pants.option.scope import OptionsParsingSettings, ScopeInfo, normalize_scope
7✔
23
from pants.option.subsystem import Subsystem
7✔
24
from pants.util.frozendict import FrozenDict
7✔
25
from pants.util.ordered_set import FrozenOrderedSet
7✔
26
from pants.vcs.changed import Changed
7✔
27

28
logger = logging.getLogger(__name__)
7✔
29

30

31
# No goal or target_type can have a name from this set, so that `./pants help <name>`
32
# is unambiguous.
33
_RESERVED_NAMES = {
7✔
34
    "api-types",
35
    "backends",
36
    "global",
37
    "goals",
38
    "subsystems",
39
    "symbols",
40
    "targets",
41
    "tools",
42
}
43

44

45
# Subsystems used outside of any rule.
46
_GLOBAL_SUBSYSTEMS: set[type[Subsystem]] = {
7✔
47
    GlobalOptions,
48
    Changed,
49
    CliOptions,
50
    FilterSubsystem,
51
    EnvironmentsSubsystem,
52
}
53

54

55
@dataclass(frozen=True)
7✔
56
class BuildConfiguration:
7✔
57
    """Stores the types and helper functions exposed to BUILD files."""
58

59
    registered_aliases: BuildFileAliases
7✔
60
    subsystem_to_providers: FrozenDict[type[Subsystem], tuple[str, ...]]
7✔
61
    target_type_to_providers: FrozenDict[type[Target], tuple[str, ...]]
7✔
62
    rule_to_providers: FrozenDict[Rule, tuple[str, ...]]
7✔
63
    union_rule_to_providers: FrozenDict[UnionRule, tuple[str, ...]]
7✔
64
    allow_unknown_options: bool
7✔
65
    remote_auth_plugin_func: Callable | None
7✔
66

67
    @property
7✔
68
    def all_subsystems(self) -> tuple[type[Subsystem], ...]:
7✔
69
        """Return all subsystems in the system: global and those registered via rule usage."""
70
        # Sort by options_scope, for consistency.
71
        return tuple(
7✔
72
            sorted(
73
                _GLOBAL_SUBSYSTEMS | self.subsystem_to_providers.keys(),
74
                key=lambda x: x.options_scope,
75
            )
76
        )
77

78
    @property
7✔
79
    def known_scope_infos(self) -> tuple[ScopeInfo, ...]:
7✔
80
        return tuple(subsystem.get_scope_info() for subsystem in self.all_subsystems)
7✔
81

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

86
    @property
7✔
87
    def rules(self) -> FrozenOrderedSet[Rule]:
7✔
88
        return FrozenOrderedSet(self.rule_to_providers.keys())
7✔
89

90
    @property
7✔
91
    def union_rules(self) -> FrozenOrderedSet[UnionRule]:
7✔
92
        return FrozenOrderedSet(self.union_rule_to_providers.keys())
7✔
93

94
    def __post_init__(self) -> None:
7✔
95
        class Category(Enum):
7✔
96
            goal = "goal"
7✔
97
            reserved_name = "reserved name"
7✔
98
            subsystem = "subsystem"
7✔
99
            target_type = "target type"
7✔
100

101
        name_to_categories: DefaultDict[str, set[Category]] = defaultdict(set)
7✔
102
        normalized_to_orig_name: dict[str, str] = {}
7✔
103

104
        for opt in self.all_subsystems:
7✔
105
            scope = opt.options_scope
7✔
106
            normalized_scope = normalize_scope(scope)
7✔
107
            name_to_categories[normalized_scope].add(
7✔
108
                Category.goal if issubclass(opt, GoalSubsystem) else Category.subsystem
109
            )
110
            normalized_to_orig_name[normalized_scope] = scope
7✔
111
        for tgt_type in self.target_types:
7✔
112
            name_to_categories[normalize_scope(tgt_type.alias)].add(Category.target_type)
7✔
113
        for reserved_name in _RESERVED_NAMES:
7✔
114
            name_to_categories[normalize_scope(reserved_name)].add(Category.reserved_name)
7✔
115

116
        found_collision = False
7✔
117
        for name, cats in name_to_categories.items():
7✔
118
            if len(cats) > 1:
7✔
119
                scats = sorted(cat.value for cat in cats)
×
120
                cats_str = ", ".join(f"a {cat}" for cat in scats[:-1]) + f" and a {scats[-1]}."
×
121
                colliding_names = "`/`".join(
×
122
                    sorted({name, normalized_to_orig_name.get(name, name)})
123
                )
124
                logger.error(f"Naming collision: `{colliding_names}` is registered as {cats_str}")
×
125
                found_collision = True
×
126

127
        if found_collision:
7✔
128
            raise TypeError("Found naming collisions. See log for details.")
×
129

130
    @dataclass
7✔
131
    class Builder:
7✔
132
        _exposed_object_by_alias: dict[Any, Any] = field(default_factory=dict)
7✔
133
        _exposed_context_aware_object_factory_by_alias: dict[Any, Any] = field(default_factory=dict)
7✔
134
        _subsystem_to_providers: dict[type[Subsystem], list[str]] = field(
7✔
135
            default_factory=lambda: defaultdict(list)
136
        )
137
        _target_type_to_providers: dict[type[Target], list[str]] = field(
7✔
138
            default_factory=lambda: defaultdict(list)
139
        )
140
        _rule_to_providers: dict[Rule, list[str]] = field(default_factory=lambda: defaultdict(list))
7✔
141
        _union_rule_to_providers: dict[UnionRule, list[str]] = field(
7✔
142
            default_factory=lambda: defaultdict(list)
143
        )
144
        _allow_unknown_options: bool = False
7✔
145
        _remote_auth_plugin: Callable | None = None
7✔
146

147
        def registered_aliases(self) -> BuildFileAliases:
7✔
148
            """Return the registered aliases exposed in BUILD files.
149

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

153
            :returns: A new BuildFileAliases instance containing this BuildConfiguration's
154
                      registered alias mappings.
155
            """
156
            return BuildFileAliases(
×
157
                objects=self._exposed_object_by_alias.copy(),
158
                context_aware_object_factories=self._exposed_context_aware_object_factory_by_alias.copy(),
159
            )
160

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

164
            :param aliases: The BuildFileAliases to register.
165
            :type aliases: :class:`pants.build_graph.build_file_aliases.BuildFileAliases`
166
            """
167
            if not isinstance(aliases, BuildFileAliases):
7✔
168
                raise TypeError(f"The aliases must be a BuildFileAliases, given {aliases}")
×
169

170
            for alias, obj in aliases.objects.items():
7✔
171
                self._register_exposed_object(alias, obj)
7✔
172

173
            for (
7✔
174
                alias,
175
                context_aware_object_factory,
176
            ) in aliases.context_aware_object_factories.items():
177
                self._register_exposed_context_aware_object_factory(
×
178
                    alias, context_aware_object_factory
179
                )
180

181
        def _register_exposed_object(self, alias, obj):
7✔
182
            if alias in self._exposed_object_by_alias:
7✔
183
                logger.debug(f"Object alias {alias} has already been registered. Overwriting!")
1✔
184

185
            self._exposed_object_by_alias[alias] = obj
7✔
186

187
        def _register_exposed_context_aware_object_factory(
7✔
188
            self, alias, context_aware_object_factory
189
        ):
190
            if alias in self._exposed_context_aware_object_factory_by_alias:
×
191
                logger.debug(
×
192
                    "This context aware object factory alias {} has already been registered. "
193
                    "Overwriting!".format(alias)
194
                )
195

196
            self._exposed_context_aware_object_factory_by_alias[alias] = (
×
197
                context_aware_object_factory
198
            )
199

200
        def register_subsystems(
7✔
201
            self, plugin_or_backend: str, subsystems: Iterable[type[Subsystem]]
202
        ):
203
            """Registers the given subsystem types."""
204
            if not isinstance(subsystems, Iterable):
7✔
205
                raise TypeError(f"The subsystems must be an iterable, given {subsystems}")
×
206
            subsystems = tuple(subsystems)
7✔
207
            if not subsystems:
7✔
208
                return
7✔
209

210
            invalid_subsystems = [
7✔
211
                s for s in subsystems if not isinstance(s, type) or not issubclass(s, Subsystem)
212
            ]
213
            if invalid_subsystems:
7✔
214
                raise TypeError(
×
215
                    "The following items from the given subsystems are not Subsystems "
216
                    "subclasses:\n\t{}".format("\n\t".join(str(i) for i in invalid_subsystems))
217
                )
218

219
            for subsystem in subsystems:
7✔
220
                self._subsystem_to_providers[subsystem].append(plugin_or_backend)
7✔
221

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

227
            # "Index" the rules to normalize them and expand their dependencies.
228
            rule_index = RuleIndex.create(rules)
7✔
229
            rules_and_queries: tuple[Rule, ...] = (*rule_index.rules, *rule_index.queries)
7✔
230
            for _rule in rules_and_queries:
7✔
231
                self._rule_to_providers[_rule].append(plugin_or_backend)
7✔
232
            for union_rule in rule_index.union_rules:
7✔
233
                self._union_rule_to_providers[union_rule].append(plugin_or_backend)
7✔
234
            self.register_subsystems(
7✔
235
                plugin_or_backend,
236
                (
237
                    _rule.output_type
238
                    for _rule in rules_and_queries
239
                    if issubclass(_rule.output_type, Subsystem)
240
                ),
241
            )
242

243
        # NB: We expect the parameter to be Iterable[Type[Target]], but we can't be confident in
244
        # this because we pass whatever people put in their `register.py`s to this function;
245
        # I.e., this is an impure function that reads from the outside world. So, we use the type
246
        # hint `Any` and perform runtime type checking.
247
        def register_target_types(
7✔
248
            self, plugin_or_backend: str, target_types: Iterable[type[Target]] | Any
249
        ) -> None:
250
            """Registers the given target types."""
251
            if not isinstance(target_types, Iterable):
7✔
252
                raise TypeError(
×
253
                    f"The entrypoint `target_types` must return an iterable. "
254
                    f"Given {repr(target_types)}"
255
                )
256
            bad_elements = [
7✔
257
                tgt_type
258
                for tgt_type in target_types
259
                if not isinstance(tgt_type, type) or not issubclass(tgt_type, Target)
260
            ]
261
            if bad_elements:
7✔
262
                raise TypeError(
×
263
                    "Every element of the entrypoint `target_types` must be a subclass of "
264
                    f"{Target.__name__}. Bad elements: {bad_elements}."
265
                )
266
            for target_type in target_types:
7✔
267
                self._target_type_to_providers[target_type].append(plugin_or_backend)
7✔
268
                # Access the Target.PluginField here to ensure the PluginField class is
269
                # created before the UnionMembership is instantiated, as the class hierarchy is
270
                # walked during union membership setup.
271
                _ = target_type.PluginField
7✔
272

273
        def register_remote_auth_plugin(self, remote_auth_plugin: Callable) -> None:
7✔
274
            self._remote_auth_plugin = remote_auth_plugin
×
275

276
        def register_auxiliary_goals(self, plugin_or_backend: str, auxiliary_goals: Iterable[type]):
7✔
277
            """Registers the given auxiliary goals."""
278
            if not isinstance(auxiliary_goals, Iterable):
7✔
279
                raise TypeError(
×
280
                    f"The entrypoint `auxiliary_goals` must return an iterable. "
281
                    f"Given {repr(auxiliary_goals)}"
282
                )
283
            # Import `AuxiliaryGoal` here to avoid import cycle.
284
            from pants.goal.auxiliary_goal import AuxiliaryGoal
7✔
285

286
            bad_elements = [goal for goal in auxiliary_goals if not issubclass(goal, AuxiliaryGoal)]
7✔
287
            if bad_elements:
7✔
288
                raise TypeError(
×
289
                    "Every element of the entrypoint `auxiliary_goals` must be a subclass of "
290
                    f"{AuxiliaryGoal.__name__}. Bad elements: {bad_elements}."
291
                )
292
            self.register_subsystems(plugin_or_backend, auxiliary_goals)
7✔
293

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

297
            Used to defer options failures while bootstrapping BuildConfiguration until after the
298
            complete set of plugins is known.
299
            """
300
            self._allow_unknown_options = True
2✔
301

302
        def create(self) -> BuildConfiguration:
7✔
303
            registered_aliases = BuildFileAliases(
7✔
304
                objects=self._exposed_object_by_alias.copy(),
305
                context_aware_object_factories=self._exposed_context_aware_object_factory_by_alias.copy(),
306
            )
307
            return BuildConfiguration(
7✔
308
                registered_aliases=registered_aliases,
309
                subsystem_to_providers=FrozenDict(
310
                    (k, tuple(v)) for k, v in self._subsystem_to_providers.items()
311
                ),
312
                target_type_to_providers=FrozenDict(
313
                    (k, tuple(v)) for k, v in self._target_type_to_providers.items()
314
                ),
315
                rule_to_providers=FrozenDict(
316
                    (k, tuple(v)) for k, v in self._rule_to_providers.items()
317
                ),
318
                union_rule_to_providers=FrozenDict(
319
                    (k, tuple(v)) for k, v in self._union_rule_to_providers.items()
320
                ),
321
                allow_unknown_options=self._allow_unknown_options,
322
                remote_auth_plugin_func=self._remote_auth_plugin,
323
            )
324

325

326
@rule
7✔
327
async def get_full_options_parsing_settings(
7✔
328
    build_config: BuildConfiguration,
329
) -> OptionsParsingSettings:
330
    return OptionsParsingSettings(
×
331
        build_config.known_scope_infos,
332
        build_config.allow_unknown_options,
333
    )
334

335

336
def rules():
7✔
337
    return collect_rules()
7✔
STATUS · Troubleshooting · Open an Issue · Sales · Support · CAREERS · ENTERPRISE · START FREE · SCHEDULE DEMO
ANNOUNCEMENTS · TWITTER · TOS & SLA · Supported CI Services · What's a CI service? · Automated Testing

© 2025 Coveralls, Inc