25259185675

Committed 02 May 2026 06:47PM UTC coverage: 92.141% (-0.8%) from 92.955%

Build # 25259185675

Build Type

push

github

Committed by

web-flow

Commit Message

Fix the dynamic UI. (#23306)

In #23114 we upgraded to indicatif 0.18.4,
which included a fix to respect TERM, and 
display nothing if it's unset.

Since we did not pass TERM through pantsd, the
dynamic ui is now not shown. 

This change fixes that, and also pass NO_COLOR
through, since indicatif inspects it too.

Coverage Stats

88773 of 96345 relevant lines covered (92.14%)

3.83 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	9✔
4
5	import os	9✔
6	from abc import ABC	9✔
7	from collections.abc import Iterable, Mapping, Sequence	9✔
8	from dataclasses import dataclass	9✔
9	from typing import ClassVar	9✔
10
11	from pants.core.environments.rules import _warn_on_non_local_environments	9✔
12	from pants.engine.addresses import Addresses	9✔
13	from pants.engine.console import Console	9✔
14	from pants.engine.env_vars import CompleteEnvironmentVars	9✔
15	from pants.engine.environment import EnvironmentName	9✔
16	from pants.engine.fs import Digest	9✔
17	from pants.engine.goal import Goal, GoalSubsystem	9✔
18	from pants.engine.intrinsics import run_interactive_process	9✔
19	from pants.engine.process import InteractiveProcess	9✔
20	from pants.engine.rules import collect_rules, goal_rule, implicitly, rule	9✔
21	from pants.engine.target import FilteredTargets, Target	9✔
22	from pants.engine.unions import UnionMembership, union	9✔
23	from pants.option.option_types import ArgsListOption, BoolOption, StrOption	9✔
24	from pants.util.frozendict import FrozenDict	9✔
25	from pants.util.memo import memoized_property	9✔
26	from pants.util.strutil import softwrap	9✔
27
28
29	@union(in_scope_types=[EnvironmentName])	9✔
30	@dataclass(frozen=True)	9✔
31	class ReplImplementation(ABC):	9✔
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]	9✔
38	supports_args: ClassVar[bool]	9✔
39
40	targets: Sequence[Target]	9✔
41
42	def in_chroot(self, relpath: str) -> str:	9✔
43	return os.path.join("{chroot}", relpath)	3✔
44
45	@memoized_property	9✔
46	def addresses(self) -> Addresses:	9✔
47	return Addresses(t.address for t in self.targets)	3✔
48
49
50	class ReplSubsystem(GoalSubsystem):	9✔
51	name = "repl"	9✔
52	help = "Open a REPL with the specified code loadable."	9✔
53
54	@classmethod	9✔
55	def activated(cls, union_membership: UnionMembership) -> bool:	9✔
56	return ReplImplementation in union_membership	×
57
58	shell = StrOption(	9✔
59	default=None,
60	help="Override the automatically-detected REPL program for the target(s) specified.",
61	)
62	args = ArgsListOption(	9✔
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(	9✔
69	default=False,
70	help="True if the REPL should be restarted if its inputs have changed.",
71	)
72
73
74	class Repl(Goal):	9✔
75	subsystem_cls = ReplSubsystem	9✔
76	environment_behavior = Goal.EnvironmentBehavior.LOCAL_ONLY	9✔
77
78
79	@dataclass(frozen=True)	9✔
80	class ReplRequest:	9✔
81	digest: Digest	9✔
82	args: tuple[str, ...]	9✔
83	extra_env: FrozenDict[str, str]	9✔
84	immutable_input_digests: FrozenDict[str, Digest]	9✔
85	append_only_caches: FrozenDict[str, str]	9✔
86	run_in_workspace: bool	9✔
87
88	def __init__(	9✔
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)	3✔
99	object.__setattr__(self, "args", tuple(args))	3✔
100	object.__setattr__(self, "extra_env", FrozenDict(extra_env or {}))	3✔
101	object.__setattr__(	3✔
102	self, "immutable_input_digests", FrozenDict(immutable_input_digests or {})
103	)
104	object.__setattr__(self, "append_only_caches", FrozenDict(append_only_caches or {}))	3✔
105	object.__setattr__(self, "run_in_workspace", run_in_workspace)	3✔
106
107
108	@rule(polymorphic=True)	9✔
109	async def get_repl_request(	9✔
110	impl: ReplImplementation,
111	environment_name: EnvironmentName,
112	) -> ReplRequest:
113	raise NotImplementedError()	×
114
115
116	@goal_rule	9✔
117	async def run_repl(	9✔
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")	3✔
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"	3✔
129	implementations = {impl.name: impl for impl in union_membership[ReplImplementation]}	3✔
130	repl_implementation_cls = implementations.get(repl_shell_name)	3✔
131	if repl_implementation_cls is None:	3✔
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:	3✔
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)	3✔
154	request = await get_repl_request(**implicitly({repl_impl: ReplImplementation}))	3✔
155
156	env = {complete_env, request.extra_env}	3✔
157
158	result = await run_interactive_process(	3✔
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)	3✔
170
171
172	def rules():	9✔
173	return collect_rules()	9✔

pantsbuild / pants / 25259185675

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