22740642519

Committed 05 Mar 2026 11:00PM UTC coverage: 52.677% (-40.3%) from 92.931%

Build # 22740642519

Build Type

Pull #23157

github

Committed by

web-flow

Commit Message

Merge 2aa18e6d4 into f0030f5e7

Pull Request Pull Request #23157: [pants ng] Partition source files by config.

Run Details

31678 of 60136 relevant lines covered (52.68%)

0.53 hits per line

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

91.57

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

# Copyright 2020 Pants project contributors (see CONTRIBUTORS.md).
# Licensed under the Apache License, Version 2.0 (see LICENSE).
from __future__ import annotations

import os
from abc import ABC
from collections.abc import Iterable, Mapping, Sequence
from dataclasses import dataclass
from typing import ClassVar

from pants.core.environments.rules import _warn_on_non_local_environments
from pants.engine.addresses import Addresses
from pants.engine.console import Console
from pants.engine.env_vars import CompleteEnvironmentVars
from pants.engine.environment import EnvironmentName
from pants.engine.fs import Digest
from pants.engine.goal import Goal, GoalSubsystem
from pants.engine.intrinsics import run_interactive_process
from pants.engine.process import InteractiveProcess
from pants.engine.rules import collect_rules, goal_rule, implicitly, rule
from pants.engine.target import FilteredTargets, Target
from pants.engine.unions import UnionMembership, union
from pants.option.option_types import ArgsListOption, BoolOption, StrOption
from pants.util.frozendict import FrozenDict
from pants.util.memo import memoized_property
from pants.util.strutil import softwrap


@union(in_scope_types=[EnvironmentName])
@dataclass(frozen=True)
class ReplImplementation(ABC):
    """A REPL implementation for a specific language or runtime.

    Proxies from the top-level `repl` goal to an actual implementation.
    """

    name: ClassVar[str]
    supports_args: ClassVar[bool]

    targets: Sequence[Target]

    def in_chroot(self, relpath: str) -> str:
        return os.path.join("{chroot}", relpath)

    @memoized_property
    def addresses(self) -> Addresses:
        return Addresses(t.address for t in self.targets)


class ReplSubsystem(GoalSubsystem):
    name = "repl"
    help = "Open a REPL with the specified code loadable."

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

    shell = StrOption(
        default=None,
        help="Override the automatically-detected REPL program for the target(s) specified.",
    )
    args = ArgsListOption(
        example="-i helloworld/main.py",
        tool_name="the repl program",
        passthrough=True,
        extra_help="Currently supported only for the ipython shell.",
    )
    restartable = BoolOption(
        default=False,
        help="True if the REPL should be restarted if its inputs have changed.",
    )


class Repl(Goal):
    subsystem_cls = ReplSubsystem
    environment_behavior = Goal.EnvironmentBehavior.LOCAL_ONLY


@dataclass(frozen=True)
class ReplRequest:
    digest: Digest
    args: tuple[str, ...]
    extra_env: FrozenDict[str, str]
    immutable_input_digests: FrozenDict[str, Digest]
    append_only_caches: FrozenDict[str, str]
    run_in_workspace: bool

    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,
        run_in_workspace: bool = True,
    ) -> 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 {}))
        object.__setattr__(self, "run_in_workspace", run_in_workspace)


@rule(polymorphic=True)
async def get_repl_request(
    impl: ReplImplementation,
    environment_name: EnvironmentName,
) -> ReplRequest:
    raise NotImplementedError()


