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.24

/src/python/pants/backend/shell/dependency_inference.py

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

from __future__ import annotations

import json
import logging
import os
import re
from collections import defaultdict
from collections.abc import Iterable
from dataclasses import dataclass
from typing import DefaultDict

from pants.backend.shell.lint.shellcheck.subsystem import Shellcheck
from pants.backend.shell.subsystems.shell_setup import ShellSetup
from pants.backend.shell.target_types import ShellDependenciesField, ShellSourceField
from pants.core.util_rules.external_tool import download_external_tool
from pants.engine.addresses import Address
from pants.engine.collection import DeduplicatedCollection
from pants.engine.fs import Digest
from pants.engine.internals.graph import determine_explicitly_provided_dependencies, hydrate_sources
from pants.engine.intrinsics import execute_process
from pants.engine.platform import Platform
from pants.engine.process import Process, ProcessCacheScope
from pants.engine.rules import Rule, collect_rules, concurrently, implicitly, rule
from pants.engine.target import (
    AllTargets,
    DependenciesRequest,
    FieldSet,
    HydrateSourcesRequest,
    InferDependenciesRequest,
    InferredDependencies,
    Targets,
)
from pants.engine.unions import UnionRule
from pants.util.frozendict import FrozenDict
from pants.util.logging import LogLevel
from pants.util.ordered_set import OrderedSet

logger = logging.getLogger(__name__)


class AllShellTargets(Targets):
    pass


@rule(desc="Find all Shell targets in project", level=LogLevel.DEBUG)
async def find_all_shell_targets(all_tgts: AllTargets) -> AllShellTargets:
    return AllShellTargets(tgt for tgt in all_tgts if tgt.has_field(ShellSourceField))


@dataclass(frozen=True)
class ShellMapping:
    """A mapping of Shell file names to their owning file address."""

    mapping: FrozenDict[str, Address]
    ambiguous_modules: FrozenDict[str, tuple[Address, ...]]


@rule(desc="Creating map of Shell file names to Shell targets", level=LogLevel.DEBUG)
async def map_shell_files(tgts: AllShellTargets) -> ShellMapping:
    files_to_addresses: dict[str, Address] = {}
    files_with_multiple_owners: DefaultDict[str, set[Address]] = defaultdict(set)
    for tgt in tgts:
        fp = tgt[ShellSourceField].file_path
        if fp in files_to_addresses:
            files_with_multiple_owners[fp].update({files_to_addresses[fp], tgt.address})
        else:
            files_to_addresses[fp] = tgt.address

    # Remove files with ambiguous owners.
    for ambiguous_f in files_with_multiple_owners:
        files_to_addresses.pop(ambiguous_f)

    return ShellMapping(
        mapping=FrozenDict(sorted(files_to_addresses.items())),
        ambiguous_modules=FrozenDict(
            (k, tuple(sorted(v))) for k, v in sorted(files_with_multiple_owners.items())
        ),
    )


class ParsedShellImports(DeduplicatedCollection):
    sort_input = True


@dataclass(frozen=True)
class ParseShellImportsRequest:
    digest: Digest
    fp: str


PATH_FROM_SHELLCHECK_ERROR = re.compile(r"Not following: (.+) was not specified as input")


