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

61.9

/src/python/pants/core/goals/run.py

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

from __future__ import annotations

import logging
from abc import ABCMeta
from collections.abc import Iterable, Mapping
from dataclasses import dataclass
from enum import Enum
from typing import Any, ClassVar, TypeVar, final

from pants.core.environments.rules import _warn_on_non_local_environments
from pants.core.subsystems.debug_adapter import DebugAdapterSubsystem
from pants.engine.env_vars import CompleteEnvironmentVars
from pants.engine.environment import EnvironmentName
from pants.engine.fs import Digest, Workspace
from pants.engine.goal import Goal, GoalSubsystem
from pants.engine.internals.specs_rules import (
    AmbiguousImplementationsException,
    TooManyTargetsException,
    find_valid_field_sets_for_target_roots,
)
from pants.engine.intrinsics import run_interactive_process
from pants.engine.process import InteractiveProcess
from pants.engine.rules import Rule, _uncacheable_rule, collect_rules, goal_rule, implicitly, rule
from pants.engine.target import (
    BoolField,
    FieldSet,
    NoApplicableTargetsBehavior,
    Target,
    TargetRootsToFieldSetsRequest,
)
from pants.engine.unions import UnionMembership, UnionRule, union
from pants.option.global_options import GlobalOptions
from pants.option.option_types import ArgsListOption, BoolOption
from pants.util.frozendict import FrozenDict
from pants.util.memo import memoized
from pants.util.strutil import help_text, softwrap

logger = logging.getLogger(__name__)

_T = TypeVar("_T")


class RunInSandboxBehavior(Enum):
    """Defines the behavior of rules that act on a `RunFieldSet` subclass with regards to use in the
    sandbox.

    This is used to automatically generate rules used to fulfill `experimental_run_in_sandbox`
    targets.

    The behaviors are as follows:

    * `RUN_REQUEST_HERMETIC`: Use the existing `RunRequest`-generating rule, and enable cacheing.
       Use this if you are confident the behavior of the rule relies only on state that is
       captured by pants (e.g. binary paths are found using `EnvironmentVarsRequest`), and that
       the rule only refers to files in the sandbox.
    * `RUN_REQUEST_NOT_HERMETIC`: Use the existing `RunRequest`-generating rule, and do not
       enable cacheing. Use this if your existing rule is mostly suitable for use in the sandbox,
       but you cannot guarantee reproducible behavior.
    * `CUSTOM`: Opt to write your own rule that returns `RunInSandboxRequest`.
    * `NOT_SUPPORTED`: Opt out of being usable in `experimental_run_in_sandbox`. Attempting to use
       such a target will result in a runtime exception.
    """

    RUN_REQUEST_HERMETIC = 1
    RUN_REQUEST_NOT_HERMETIC = 2
    CUSTOM = 3
    NOT_SUPPORTED = 4


@union(in_scope_types=[EnvironmentName])
class RunFieldSet(FieldSet, metaclass=ABCMeta):
    """The fields necessary from a target to run a program/script."""

    supports_debug_adapter: ClassVar[bool] = False
    run_in_sandbox_behavior: ClassVar[RunInSandboxBehavior]

    @final
    @classmethod
    def rules(cls) -> Iterable[Rule | UnionRule]:
        yield UnionRule(RunFieldSet, cls)
        if not cls.supports_debug_adapter:
            yield from _unsupported_debug_adapter_rules(cls)
        yield from _run_in_sandbox_behavior_rule(cls)


class RestartableField(BoolField):
    alias = "restartable"
    default = False
    help = help_text(
        """
        If true, runs of this target with the `run` goal may be interrupted and
        restarted when its input files change.
        """
    )


@dataclass(frozen=True)
class RunRequest:
    digest: Digest
    # Values in args and in env can contain the format specifier "{chroot}", which will
    # be substituted with the (absolute) chroot path.
    args: tuple[str, ...]
    extra_env: FrozenDict[str, str]
    immutable_input_digests: Mapping[str, Digest] | None = None
    append_only_caches: Mapping[str, str] | None = None

    def __init__(
        self,
        *,
        digest: Digest,
        args: Iterable[str],
        extra_env: Mapping[str, str] | None = None,
        immutable_input_digests: Mapping[str, Digest] | None = None,
        append_only_caches: Mapping[str, str] | None = None,
    ) -> None:
        object.__setattr__(self, "digest", digest)
        object.__setattr__(self, "args", tuple(args))
        object.__setattr__(self, "extra_env", FrozenDict(extra_env or {}))
        object.__setattr__(
            self, "immutable_input_digests", FrozenDict(immutable_input_digests or {})
        )
        object.__setattr__(self, "append_only_caches", FrozenDict(append_only_caches or {}))

    def to_run_in_sandbox_request(self) -> RunInSandboxRequest:
        return RunInSandboxRequest(
            args=self.args,
            digest=self.digest,
            extra_env=self.extra_env,
            immutable_input_digests=self.immutable_input_digests,
            append_only_caches=self.append_only_caches,
        )


