25441711719

Committed 06 May 2026 02:31PM UTC coverage: 92.915%. Remained the same

Build # 25441711719

Build Type

push

github

Committed by

web-flow

Commit Message

use sha pin (with comment) format for generated actions (#23312)

Per the GitHub Action best practices we recently enabled at #23249, we
should pin each action to a SHA so that the reference is actually
immutable.

This will -- I hope -- knock out a large chunk of the 421 alerts we
currently get from zizmor. The next followup would then be upgrades and
harmonizing the generated and none-generated pins.

Notice: This idea was suggested by Claude while going over pinact output
and I was surprised to see that post processing the yaml wasn't too
gross.

Coverage Stats

92206 of 99237 relevant lines covered (92.91%)

4.04 hits per line

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

95.18

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

pantsbuild / pants / 25441711719

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