@goal_rule
async def run_repl(
    console: Console,
    repl_subsystem: ReplSubsystem,
    specified_targets: FilteredTargets,
    union_membership: UnionMembership,
    complete_env: CompleteEnvironmentVars,
) -> Repl:
    await _warn_on_non_local_environments(specified_targets, "the `repl` goal")

    # TODO: When we support multiple languages, detect the default repl to use based
    #  on the targets.  For now we default to the python repl.
    repl_shell_name = repl_subsystem.shell or "python"
    implementations = {impl.name: impl for impl in union_membership[ReplImplementation]}
    repl_implementation_cls = implementations.get(repl_shell_name)
    if repl_implementation_cls is None:
        available = sorted(implementations.keys())
        console.print_stderr(
            softwrap(
                f"""
                {repr(repl_shell_name)} is not a registered REPL. Available REPLs (which may
                be specified through the option `--repl-shell`): {available}
                """
            )
        )
        return Repl(-1)

    if repl_subsystem.args and not repl_implementation_cls.supports_args:
        console.print_stderr(
            softwrap(
                f"""
                REPL goal does not support passing args to a {repr(repl_shell_name)} shell.
                """
            )
        )
        return Repl(-1)

    repl_impl = repl_implementation_cls(targets=specified_targets)
    request = await get_repl_request(**implicitly({repl_impl: ReplImplementation}))

    env = {**complete_env, **request.extra_env}

    result = await run_interactive_process(
        InteractiveProcess(
            argv=(*request.args, *repl_subsystem.args),
            env=env,
            input_digest=request.digest,
            run_in_workspace=request.run_in_workspace,
            restartable=repl_subsystem.restartable,
            immutable_input_digests=request.immutable_input_digests,
            append_only_caches=request.append_only_caches,
        ),
    )
    return Repl(result.exit_code)


def rules():
    return collect_rules()