class RunDebugAdapterRequest(RunRequest):
    """Like RunRequest, but launches the process using the relevant Debug Adapter server.

    The process should be launched waiting for the client to connect.
    """


class RunInSandboxRequest(RunRequest):
    """A run request that launches the process in the sandbox for use as part of a build rule.

    The arguments and environment should only use values relative to the build root (or prefixed
    with `{chroot}`), or refer to binaries that were fetched with `BinaryPathRequest`.

    Presently, implementors can opt to use the existing as not guaranteeing hermeticity, which will
    internally mark the rule as uncacheable. In such a case, non-safe APIs can be used, however,
    this behavior can result in poorer performance, and only exists as a stop-gap while
    implementors work to make sure their `RunRequest`-generating rules can be used in a hermetic
    context, or writing new custom rules. (See the Plugin Upgrade Guide for details).
    """


@rule(polymorphic=True)
async def generate_run_request(run_field_set: RunFieldSet, env_name: EnvironmentName) -> RunRequest:
    raise NotImplementedError()


@rule(polymorphic=True)
async def generate_run_debug_adapter_request(
    run_field_set: RunFieldSet, env_name: EnvironmentName
) -> RunDebugAdapterRequest:
    raise NotImplementedError()


@rule(polymorphic=True)
async def generate_run_in_sandbox_request(
    run_field_set: RunFieldSet, env_name: EnvironmentName
) -> RunInSandboxRequest:
    raise NotImplementedError()


class RunSubsystem(GoalSubsystem):
    name = "run"
    help = help_text(
        """
        Runs a binary target.

        This goal propagates the return code of the underlying executable.

        If your application can safely be restarted while it is running, you can pass
        `restartable=True` on your binary target (for supported types), and the `run` goal
        will automatically restart them as all relevant files change. This can be particularly
        useful for server applications.
        """
    )

    @classmethod
    def activated(cls, union_membership: UnionMembership) -> bool:
        return RunFieldSet in union_membership

    args = ArgsListOption(
        example="val1 val2 --debug",
        tool_name="the executed target",
        passthrough=True,
    )
    # See also `test.py`'s same option
    debug_adapter = BoolOption(
        default=False,
        help=softwrap(
            """
            Run the interactive process using a Debug Adapter
            (https://microsoft.github.io/debug-adapter-protocol/) for the language if supported.

            The interactive process used will be immediately blocked waiting for a client before
            continuing.
            """
        ),
    )


class Run(Goal):
    subsystem_cls = RunSubsystem
    environment_behavior = Goal.EnvironmentBehavior.LOCAL_ONLY


async def _find_what_to_run(
    goal_description: str,
) -> tuple[RunFieldSet, Target]:
    targets_to_valid_field_sets = await find_valid_field_sets_for_target_roots(
        TargetRootsToFieldSetsRequest(
            RunFieldSet,
            goal_description=goal_description,
            no_applicable_targets_behavior=NoApplicableTargetsBehavior.error,
        ),
        **implicitly(),
    )
    mapping = targets_to_valid_field_sets.mapping

    if len(mapping) > 1:
        raise TooManyTargetsException(mapping, goal_description=goal_description)

    target, field_sets = next(iter(mapping.items()))
    if len(field_sets) > 1:
        raise AmbiguousImplementationsException(
            target,
            field_sets,
            goal_description=goal_description,
        )

    return field_sets[0], target