@rule
async def parse_shell_imports(
    request: ParseShellImportsRequest, shellcheck: Shellcheck, platform: Platform
) -> ParsedShellImports:
    # We use Shellcheck to parse for us by running it against each file in isolation, which means
    # that all `source` statements will error. Then, we can extract the problematic paths from the
    # JSON output.
    downloaded_shellcheck = await download_external_tool(shellcheck.get_request(platform))

    immutable_input_key = "__shellcheck_tool"
    exe_path = os.path.join(immutable_input_key, downloaded_shellcheck.exe)

    process_result = await execute_process(
        Process(
            # NB: We do not load up `[shellcheck].{args,config}` because it would risk breaking
            # determinism of dependency inference in an unexpected way.
            [exe_path, "--format=json", request.fp],
            input_digest=request.digest,
            immutable_input_digests={immutable_input_key: downloaded_shellcheck.digest},
            description=f"Detect Shell imports for {request.fp}",
            level=LogLevel.DEBUG,
            # We expect this to always fail, but it should still be cached because the process is
            # deterministic.
            cache_scope=ProcessCacheScope.ALWAYS,
        ),
        **implicitly(),
    )

    try:
        output = json.loads(process_result.stdout)
    except json.JSONDecodeError:
        logger.error(
            f"Parsing {request.fp} for dependency inference failed because Shellcheck's output "
            f"could not be loaded as JSON. Please open a GitHub issue at "
            f"https://github.com/pantsbuild/pants/issues/new with this error message attached.\n\n"
            f"\nshellcheck version: {shellcheck.version}\n"
            f"process_result.stdout: {process_result.stdout.decode()}"
        )
        return ParsedShellImports()

    paths = set()
    for error in output:
        if not error.get("code", "") == 1091:
            continue
        msg = error.get("message", "")
        matches = PATH_FROM_SHELLCHECK_ERROR.match(msg)
        if matches:
            paths.add(matches.group(1))
        else:
            logger.error(
                f"Parsing {request.fp} for dependency inference failed because Shellcheck's error "
                f"message was not in the expected format. Please open a GitHub issue at "
                f"https://github.com/pantsbuild/pants/issues/new with this error message "
                f"attached.\n\n\nshellcheck version: {shellcheck.version}\n"
                f"error JSON entry: {error}"
            )
    return ParsedShellImports(paths)


@dataclass(frozen=True)
class ShellDependenciesInferenceFieldSet(FieldSet):
    required_fields = (ShellSourceField, ShellDependenciesField)

    source: ShellSourceField
    dependencies: ShellDependenciesField


class InferShellDependencies(InferDependenciesRequest):
    infer_from = ShellDependenciesInferenceFieldSet


@rule(desc="Inferring Shell dependencies by analyzing imports")
async def infer_shell_dependencies(
    request: InferShellDependencies, shell_mapping: ShellMapping, shell_setup: ShellSetup
) -> InferredDependencies:
    if not shell_setup.dependency_inference:
        return InferredDependencies([])

    address = request.field_set.address
    explicitly_provided_deps, hydrated_sources = await concurrently(
        determine_explicitly_provided_dependencies(
            **implicitly(DependenciesRequest(request.field_set.dependencies))
        ),
        hydrate_sources(HydrateSourcesRequest(request.field_set.source), **implicitly()),
    )
    assert len(hydrated_sources.snapshot.files) == 1

    detected_imports = await parse_shell_imports(
        ParseShellImportsRequest(
            hydrated_sources.snapshot.digest, hydrated_sources.snapshot.files[0]
        ),
        **implicitly(),
    )
    result: OrderedSet[Address] = OrderedSet()
    for import_path in detected_imports:
        unambiguous = shell_mapping.mapping.get(import_path)
        ambiguous = shell_mapping.ambiguous_modules.get(import_path)
        if unambiguous:
            result.add(unambiguous)
        elif ambiguous:
            explicitly_provided_deps.maybe_warn_of_ambiguous_dependency_inference(
                ambiguous,
                address,
                import_reference="file",
                context=f"The target {address} sources `{import_path}`",
            )
            maybe_disambiguated = explicitly_provided_deps.disambiguated(ambiguous)
            if maybe_disambiguated:
                result.add(maybe_disambiguated)
    return InferredDependencies(sorted(result))


def rules() -> Iterable[Rule | UnionRule]:
    return (*collect_rules(), UnionRule(InferDependenciesRequest, InferShellDependencies))