1	# Copyright 2020 Pants project contributors (see CONTRIBUTORS.md).
2	# Licensed under the Apache License, Version 2.0 (see LICENSE).
3	from __future__ import annotations	1✔
4
5	import os	1✔
6	from abc import ABC	1✔
7	from collections.abc import Iterable, Mapping, Sequence	1✔
8	from dataclasses import dataclass	1✔
9	from typing import ClassVar	1✔
10
11	from pants.core.environments.rules import _warn_on_non_local_environments	1✔
12	from pants.engine.addresses import Addresses	1✔
13	from pants.engine.console import Console	1✔
14	from pants.engine.env_vars import CompleteEnvironmentVars	1✔
15	from pants.engine.environment import EnvironmentName	1✔
16	from pants.engine.fs import Digest	1✔
17	from pants.engine.goal import Goal, GoalSubsystem	1✔
18	from pants.engine.intrinsics import run_interactive_process	1✔
19	from pants.engine.process import InteractiveProcess	1✔
20	from pants.engine.rules import collect_rules, goal_rule, implicitly, rule	1✔
21	from pants.engine.target import FilteredTargets, Target	1✔
22	from pants.engine.unions import UnionMembership, union	1✔
23	from pants.option.option_types import ArgsListOption, BoolOption, StrOption	1✔
24	from pants.util.frozendict import FrozenDict	1✔
25	from pants.util.memo import memoized_property	1✔
26	from pants.util.strutil import softwrap	1✔
27
28
29	@union(in_scope_types=[EnvironmentName])	1✔
30	@dataclass(frozen=True)	1✔
31	class ReplImplementation(ABC):	1✔
32	"""A REPL implementation for a specific language or runtime.
33
34	Proxies from the top-level `repl` goal to an actual implementation.
35	"""
36
37	name: ClassVar[str]	1✔
38	supports_args: ClassVar[bool]	1✔
39
40	targets: Sequence[Target]	1✔
41
42	def in_chroot(self, relpath: str) -> str:	1✔
43	return os.path.join("{chroot}", relpath)	1✔
44
45	@memoized_property	1✔
46	def addresses(self) -> Addresses:	1✔
47	return Addresses(t.address for t in self.targets)	1✔
48
49
50	class ReplSubsystem(GoalSubsystem):	1✔
51	name = "repl"	1✔
52	help = "Open a REPL with the specified code loadable."	1✔
53
54	@classmethod	1✔
55	def activated(cls, union_membership: UnionMembership) -> bool:	1✔
56	return ReplImplementation in union_membership	×
57
58	shell = StrOption(	1✔
59	default=None,
60	help="Override the automatically-detected REPL program for the target(s) specified.",
61	)
62	args = ArgsListOption(	1✔
63	example="-i helloworld/main.py",
64	tool_name="the repl program",
65	passthrough=True,
66	extra_help="Currently supported only for the ipython shell.",
67	)
68	restartable = BoolOption(	1✔
69	default=False,
70	help="True if the REPL should be restarted if its inputs have changed.",
71	)
72
73
74	class Repl(Goal):	1✔
75	subsystem_cls = ReplSubsystem	1✔
76	environment_behavior = Goal.EnvironmentBehavior.LOCAL_ONLY	1✔
77
78
79	@dataclass(frozen=True)	1✔
80	class ReplRequest:	1✔
81	digest: Digest	1✔
82	args: tuple[str, ...]	1✔
83	extra_env: FrozenDict[str, str]	1✔
84	immutable_input_digests: FrozenDict[str, Digest]	1✔
85	append_only_caches: FrozenDict[str, str]	1✔
86	run_in_workspace: bool	1✔
87
88	def __init__(	1✔
89	self,
90	*,
91	digest: Digest,
92	args: Iterable[str],
93	extra_env: Mapping[str, str] \| None = None,
94	immutable_input_digests: Mapping[str, Digest] \| None = None,
95	append_only_caches: Mapping[str, str] \| None = None,
96	run_in_workspace: bool = True,
97	) -> None:
98	object.__setattr__(self, "digest", digest)	1✔
99	object.__setattr__(self, "args", tuple(args))	1✔
100	object.__setattr__(self, "extra_env", FrozenDict(extra_env or {}))	1✔
101	object.__setattr__(	1✔
102	self, "immutable_input_digests", FrozenDict(immutable_input_digests or {})
103	)
104	object.__setattr__(self, "append_only_caches", FrozenDict(append_only_caches or {}))	1✔
105	object.__setattr__(self, "run_in_workspace", run_in_workspace)	1✔
106
107
108	@rule(polymorphic=True)	1✔
109	async def get_repl_request(	1✔
110	impl: ReplImplementation,
111	environment_name: EnvironmentName,
112	) -> ReplRequest:
113	raise NotImplementedError()	×
114
115
116	@goal_rule	1✔
117	async def run_repl(	1✔
118	console: Console,
119	repl_subsystem: ReplSubsystem,
120	specified_targets: FilteredTargets,
121	union_membership: UnionMembership,
122	complete_env: CompleteEnvironmentVars,
123	) -> Repl:
124	await _warn_on_non_local_environments(specified_targets, "the `repl` goal")	1✔
125
126	# TODO: When we support multiple languages, detect the default repl to use based
127	# on the targets. For now we default to the python repl.
128	repl_shell_name = repl_subsystem.shell or "python"	1✔
129	implementations = {impl.name: impl for impl in union_membership[ReplImplementation]}	1✔
130	repl_implementation_cls = implementations.get(repl_shell_name)	1✔
131	if repl_implementation_cls is None:	1✔
132	available = sorted(implementations.keys())	×
133	console.print_stderr(	×
134	softwrap(
135	f"""
136	{repr(repl_shell_name)} is not a registered REPL. Available REPLs (which may
137	be specified through the option `--repl-shell`): {available}
138	"""
139	)
140	)
141	return Repl(-1)	×
142
143	if repl_subsystem.args and not repl_implementation_cls.supports_args:	1✔
144	console.print_stderr(	×
145	softwrap(
146	f"""
147	REPL goal does not support passing args to a {repr(repl_shell_name)} shell.
148	"""
149	)
150	)
151	return Repl(-1)	×
152
153	repl_impl = repl_implementation_cls(targets=specified_targets)	1✔
154	request = await get_repl_request(**implicitly({repl_impl: ReplImplementation}))	1✔
155
156	env = {complete_env, request.extra_env}	1✔
157
158	result = await run_interactive_process(	1✔
159	InteractiveProcess(
160	argv=(request.args, repl_subsystem.args),
161	env=env,
162	input_digest=request.digest,
163	run_in_workspace=request.run_in_workspace,
164	restartable=repl_subsystem.restartable,
165	immutable_input_digests=request.immutable_input_digests,
166	append_only_caches=request.append_only_caches,
167	),
168	)
169	return Repl(result.exit_code)	1✔
170
171
172	def rules():	1✔
173	return collect_rules()	1✔

pantsbuild / pants / 22740642519

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