@goal_rule
async def run(
    run_subsystem: RunSubsystem,
    debug_adapter: DebugAdapterSubsystem,
    global_options: GlobalOptions,
    workspace: Workspace,  # Needed to enable side-effecting.
    complete_env: CompleteEnvironmentVars,
) -> Run:
    field_set, target = await _find_what_to_run("the `run` goal")

    await _warn_on_non_local_environments((target,), "the `run` goal")

    request = await (
        generate_run_request(**implicitly({field_set: RunFieldSet}))
        if not run_subsystem.debug_adapter
        else generate_run_debug_adapter_request(**implicitly({field_set: RunFieldSet}))
    )
    restartable = target.get(RestartableField).value
    if run_subsystem.debug_adapter:
        logger.info(
            softwrap(
                f"""
                Launching debug adapter at '{debug_adapter.host}:{debug_adapter.port}',
                which will wait for a client connection...
                """
            )
        )

    result = await run_interactive_process(
        InteractiveProcess(
            argv=(*request.args, *run_subsystem.args),
            env={**complete_env, **request.extra_env},
            input_digest=request.digest,
            run_in_workspace=True,
            restartable=restartable,
            keep_sandboxes=global_options.keep_sandboxes,
            immutable_input_digests=request.immutable_input_digests,
            append_only_caches=request.append_only_caches,
        )
    )

    return Run(result.exit_code)


@memoized
def _unsupported_debug_adapter_rules(cls: type[RunFieldSet]) -> Iterable:
    """Returns a rule that implements DebugAdapterRequest by raising an error."""

    @rule(canonical_name_suffix=cls.__name__, _param_type_overrides={"request": cls})
    async def get_run_debug_adapter_request(request: RunFieldSet) -> RunDebugAdapterRequest:
        raise NotImplementedError(
            "Running this target type with a debug adapter is not yet supported."
        )

    return collect_rules(locals())


async def _run_request(request: RunFieldSet) -> RunInSandboxRequest:
    run_req = await generate_run_request(**implicitly({request: RunFieldSet}))
    return run_req.to_run_in_sandbox_request()


@memoized
def _run_in_sandbox_behavior_rule(cls: type[RunFieldSet]) -> Iterable:
    """Returns a default rule that helps fulfil `experimental_run_in_sandbox` targets.

    If `RunInSandboxBehavior.CUSTOM` is specified, rule implementors must write a rule that returns
    a `RunInSandboxRequest`.
    """

    @rule(canonical_name_suffix=cls.__name__, _param_type_overrides={"request": cls})
    async def not_supported(request: RunFieldSet) -> RunInSandboxRequest:
        raise NotImplementedError(
            "Running this target type within the sandbox is not yet supported."
        )

    @rule(canonical_name_suffix=cls.__name__, _param_type_overrides={"request": cls})
    async def run_request_hermetic(request: RunFieldSet) -> RunInSandboxRequest:
        return await _run_request(request)

    @_uncacheable_rule(canonical_name_suffix=cls.__name__, _param_type_overrides={"request": cls})
    async def run_request_not_hermetic(request: RunFieldSet) -> RunInSandboxRequest:
        return await _run_request(request)

    default_rules: dict[RunInSandboxBehavior, list[Any]] = {
        RunInSandboxBehavior.NOT_SUPPORTED: [not_supported],
        RunInSandboxBehavior.RUN_REQUEST_HERMETIC: [run_request_hermetic],
        RunInSandboxBehavior.RUN_REQUEST_NOT_HERMETIC: [run_request_not_hermetic],
        RunInSandboxBehavior.CUSTOM: [],
    }

    return collect_rules(
        {_rule.__name__: _rule for _rule in default_rules[cls.run_in_sandbox_behavior]}
    )


def rules():
    return collect_rules()