1	# Copyright 2021 Pants project contributors (see CONTRIBUTORS.md).
2	# Licensed under the Apache License, Version 2.0 (see LICENSE).
3
4	from __future__ import annotations	4✔
5
6	import json	4✔
7	import logging	4✔
8	import os	4✔
9	import re	4✔
10	from collections import defaultdict	4✔
11	from collections.abc import Iterable	4✔
12	from dataclasses import dataclass	4✔
13	from typing import DefaultDict	4✔
14
15	from pants.backend.shell.lint.shellcheck.subsystem import Shellcheck	4✔
16	from pants.backend.shell.subsystems.shell_setup import ShellSetup	4✔
17	from pants.backend.shell.target_types import ShellDependenciesField, ShellSourceField	4✔
18	from pants.core.util_rules.external_tool import download_external_tool	4✔
19	from pants.engine.addresses import Address	4✔
20	from pants.engine.collection import DeduplicatedCollection	4✔
21	from pants.engine.fs import Digest	4✔
22	from pants.engine.internals.graph import determine_explicitly_provided_dependencies, hydrate_sources	4✔
23	from pants.engine.intrinsics import execute_process	4✔
24	from pants.engine.platform import Platform	4✔
25	from pants.engine.process import Process, ProcessCacheScope	4✔
26	from pants.engine.rules import Rule, collect_rules, concurrently, implicitly, rule	4✔
27	from pants.engine.target import (	4✔
28	AllTargets,
29	DependenciesRequest,
30	FieldSet,
31	HydrateSourcesRequest,
32	InferDependenciesRequest,
33	InferredDependencies,
34	Targets,
35	)
36	from pants.engine.unions import UnionRule	4✔
37	from pants.util.frozendict import FrozenDict	4✔
38	from pants.util.logging import LogLevel	4✔
39	from pants.util.ordered_set import OrderedSet	4✔
40
41	logger = logging.getLogger(__name__)	4✔
42
43
44	class AllShellTargets(Targets):	4✔
45	pass	4✔
46
47
48	@rule(desc="Find all Shell targets in project", level=LogLevel.DEBUG)	4✔
49	async def find_all_shell_targets(all_tgts: AllTargets) -> AllShellTargets:	4✔
50	return AllShellTargets(tgt for tgt in all_tgts if tgt.has_field(ShellSourceField))	1✔
51
52
53	@dataclass(frozen=True)	4✔
54	class ShellMapping:	4✔
55	"""A mapping of Shell file names to their owning file address."""
56
57	mapping: FrozenDict[str, Address]	4✔
58	ambiguous_modules: FrozenDict[str, tuple[Address, ...]]	4✔
59
60
61	@rule(desc="Creating map of Shell file names to Shell targets", level=LogLevel.DEBUG)	4✔
62	async def map_shell_files(tgts: AllShellTargets) -> ShellMapping:	4✔
63	files_to_addresses: dict[str, Address] = {}	1✔
64	files_with_multiple_owners: DefaultDict[str, set[Address]] = defaultdict(set)	1✔
65	for tgt in tgts:	1✔
66	fp = tgt[ShellSourceField].file_path	1✔
67	if fp in files_to_addresses:	1✔
68	files_with_multiple_owners[fp].update({files_to_addresses[fp], tgt.address})	1✔
69	else:
70	files_to_addresses[fp] = tgt.address	1✔
71
72	# Remove files with ambiguous owners.
73	for ambiguous_f in files_with_multiple_owners:	1✔
74	files_to_addresses.pop(ambiguous_f)	1✔
75
76	return ShellMapping(	1✔
77	mapping=FrozenDict(sorted(files_to_addresses.items())),
78	ambiguous_modules=FrozenDict(
79	(k, tuple(sorted(v))) for k, v in sorted(files_with_multiple_owners.items())
80	),
81	)
82
83
84	class ParsedShellImports(DeduplicatedCollection):	4✔
85	sort_input = True	4✔
86
87
88	@dataclass(frozen=True)	4✔
89	class ParseShellImportsRequest:	4✔
90	digest: Digest	4✔
91	fp: str	4✔
92
93
94	PATH_FROM_SHELLCHECK_ERROR = re.compile(r"Not following: (.+) was not specified as input")	4✔
95
96
97	@rule	4✔
98	async def parse_shell_imports(	4✔
99	request: ParseShellImportsRequest, shellcheck: Shellcheck, platform: Platform
100	) -> ParsedShellImports:
101	# We use Shellcheck to parse for us by running it against each file in isolation, which means
102	# that all `source` statements will error. Then, we can extract the problematic paths from the
103	# JSON output.
104	downloaded_shellcheck = await download_external_tool(shellcheck.get_request(platform))	1✔
105
106	immutable_input_key = "__shellcheck_tool"	1✔
107	exe_path = os.path.join(immutable_input_key, downloaded_shellcheck.exe)	1✔
108
109	process_result = await execute_process(	1✔
110	Process(
111	# NB: We do not load up `[shellcheck].{args,config}` because it would risk breaking
112	# determinism of dependency inference in an unexpected way.
113	[exe_path, "--format=json", request.fp],
114	input_digest=request.digest,
115	immutable_input_digests={immutable_input_key: downloaded_shellcheck.digest},
116	description=f"Detect Shell imports for {request.fp}",
117	level=LogLevel.DEBUG,
118	# We expect this to always fail, but it should still be cached because the process is
119	# deterministic.
120	cache_scope=ProcessCacheScope.ALWAYS,
121	),
122	**implicitly(),
123	)
124
125	try:	1✔
126	output = json.loads(process_result.stdout)	1✔
127	except json.JSONDecodeError:	×
128	logger.error(	×
129	f"Parsing {request.fp} for dependency inference failed because Shellcheck's output "
130	f"could not be loaded as JSON. Please open a GitHub issue at "
131	f"https://github.com/pantsbuild/pants/issues/new with this error message attached.\n\n"
132	f"\nshellcheck version: {shellcheck.version}\n"
133	f"process_result.stdout: {process_result.stdout.decode()}"
134	)
135	return ParsedShellImports()	×
136
137	paths = set()	1✔
138	for error in output:	1✔
139	if not error.get("code", "") == 1091:	1✔
140	continue	1✔
141	msg = error.get("message", "")	1✔
142	matches = PATH_FROM_SHELLCHECK_ERROR.match(msg)	1✔
143	if matches:	1✔
144	paths.add(matches.group(1))	1✔
145	else:
146	logger.error(	×
147	f"Parsing {request.fp} for dependency inference failed because Shellcheck's error "
148	f"message was not in the expected format. Please open a GitHub issue at "
149	f"https://github.com/pantsbuild/pants/issues/new with this error message "
150	f"attached.\n\n\nshellcheck version: {shellcheck.version}\n"
151	f"error JSON entry: {error}"
152	)
153	return ParsedShellImports(paths)	1✔
154
155
156	@dataclass(frozen=True)	4✔
157	class ShellDependenciesInferenceFieldSet(FieldSet):	4✔
158	required_fields = (ShellSourceField, ShellDependenciesField)	4✔
159
160	source: ShellSourceField	4✔
161	dependencies: ShellDependenciesField	4✔
162
163
164	class InferShellDependencies(InferDependenciesRequest):	4✔
165	infer_from = ShellDependenciesInferenceFieldSet	4✔
166
167
168	@rule(desc="Inferring Shell dependencies by analyzing imports")	4✔
169	async def infer_shell_dependencies(	4✔
170	request: InferShellDependencies, shell_mapping: ShellMapping, shell_setup: ShellSetup
171	) -> InferredDependencies:
172	if not shell_setup.dependency_inference:	1✔
173	return InferredDependencies([])	×
174
175	address = request.field_set.address	1✔
176	explicitly_provided_deps, hydrated_sources = await concurrently(	1✔
177	determine_explicitly_provided_dependencies(
178	**implicitly(DependenciesRequest(request.field_set.dependencies))
179	),
180	hydrate_sources(HydrateSourcesRequest(request.field_set.source), **implicitly()),
181	)
182	assert len(hydrated_sources.snapshot.files) == 1	1✔
183
184	detected_imports = await parse_shell_imports(	1✔
185	ParseShellImportsRequest(
186	hydrated_sources.snapshot.digest, hydrated_sources.snapshot.files[0]
187	),
188	**implicitly(),
189	)
190	result: OrderedSet[Address] = OrderedSet()	1✔
191	for import_path in detected_imports:	1✔
192	unambiguous = shell_mapping.mapping.get(import_path)	1✔
193	ambiguous = shell_mapping.ambiguous_modules.get(import_path)	1✔
194	if unambiguous:	1✔
195	result.add(unambiguous)	1✔
196	elif ambiguous:	1✔
197	explicitly_provided_deps.maybe_warn_of_ambiguous_dependency_inference(	1✔
198	ambiguous,
199	address,
200	import_reference="file",
201	context=f"The target {address} sources `{import_path}`",
202	)
203	maybe_disambiguated = explicitly_provided_deps.disambiguated(ambiguous)	1✔
204	if maybe_disambiguated:	1✔
205	result.add(maybe_disambiguated)	1✔
206	return InferredDependencies(sorted(result))	1✔
207
208
209	def rules() -> Iterable[Rule \| UnionRule]:	4✔
210	return (*collect_rules(), UnionRule(InferDependenciesRequest, InferShellDependencies))	4✔

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