19250292619

Committed 11 Nov 2025 12:09AM UTC coverage: 77.865% (-2.4%) from 80.298%

Build # 19250292619

Build Type

push

github

Committed by

web-flow

Commit Message

flag non-runnable targets used with `code_quality_tool` (#22875)

Run Details

2 of 5 new or added lines in 2 files covered. (40.0%)

1487 existing lines in 72 files now uncovered.

71448 of 91759 relevant lines covered (77.86%)

3.22 hits per line

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

77.7

/src/python/pants/init/engine_initializer.py

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

from __future__ import annotations

import dataclasses
import logging
from collections.abc import Iterable, Mapping
from dataclasses import dataclass
from pathlib import Path
from typing import Any, ClassVar, cast

from pants.base.build_environment import get_buildroot
from pants.base.build_root import BuildRoot
from pants.base.exiter import PANTS_SUCCEEDED_EXIT_CODE
from pants.base.specs import Specs
from pants.build_graph.build_configuration import BuildConfiguration
from pants.core.environments import rules as environments_rules
from pants.core.environments.rules import determine_bootstrap_environment
from pants.core.util_rules import system_binaries
from pants.engine import desktop, download_file, fs, intrinsics, process
from pants.engine.console import Console
from pants.engine.environment import EnvironmentName
from pants.engine.fs import PathGlobs, Snapshot, Workspace
from pants.engine.goal import CurrentExecutingGoals, Goal
from pants.engine.internals import (
    build_files,
    dep_rules,
    graph,
    options_parsing,
    platform_rules,
    specs_rules,
    synthetic_targets,
)
from pants.engine.internals.native_engine import PyExecutor, PySessionCancellationLatch
from pants.engine.internals.parser import Parser
from pants.engine.internals.scheduler import Scheduler, SchedulerSession
from pants.engine.internals.selectors import Params
from pants.engine.internals.session import SessionValues
from pants.engine.rules import QueryRule, collect_rules, rule
from pants.engine.streaming_workunit_handler import rules as streaming_workunit_handler_rules
from pants.engine.target import RegisteredTargetTypes
from pants.engine.unions import UnionMembership, UnionRule
from pants.init import specs_calculator
from pants.init.bootstrap_scheduler import BootstrapStatus
from pants.option.bootstrap_options import (
    DEFAULT_EXECUTION_OPTIONS,
    DynamicRemoteOptions,
    ExecutionOptions,
    LocalStoreOptions,
)
from pants.option.global_options import GlobalOptions
from pants.option.option_value_container import OptionValueContainer
from pants.option.subsystem import Subsystem
from pants.util.docutil import bin_name
from pants.util.logging import LogLevel
from pants.util.ordered_set import FrozenOrderedSet
from pants.util.strutil import softwrap
from pants.vcs.changed import rules as changed_rules
from pants.vcs.git import rules as git_rules

logger = logging.getLogger(__name__)


@dataclass(frozen=True)
class GraphScheduler:
    """A thin wrapper around a Scheduler configured with @rules."""

    scheduler: Scheduler
    goal_map: Any

    def new_session(
        self,
        build_id,
        dynamic_ui: bool = False,
        ui_use_prodash: bool = False,
        use_colors=True,
        max_workunit_level: LogLevel = LogLevel.DEBUG,
        session_values: SessionValues | None = None,
        cancellation_latch: PySessionCancellationLatch | None = None,
    ) -> GraphSession:
        session = self.scheduler.new_session(
            build_id,
            dynamic_ui,
            ui_use_prodash,
            max_workunit_level=max_workunit_level,
            session_values=session_values,
            cancellation_latch=cancellation_latch,
        )
        console = Console(use_colors=use_colors, session=session if dynamic_ui else None)
        return GraphSession(session, console, self.goal_map)


@dataclass(frozen=True)
class GraphSession:
    """A thin wrapper around a SchedulerSession configured with @rules."""

    scheduler_session: SchedulerSession
    console: Console
    goal_map: Any

    # NB: Keep this in sync with the method `run_goal_rules`.
    goal_param_types: ClassVar[tuple[type, ...]] = (Specs, Console, Workspace, EnvironmentName)

    def goal_consumed_subsystem_scopes(self, goal_name: str) -> tuple[str, ...]:
        """Return the scopes of subsystems that could be consumed while running the given goal."""
        goal_product = self.goal_map.get(goal_name)
        if not goal_product:
            return tuple()
        consumed_types = self.goal_consumed_types(goal_product)
        return tuple(
            sorted({typ.options_scope for typ in consumed_types if issubclass(typ, Subsystem)})
        )

    def goal_consumed_types(self, goal_product: type) -> set[type]:
        """Return the set of types that could possibly be consumed while running the given goal."""
        return set(
            self.scheduler_session.scheduler.rule_graph_consumed_types(
                self.goal_param_types, goal_product
            )
        )

    def run_goal_rules(
        self,
        *,
        union_membership: UnionMembership,
        goals: Iterable[str],
        specs: Specs,
        poll: bool = False,
        poll_delay: float | None = None,
    ) -> int:
        """Runs @goal_rules sequentially and interactively by requesting their implicit Goal
        products.

        For retryable failures, raises scheduler.ExecutionError.

        :returns: An exit code.
        """

        workspace = Workspace(self.scheduler_session)
        env_name = determine_bootstrap_environment(self.scheduler_session)

        for goal in goals:
            goal_product = self.goal_map[goal]
            if not goal_product.subsystem_cls.activated(union_membership):
                raise GoalNotActivatedException(goal)
            # NB: Keep this in sync with the property `goal_param_types`.
            params = Params(specs, self.console, workspace, env_name)
            logger.debug(f"requesting {goal_product} to satisfy execution of `{goal}` goal")
            try:
                exit_code = self.scheduler_session.run_goal_rule(
                    goal_product, params, poll=poll, poll_delay=poll_delay
                )
            finally:
                self.console.flush()

            if exit_code != PANTS_SUCCEEDED_EXIT_CODE:
                return exit_code

        return PANTS_SUCCEEDED_EXIT_CODE


class EngineInitializer:
    """Constructs the components necessary to run the engine."""

    class GoalMappingError(Exception):
        """Raised when a goal cannot be mapped to an @rule."""

    @staticmethod
    def _make_goal_map_from_rules(rules) -> Mapping[str, type[Goal]]:
        goal_map: dict[str, type[Goal]] = {}
        for r in rules:
            output_type = getattr(r, "output_type", None)
            if not output_type or not issubclass(output_type, Goal):
                continue

            goal = r.output_type.name
            deprecated_goal = r.output_type.subsystem_cls.deprecated_options_scope
            for goal_name in [goal, deprecated_goal] if deprecated_goal else [goal]:
                if goal_name in goal_map:
                    raise EngineInitializer.GoalMappingError(
                        f"could not map goal `{goal_name}` to rule `{r}`: already claimed by product "
                        f"`{goal_map[goal_name]}`"
                    )
                goal_map[goal_name] = r.output_type
        return goal_map

    @staticmethod
    def setup_graph(
        bootstrap_options: OptionValueContainer,
        build_configuration: BuildConfiguration,
        dynamic_remote_options: DynamicRemoteOptions,
        executor: PyExecutor,
        is_bootstrap: bool = False,
    ) -> GraphScheduler:
        build_root = get_buildroot()
        executor = executor or GlobalOptions.create_py_executor(bootstrap_options)
        execution_options = ExecutionOptions.from_options(bootstrap_options, dynamic_remote_options)
        if is_bootstrap:
            # Don't spawn a single-use sandboxer process that will then get preempted by a
            # post-bootstrap one created by pantsd. This is fine: our plugin resolving sequence
            # isn't susceptible to the race condition that the sandboxer solves.
            # TODO: Are we sure? In any case we plan to replace the plugin resolver and
            #  get rid of the bootstrap scheduler, so this should be moot soon enough.
            execution_options = dataclasses.replace(execution_options, use_sandboxer=False)
        local_store_options = LocalStoreOptions.from_options(bootstrap_options)
        return EngineInitializer.setup_graph_extended(
            build_configuration,
            execution_options,
            executor=executor,
            pants_ignore_patterns=GlobalOptions.compute_pants_ignore(build_root, bootstrap_options),
            use_gitignore=bootstrap_options.pants_ignore_use_gitignore,
            local_store_options=local_store_options,
            local_execution_root_dir=bootstrap_options.local_execution_root_dir,
            named_caches_dir=bootstrap_options.named_caches_dir,
            ca_certs_path=bootstrap_options.ca_certs_path,
            build_root=build_root,
            pants_workdir=bootstrap_options.pants_workdir,
            include_trace_on_error=bootstrap_options.print_stacktrace,
            engine_visualize_to=bootstrap_options.engine_visualize_to,
            watch_filesystem=bootstrap_options.watch_filesystem,
            is_bootstrap=is_bootstrap,
        )

    @staticmethod
    def setup_graph_extended(
        build_configuration: BuildConfiguration,
        execution_options: ExecutionOptions,
        *,
        executor: PyExecutor,
        pants_ignore_patterns: list[str],
        use_gitignore: bool,
        local_store_options: LocalStoreOptions,
        local_execution_root_dir: str,
        named_caches_dir: str,
        pants_workdir: str,
        ca_certs_path: str | None = None,
        build_root: str | None = None,
        include_trace_on_error: bool = True,
        engine_visualize_to: str | None = None,
        watch_filesystem: bool = True,
        is_bootstrap: bool = False,
    ) -> GraphScheduler:
        build_root_path = build_root or get_buildroot()

        rules = build_configuration.rules
        union_membership: UnionMembership
        registered_target_types = RegisteredTargetTypes.create(build_configuration.target_types)

        execution_options = execution_options or DEFAULT_EXECUTION_OPTIONS

        @rule
        async def parser_singleton() -> Parser:
            return Parser(
                build_root=build_root_path,
                registered_target_types=registered_target_types,
                union_membership=union_membership,
                object_aliases=build_configuration.registered_aliases,
                ignore_unrecognized_symbols=is_bootstrap,
            )

        @rule
        async def bootstrap_status() -> BootstrapStatus:
            return BootstrapStatus(is_bootstrap)

        @rule
        async def build_configuration_singleton() -> BuildConfiguration:
            return build_configuration

        @rule
        async def registered_target_types_singleton() -> RegisteredTargetTypes:
            return registered_target_types

        @rule
        async def union_membership_singleton() -> UnionMembership:
            return union_membership

        @rule
        async def build_root_singleton() -> BuildRoot:
            return cast(BuildRoot, BuildRoot.instance)

        @rule
        async def current_executing_goals(session_values: SessionValues) -> CurrentExecutingGoals:
            return session_values.get(CurrentExecutingGoals) or CurrentExecutingGoals()

        # Create a Scheduler containing graph and filesystem rules, with no installed goals.
        rules = FrozenOrderedSet(
            (
                *collect_rules(locals()),
                *intrinsics.rules(),
                *build_files.rules(),
                *fs.rules(),
                *dep_rules.rules(),
                *desktop.rules(),
                *download_file.rules(),
                *git_rules(),
                *graph.rules(),
                *specs_rules.rules(),
                *options_parsing.rules(),
                *process.rules(),
                *environments_rules.rules(),
                *system_binaries.rules(),
                *platform_rules.rules(),
                *changed_rules(),
                *streaming_workunit_handler_rules(),
                *specs_calculator.rules(),
                *synthetic_targets.rules(),
                *rules,
            )
        )

        goal_map = EngineInitializer._make_goal_map_from_rules(rules)

        union_membership = UnionMembership.from_rules(
            (
                *build_configuration.union_rules,
                *(r for r in rules if isinstance(r, UnionRule)),
            )
        )

        # param types for goals with the `USES_ENVIRONMENT` behaviour (see `goal.py`)
        environment_selecting_goal_param_types = [
            t for t in GraphSession.goal_param_types if t != EnvironmentName
        ]
        rules = FrozenOrderedSet(
            (
                *rules,
                # Install queries for each Goal.
                *(
                    QueryRule(
                        goal_type,
                        (
                            environment_selecting_goal_param_types
                            if goal_type._selects_environments()
                            else GraphSession.goal_param_types
                        ),
                    )
                    for goal_type in goal_map.values()
                ),
                QueryRule(Snapshot, [PathGlobs]),  # Used by the SchedulerService.
            )
        )

        def ensure_absolute_path(v: str) -> str:
            return Path(v).resolve().as_posix()

        def ensure_optional_absolute_path(v: str | None) -> str | None:
            if v is None:
                return None
            return ensure_absolute_path(v)

        scheduler = Scheduler(
            ignore_patterns=pants_ignore_patterns,
            use_gitignore=use_gitignore,
            build_root=build_root_path,
            pants_workdir=pants_workdir,
            local_execution_root_dir=ensure_absolute_path(local_execution_root_dir),
            named_caches_dir=ensure_absolute_path(named_caches_dir),
            ca_certs_path=ensure_optional_absolute_path(ca_certs_path),
            rules=rules,
            union_membership=union_membership,
            executor=executor,
            execution_options=execution_options,
            local_store_options=local_store_options,
            include_trace_on_error=include_trace_on_error,
            visualize_to_dir=engine_visualize_to,
            watch_filesystem=watch_filesystem,
        )

        return GraphScheduler(scheduler, goal_map)


class GoalNotActivatedException(Exception):
    def __init__(self, goal_name: str) -> None:
        super().__init__(
            softwrap(
                f"""
                No relevant backends activate the `{goal_name}` goal, so the goal would do
                nothing.

                This usually means that you have not yet set the option
                `[GLOBAL].backend_packages` in `pants.toml`, which is how Pants knows
                which languages and tools to support. Run `{bin_name()} help backends`.
                """
            )
        )

1	# Copyright 2016 Pants project contributors (see CONTRIBUTORS.md).
2	# Licensed under the Apache License, Version 2.0 (see LICENSE).
3
4	from __future__ import annotations	11✔
5
6	import dataclasses	11✔
7	import logging	11✔
8	from collections.abc import Iterable, Mapping	11✔
9	from dataclasses import dataclass	11✔
10	from pathlib import Path	11✔
11	from typing import Any, ClassVar, cast	11✔
12
13	from pants.base.build_environment import get_buildroot	11✔
14	from pants.base.build_root import BuildRoot	11✔
15	from pants.base.exiter import PANTS_SUCCEEDED_EXIT_CODE	11✔
16	from pants.base.specs import Specs	11✔
17	from pants.build_graph.build_configuration import BuildConfiguration	11✔
18	from pants.core.environments import rules as environments_rules	11✔
19	from pants.core.environments.rules import determine_bootstrap_environment	11✔
20	from pants.core.util_rules import system_binaries	11✔
21	from pants.engine import desktop, download_file, fs, intrinsics, process	11✔
22	from pants.engine.console import Console	11✔
23	from pants.engine.environment import EnvironmentName	11✔
24	from pants.engine.fs import PathGlobs, Snapshot, Workspace	11✔
25	from pants.engine.goal import CurrentExecutingGoals, Goal	11✔
26	from pants.engine.internals import (	11✔
27	build_files,
28	dep_rules,
29	graph,
30	options_parsing,
31	platform_rules,
32	specs_rules,
33	synthetic_targets,
34	)
35	from pants.engine.internals.native_engine import PyExecutor, PySessionCancellationLatch	11✔
36	from pants.engine.internals.parser import Parser	11✔
37	from pants.engine.internals.scheduler import Scheduler, SchedulerSession	11✔
38	from pants.engine.internals.selectors import Params	11✔
39	from pants.engine.internals.session import SessionValues	11✔
40	from pants.engine.rules import QueryRule, collect_rules, rule	11✔
41	from pants.engine.streaming_workunit_handler import rules as streaming_workunit_handler_rules	11✔
42	from pants.engine.target import RegisteredTargetTypes	11✔
43	from pants.engine.unions import UnionMembership, UnionRule	11✔
44	from pants.init import specs_calculator	11✔
45	from pants.init.bootstrap_scheduler import BootstrapStatus	11✔
46	from pants.option.bootstrap_options import (	11✔
47	DEFAULT_EXECUTION_OPTIONS,
48	DynamicRemoteOptions,
49	ExecutionOptions,
50	LocalStoreOptions,
51	)
52	from pants.option.global_options import GlobalOptions	11✔
53	from pants.option.option_value_container import OptionValueContainer	11✔
54	from pants.option.subsystem import Subsystem	11✔
55	from pants.util.docutil import bin_name	11✔
56	from pants.util.logging import LogLevel	11✔
57	from pants.util.ordered_set import FrozenOrderedSet	11✔
58	from pants.util.strutil import softwrap	11✔
59	from pants.vcs.changed import rules as changed_rules	11✔
60	from pants.vcs.git import rules as git_rules	11✔
61
62	logger = logging.getLogger(__name__)	11✔
63
64
65	@dataclass(frozen=True)	11✔
66	class GraphScheduler:	11✔
67	"""A thin wrapper around a Scheduler configured with @rules."""
68
69	scheduler: Scheduler	11✔
70	goal_map: Any	11✔
71
72	def new_session(	11✔
73	self,
74	build_id,
75	dynamic_ui: bool = False,
76	ui_use_prodash: bool = False,
77	use_colors=True,
78	max_workunit_level: LogLevel = LogLevel.DEBUG,
79	session_values: SessionValues \| None = None,
80	cancellation_latch: PySessionCancellationLatch \| None = None,
81	) -> GraphSession:
82	session = self.scheduler.new_session(	×
83	build_id,
84	dynamic_ui,
85	ui_use_prodash,
86	max_workunit_level=max_workunit_level,
87	session_values=session_values,
88	cancellation_latch=cancellation_latch,
89	)
90	console = Console(use_colors=use_colors, session=session if dynamic_ui else None)	×
91	return GraphSession(session, console, self.goal_map)	×
92
93
94	@dataclass(frozen=True)	11✔
95	class GraphSession:	11✔
96	"""A thin wrapper around a SchedulerSession configured with @rules."""
97
98	scheduler_session: SchedulerSession	11✔
99	console: Console	11✔
100	goal_map: Any	11✔
101
102	# NB: Keep this in sync with the method `run_goal_rules`.
103	goal_param_types: ClassVar[tuple[type, ...]] = (Specs, Console, Workspace, EnvironmentName)	11✔
104
105	def goal_consumed_subsystem_scopes(self, goal_name: str) -> tuple[str, ...]:	11✔
106	"""Return the scopes of subsystems that could be consumed while running the given goal."""
107	goal_product = self.goal_map.get(goal_name)	×
108	if not goal_product:	×
109	return tuple()	×
110	consumed_types = self.goal_consumed_types(goal_product)	×
111	return tuple(	×
112	sorted({typ.options_scope for typ in consumed_types if issubclass(typ, Subsystem)})
113	)
114
115	def goal_consumed_types(self, goal_product: type) -> set[type]:	11✔
116	"""Return the set of types that could possibly be consumed while running the given goal."""
117	return set(	×
118	self.scheduler_session.scheduler.rule_graph_consumed_types(
119	self.goal_param_types, goal_product
120	)
121	)
122
123	def run_goal_rules(	11✔
124	self,
125	*,
126	union_membership: UnionMembership,
127	goals: Iterable[str],
128	specs: Specs,
129	poll: bool = False,
130	poll_delay: float \| None = None,
131	) -> int:
132	"""Runs @goal_rules sequentially and interactively by requesting their implicit Goal
133	products.
134
135	For retryable failures, raises scheduler.ExecutionError.
136
137	:returns: An exit code.
138	"""
139
140	workspace = Workspace(self.scheduler_session)	×
141	env_name = determine_bootstrap_environment(self.scheduler_session)	×
142
143	for goal in goals:	×
144	goal_product = self.goal_map[goal]	×
145	if not goal_product.subsystem_cls.activated(union_membership):	×
146	raise GoalNotActivatedException(goal)	×
147	# NB: Keep this in sync with the property `goal_param_types`.
148	params = Params(specs, self.console, workspace, env_name)	×
149	logger.debug(f"requesting {goal_product} to satisfy execution of `{goal}` goal")	×
150	try:	×
151	exit_code = self.scheduler_session.run_goal_rule(	×
152	goal_product, params, poll=poll, poll_delay=poll_delay
153	)
154	finally:
155	self.console.flush()	×
156
157	if exit_code != PANTS_SUCCEEDED_EXIT_CODE:	×
158	return exit_code	×
159
160	return PANTS_SUCCEEDED_EXIT_CODE	×
161
162
163	class EngineInitializer:	11✔
164	"""Constructs the components necessary to run the engine."""
165
166	class GoalMappingError(Exception):	11✔
167	"""Raised when a goal cannot be mapped to an @rule."""
168
169	@staticmethod	11✔
170	def _make_goal_map_from_rules(rules) -> Mapping[str, type[Goal]]:	11✔
171	goal_map: dict[str, type[Goal]] = {}	11✔
172	for r in rules:	11✔
173	output_type = getattr(r, "output_type", None)	11✔
174	if not output_type or not issubclass(output_type, Goal):	11✔
175	continue	11✔
176
177	goal = r.output_type.name	11✔
178	deprecated_goal = r.output_type.subsystem_cls.deprecated_options_scope	11✔
179	for goal_name in [goal, deprecated_goal] if deprecated_goal else [goal]:	11✔
180	if goal_name in goal_map:	11✔
181	raise EngineInitializer.GoalMappingError(	×
182	f"could not map goal `{goal_name}` to rule `{r}`: already claimed by product "
183	f"`{goal_map[goal_name]}`"
184	)
185	goal_map[goal_name] = r.output_type	11✔
186	return goal_map	11✔
187
188	@staticmethod	11✔
189	def setup_graph(	11✔
190	bootstrap_options: OptionValueContainer,
191	build_configuration: BuildConfiguration,
192	dynamic_remote_options: DynamicRemoteOptions,
193	executor: PyExecutor,
194	is_bootstrap: bool = False,
195	) -> GraphScheduler:
196	build_root = get_buildroot()	4✔
197	executor = executor or GlobalOptions.create_py_executor(bootstrap_options)	4✔
198	execution_options = ExecutionOptions.from_options(bootstrap_options, dynamic_remote_options)	4✔
199	if is_bootstrap:	4✔
200	# Don't spawn a single-use sandboxer process that will then get preempted by a
201	# post-bootstrap one created by pantsd. This is fine: our plugin resolving sequence
202	# isn't susceptible to the race condition that the sandboxer solves.
203	# TODO: Are we sure? In any case we plan to replace the plugin resolver and
204	# get rid of the bootstrap scheduler, so this should be moot soon enough.
205	execution_options = dataclasses.replace(execution_options, use_sandboxer=False)	4✔
206	local_store_options = LocalStoreOptions.from_options(bootstrap_options)	4✔
207	return EngineInitializer.setup_graph_extended(	4✔
208	build_configuration,
209	execution_options,
210	executor=executor,
211	pants_ignore_patterns=GlobalOptions.compute_pants_ignore(build_root, bootstrap_options),
212	use_gitignore=bootstrap_options.pants_ignore_use_gitignore,
213	local_store_options=local_store_options,
214	local_execution_root_dir=bootstrap_options.local_execution_root_dir,
215	named_caches_dir=bootstrap_options.named_caches_dir,
216	ca_certs_path=bootstrap_options.ca_certs_path,
217	build_root=build_root,
218	pants_workdir=bootstrap_options.pants_workdir,
219	include_trace_on_error=bootstrap_options.print_stacktrace,
220	engine_visualize_to=bootstrap_options.engine_visualize_to,
221	watch_filesystem=bootstrap_options.watch_filesystem,
222	is_bootstrap=is_bootstrap,
223	)
224
225	@staticmethod	11✔
226	def setup_graph_extended(	11✔
227	build_configuration: BuildConfiguration,
228	execution_options: ExecutionOptions,
229	*,
230	executor: PyExecutor,
231	pants_ignore_patterns: list[str],
232	use_gitignore: bool,
233	local_store_options: LocalStoreOptions,
234	local_execution_root_dir: str,
235	named_caches_dir: str,
236	pants_workdir: str,
237	ca_certs_path: str \| None = None,
238	build_root: str \| None = None,
239	include_trace_on_error: bool = True,
240	engine_visualize_to: str \| None = None,
241	watch_filesystem: bool = True,
242	is_bootstrap: bool = False,
243	) -> GraphScheduler:
244	build_root_path = build_root or get_buildroot()	11✔
245
246	rules = build_configuration.rules	11✔
247	union_membership: UnionMembership
248	registered_target_types = RegisteredTargetTypes.create(build_configuration.target_types)	11✔
249
250	execution_options = execution_options or DEFAULT_EXECUTION_OPTIONS	11✔
251
252	@rule	11✔
253	async def parser_singleton() -> Parser:	11✔
254	return Parser(	×
255	build_root=build_root_path,
256	registered_target_types=registered_target_types,
257	union_membership=union_membership,
258	object_aliases=build_configuration.registered_aliases,
259	ignore_unrecognized_symbols=is_bootstrap,
260	)
261
262	@rule	11✔
263	async def bootstrap_status() -> BootstrapStatus:	11✔
264	return BootstrapStatus(is_bootstrap)	×
265
266	@rule	11✔
267	async def build_configuration_singleton() -> BuildConfiguration:	11✔
268	return build_configuration	×
269
270	@rule	11✔
271	async def registered_target_types_singleton() -> RegisteredTargetTypes:	11✔
272	return registered_target_types	×
273
274	@rule	11✔
275	async def union_membership_singleton() -> UnionMembership:	11✔
276	return union_membership	×
277
278	@rule	11✔
279	async def build_root_singleton() -> BuildRoot:	11✔
280	return cast(BuildRoot, BuildRoot.instance)	×
281
282	@rule	11✔
283	async def current_executing_goals(session_values: SessionValues) -> CurrentExecutingGoals:	11✔
284	return session_values.get(CurrentExecutingGoals) or CurrentExecutingGoals()	×
285
286	# Create a Scheduler containing graph and filesystem rules, with no installed goals.
287	rules = FrozenOrderedSet(	11✔
288	(
289	*collect_rules(locals()),
290	*intrinsics.rules(),
291	*build_files.rules(),
292	*fs.rules(),
293	*dep_rules.rules(),
294	*desktop.rules(),
295	*download_file.rules(),
296	*git_rules(),
297	*graph.rules(),
298	*specs_rules.rules(),
299	*options_parsing.rules(),
300	*process.rules(),
301	*environments_rules.rules(),
302	*system_binaries.rules(),
303	*platform_rules.rules(),
304	*changed_rules(),
305	*streaming_workunit_handler_rules(),
306	*specs_calculator.rules(),
307	*synthetic_targets.rules(),
308	*rules,
309	)
310	)
311
312	goal_map = EngineInitializer._make_goal_map_from_rules(rules)	11✔
313
314	union_membership = UnionMembership.from_rules(	11✔
315	(
316	*build_configuration.union_rules,
317	*(r for r in rules if isinstance(r, UnionRule)),
318	)
319	)
320
321	# param types for goals with the `USES_ENVIRONMENT` behaviour (see `goal.py`)
322	environment_selecting_goal_param_types = [	11✔
323	t for t in GraphSession.goal_param_types if t != EnvironmentName
324	]
325	rules = FrozenOrderedSet(	11✔
326	(
327	*rules,
328	# Install queries for each Goal.
329	*(
330	QueryRule(
331	goal_type,
332	(
333	environment_selecting_goal_param_types
334	if goal_type._selects_environments()
335	else GraphSession.goal_param_types
336	),
337	)
338	for goal_type in goal_map.values()
339	),
340	QueryRule(Snapshot, [PathGlobs]), # Used by the SchedulerService.
341	)
342	)
343
344	def ensure_absolute_path(v: str) -> str:	11✔
345	return Path(v).resolve().as_posix()	11✔
346
347	def ensure_optional_absolute_path(v: str \| None) -> str \| None:	11✔
348	if v is None:	11✔
349	return None	11✔
UNCOV 350	return ensure_absolute_path(v)	×
351
352	scheduler = Scheduler(	11✔
353	ignore_patterns=pants_ignore_patterns,
354	use_gitignore=use_gitignore,
355	build_root=build_root_path,
356	pants_workdir=pants_workdir,
357	local_execution_root_dir=ensure_absolute_path(local_execution_root_dir),
358	named_caches_dir=ensure_absolute_path(named_caches_dir),
359	ca_certs_path=ensure_optional_absolute_path(ca_certs_path),
360	rules=rules,
361	union_membership=union_membership,
362	executor=executor,
363	execution_options=execution_options,
364	local_store_options=local_store_options,
365	include_trace_on_error=include_trace_on_error,
366	visualize_to_dir=engine_visualize_to,
367	watch_filesystem=watch_filesystem,
368	)
369
370	return GraphScheduler(scheduler, goal_map)	11✔
371
372
373	class GoalNotActivatedException(Exception):	11✔
374	def __init__(self, goal_name: str) -> None:	11✔
375	super().__init__(	×
376	softwrap(
377	f"""
378	No relevant backends activate the `{goal_name}` goal, so the goal would do
379	nothing.
380
381	This usually means that you have not yet set the option
382	`[GLOBAL].backend_packages` in `pants.toml`, which is how Pants knows
383	which languages and tools to support. Run `{bin_name()} help backends`.
384	"""
385	)
386	)

pantsbuild / pants / 19250292619

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