1	# Copyright 2019 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 logging	1✔
7	from abc import ABCMeta	1✔
8	from collections.abc import Iterable, Mapping	1✔
9	from dataclasses import dataclass	1✔
10	from enum import Enum	1✔
11	from typing import Any, ClassVar, TypeVar, final	1✔
12
13	from pants.core.environments.rules import _warn_on_non_local_environments	1✔
14	from pants.core.subsystems.debug_adapter import DebugAdapterSubsystem	1✔
15	from pants.engine.env_vars import CompleteEnvironmentVars	1✔
16	from pants.engine.environment import EnvironmentName	1✔
17	from pants.engine.fs import Digest, Workspace	1✔
18	from pants.engine.goal import Goal, GoalSubsystem	1✔
19	from pants.engine.internals.specs_rules import (	1✔
20	AmbiguousImplementationsException,
21	TooManyTargetsException,
22	find_valid_field_sets_for_target_roots,
23	)
24	from pants.engine.intrinsics import run_interactive_process	1✔
25	from pants.engine.process import InteractiveProcess	1✔
26	from pants.engine.rules import Rule, _uncacheable_rule, collect_rules, goal_rule, implicitly, rule	1✔
27	from pants.engine.target import (	1✔
28	BoolField,
29	FieldSet,
30	NoApplicableTargetsBehavior,
31	Target,
32	TargetRootsToFieldSetsRequest,
33	)
34	from pants.engine.unions import UnionMembership, UnionRule, union	1✔
35	from pants.option.global_options import GlobalOptions	1✔
36	from pants.option.option_types import ArgsListOption, BoolOption	1✔
37	from pants.util.frozendict import FrozenDict	1✔
38	from pants.util.memo import memoized	1✔
39	from pants.util.strutil import help_text, softwrap	1✔
40
41	logger = logging.getLogger(__name__)	1✔
42
43	_T = TypeVar("_T")	1✔
44
45
46	class RunInSandboxBehavior(Enum):	1✔
47	"""Defines the behavior of rules that act on a `RunFieldSet` subclass with regards to use in the
48	sandbox.
49
50	This is used to automatically generate rules used to fulfill `experimental_run_in_sandbox`
51	targets.
52
53	The behaviors are as follows:
54
55	* `RUN_REQUEST_HERMETIC`: Use the existing `RunRequest`-generating rule, and enable cacheing.
56	Use this if you are confident the behavior of the rule relies only on state that is
57	captured by pants (e.g. binary paths are found using `EnvironmentVarsRequest`), and that
58	the rule only refers to files in the sandbox.
59	* `RUN_REQUEST_NOT_HERMETIC`: Use the existing `RunRequest`-generating rule, and do not
60	enable cacheing. Use this if your existing rule is mostly suitable for use in the sandbox,
61	but you cannot guarantee reproducible behavior.
62	* `CUSTOM`: Opt to write your own rule that returns `RunInSandboxRequest`.
63	* `NOT_SUPPORTED`: Opt out of being usable in `experimental_run_in_sandbox`. Attempting to use
64	such a target will result in a runtime exception.
65	"""
66
67	RUN_REQUEST_HERMETIC = 1	1✔
68	RUN_REQUEST_NOT_HERMETIC = 2	1✔
69	CUSTOM = 3	1✔
70	NOT_SUPPORTED = 4	1✔
71
72
73	@union(in_scope_types=[EnvironmentName])	1✔
74	class RunFieldSet(FieldSet, metaclass=ABCMeta):	1✔
75	"""The fields necessary from a target to run a program/script."""
76
77	supports_debug_adapter: ClassVar[bool] = False	1✔
78	run_in_sandbox_behavior: ClassVar[RunInSandboxBehavior]	1✔
79
80	@final	1✔
81	@classmethod	1✔
82	def rules(cls) -> Iterable[Rule \| UnionRule]:	1✔
UNCOV 83	yield UnionRule(RunFieldSet, cls)	×
UNCOV 84	if not cls.supports_debug_adapter:	×
UNCOV 85	yield from _unsupported_debug_adapter_rules(cls)	×
UNCOV 86	yield from _run_in_sandbox_behavior_rule(cls)	×
87
88
89	class RestartableField(BoolField):	1✔
90	alias = "restartable"	1✔
91	default = False	1✔
92	help = help_text(	1✔
93	"""
94	If true, runs of this target with the `run` goal may be interrupted and
95	restarted when its input files change.
96	"""
97	)
98
99
100	@dataclass(frozen=True)	1✔
101	class RunRequest:	1✔
102	digest: Digest	1✔
103	# Values in args and in env can contain the format specifier "{chroot}", which will
104	# be substituted with the (absolute) chroot path.
105	args: tuple[str, ...]	1✔
106	extra_env: FrozenDict[str, str]	1✔
107	immutable_input_digests: Mapping[str, Digest] \| None = None	1✔
108	append_only_caches: Mapping[str, str] \| None = None	1✔
109
110	def __init__(	1✔
111	self,
112	*,
113	digest: Digest,
114	args: Iterable[str],
115	extra_env: Mapping[str, str] \| None = None,
116	immutable_input_digests: Mapping[str, Digest] \| None = None,
117	append_only_caches: Mapping[str, str] \| None = None,
118	) -> None:
UNCOV 119	object.__setattr__(self, "digest", digest)	×
UNCOV 120	object.__setattr__(self, "args", tuple(args))	×
UNCOV 121	object.__setattr__(self, "extra_env", FrozenDict(extra_env or {}))	×
UNCOV 122	object.__setattr__(	×
123	self, "immutable_input_digests", FrozenDict(immutable_input_digests or {})
124	)
UNCOV 125	object.__setattr__(self, "append_only_caches", FrozenDict(append_only_caches or {}))	×
126
127	def to_run_in_sandbox_request(self) -> RunInSandboxRequest:	1✔
128	return RunInSandboxRequest(	×
129	args=self.args,
130	digest=self.digest,
131	extra_env=self.extra_env,
132	immutable_input_digests=self.immutable_input_digests,
133	append_only_caches=self.append_only_caches,
134	)
135
136
137	class RunDebugAdapterRequest(RunRequest):	1✔
138	"""Like RunRequest, but launches the process using the relevant Debug Adapter server.
139
140	The process should be launched waiting for the client to connect.
141	"""
142
143
144	class RunInSandboxRequest(RunRequest):	1✔
145	"""A run request that launches the process in the sandbox for use as part of a build rule.
146
147	The arguments and environment should only use values relative to the build root (or prefixed
148	with `{chroot}`), or refer to binaries that were fetched with `BinaryPathRequest`.
149
150	Presently, implementors can opt to use the existing as not guaranteeing hermeticity, which will
151	internally mark the rule as uncacheable. In such a case, non-safe APIs can be used, however,
152	this behavior can result in poorer performance, and only exists as a stop-gap while
153	implementors work to make sure their `RunRequest`-generating rules can be used in a hermetic
154	context, or writing new custom rules. (See the Plugin Upgrade Guide for details).
155	"""
156
157
158	@rule(polymorphic=True)	1✔
159	async def generate_run_request(run_field_set: RunFieldSet, env_name: EnvironmentName) -> RunRequest:	1✔
160	raise NotImplementedError()	×
161
162
163	@rule(polymorphic=True)	1✔
164	async def generate_run_debug_adapter_request(	1✔
165	run_field_set: RunFieldSet, env_name: EnvironmentName
166	) -> RunDebugAdapterRequest:
167	raise NotImplementedError()	×
168
169
170	@rule(polymorphic=True)	1✔
171	async def generate_run_in_sandbox_request(	1✔
172	run_field_set: RunFieldSet, env_name: EnvironmentName
173	) -> RunInSandboxRequest:
174	raise NotImplementedError()	×
175
176
177	class RunSubsystem(GoalSubsystem):	1✔
178	name = "run"	1✔
179	help = help_text(	1✔
180	"""
181	Runs a binary target.
182
183	This goal propagates the return code of the underlying executable.
184
185	If your application can safely be restarted while it is running, you can pass
186	`restartable=True` on your binary target (for supported types), and the `run` goal
187	will automatically restart them as all relevant files change. This can be particularly
188	useful for server applications.
189	"""
190	)
191
192	@classmethod	1✔
193	def activated(cls, union_membership: UnionMembership) -> bool:	1✔
194	return RunFieldSet in union_membership	×
195
196	args = ArgsListOption(	1✔
197	example="val1 val2 --debug",
198	tool_name="the executed target",
199	passthrough=True,
200	)
201	# See also `test.py`'s same option
202	debug_adapter = BoolOption(	1✔
203	default=False,
204	help=softwrap(
205	"""
206	Run the interactive process using a Debug Adapter
207	(https://microsoft.github.io/debug-adapter-protocol/) for the language if supported.
208
209	The interactive process used will be immediately blocked waiting for a client before
210	continuing.
211	"""
212	),
213	)
214
215
216	class Run(Goal):	1✔
217	subsystem_cls = RunSubsystem	1✔
218	environment_behavior = Goal.EnvironmentBehavior.LOCAL_ONLY	1✔
219
220
221	async def _find_what_to_run(	1✔
222	goal_description: str,
223	) -> tuple[RunFieldSet, Target]:
UNCOV 224	targets_to_valid_field_sets = await find_valid_field_sets_for_target_roots(	×
225	TargetRootsToFieldSetsRequest(
226	RunFieldSet,
227	goal_description=goal_description,
228	no_applicable_targets_behavior=NoApplicableTargetsBehavior.error,
229	),
230	**implicitly(),
231	)
UNCOV 232	mapping = targets_to_valid_field_sets.mapping	×
233
UNCOV 234	if len(mapping) > 1:	×
UNCOV 235	raise TooManyTargetsException(mapping, goal_description=goal_description)	×
236
UNCOV 237	target, field_sets = next(iter(mapping.items()))	×
UNCOV 238	if len(field_sets) > 1:	×
UNCOV 239	raise AmbiguousImplementationsException(	×
240	target,
241	field_sets,
242	goal_description=goal_description,
243	)
244
UNCOV 245	return field_sets[0], target	×
246
247
248	@goal_rule	1✔
249	async def run(	1✔
250	run_subsystem: RunSubsystem,
251	debug_adapter: DebugAdapterSubsystem,
252	global_options: GlobalOptions,
253	workspace: Workspace, # Needed to enable side-effecting.
254	complete_env: CompleteEnvironmentVars,
255	) -> Run:
UNCOV 256	field_set, target = await _find_what_to_run("the `run` goal")	×
257
UNCOV 258	await _warn_on_non_local_environments((target,), "the `run` goal")	×
259
UNCOV 260	request = await (	×
261	generate_run_request(**implicitly({field_set: RunFieldSet}))
262	if not run_subsystem.debug_adapter
263	else generate_run_debug_adapter_request(**implicitly({field_set: RunFieldSet}))
264	)
UNCOV 265	restartable = target.get(RestartableField).value	×
UNCOV 266	if run_subsystem.debug_adapter:	×
267	logger.info(	×
268	softwrap(
269	f"""
270	Launching debug adapter at '{debug_adapter.host}:{debug_adapter.port}',
271	which will wait for a client connection...
272	"""
273	)
274	)
275
UNCOV 276	result = await run_interactive_process(	×
277	InteractiveProcess(
278	argv=(request.args, run_subsystem.args),
279	env={complete_env, request.extra_env},
280	input_digest=request.digest,
281	run_in_workspace=True,
282	restartable=restartable,
283	keep_sandboxes=global_options.keep_sandboxes,
284	immutable_input_digests=request.immutable_input_digests,
285	append_only_caches=request.append_only_caches,
286	)
287	)
288
UNCOV 289	return Run(result.exit_code)	×
290
291
292	@memoized	1✔
293	def _unsupported_debug_adapter_rules(cls: type[RunFieldSet]) -> Iterable:	1✔
294	"""Returns a rule that implements DebugAdapterRequest by raising an error."""
295
UNCOV 296	@rule(canonical_name_suffix=cls.__name__, _param_type_overrides={"request": cls})	×
UNCOV 297	async def get_run_debug_adapter_request(request: RunFieldSet) -> RunDebugAdapterRequest:	×
298	raise NotImplementedError(	×
299	"Running this target type with a debug adapter is not yet supported."
300	)
301
UNCOV 302	return collect_rules(locals())	×
303
304
305	async def _run_request(request: RunFieldSet) -> RunInSandboxRequest:	1✔
306	run_req = await generate_run_request(**implicitly({request: RunFieldSet}))	×
307	return run_req.to_run_in_sandbox_request()	×
308
309
310	@memoized	1✔
311	def _run_in_sandbox_behavior_rule(cls: type[RunFieldSet]) -> Iterable:	1✔
312	"""Returns a default rule that helps fulfil `experimental_run_in_sandbox` targets.
313
314	If `RunInSandboxBehavior.CUSTOM` is specified, rule implementors must write a rule that returns
315	a `RunInSandboxRequest`.
316	"""
317
UNCOV 318	@rule(canonical_name_suffix=cls.__name__, _param_type_overrides={"request": cls})	×
UNCOV 319	async def not_supported(request: RunFieldSet) -> RunInSandboxRequest:	×
320	raise NotImplementedError(	×
321	"Running this target type within the sandbox is not yet supported."
322	)
323
UNCOV 324	@rule(canonical_name_suffix=cls.__name__, _param_type_overrides={"request": cls})	×
UNCOV 325	async def run_request_hermetic(request: RunFieldSet) -> RunInSandboxRequest:	×
326	return await _run_request(request)	×
327
UNCOV 328	@_uncacheable_rule(canonical_name_suffix=cls.__name__, _param_type_overrides={"request": cls})	×
UNCOV 329	async def run_request_not_hermetic(request: RunFieldSet) -> RunInSandboxRequest:	×
330	return await _run_request(request)	×
331
UNCOV 332	default_rules: dict[RunInSandboxBehavior, list[Any]] = {	×
333	RunInSandboxBehavior.NOT_SUPPORTED: [not_supported],
334	RunInSandboxBehavior.RUN_REQUEST_HERMETIC: [run_request_hermetic],
335	RunInSandboxBehavior.RUN_REQUEST_NOT_HERMETIC: [run_request_not_hermetic],
336	RunInSandboxBehavior.CUSTOM: [],
337	}
338
UNCOV 339	return collect_rules(	×
340	{_rule.__name__: _rule for _rule in default_rules[cls.run_in_sandbox_behavior]}
341	)
342
343
344	def rules():	1✔
UNCOV 345	return collect_rules()	×

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