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

90.91

/src/python/pants/ng/source_partition.py

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

from __future__ import annotations

import os
from dataclasses import dataclass
from pathlib import Path

from pants.base.glob_match_error_behavior import GlobMatchErrorBehavior
from pants.engine.collection import Collection
from pants.engine.fs import GlobExpansionConjunction, PathGlobs, PathMetadataRequest
from pants.engine.internals.native_engine import (
    Digest,
    PathMetadataKind,
    PyNgOptions,
    PyNgOptionsReader,
    PyNgSourcePartition,
)
from pants.engine.internals.session import SessionValues
from pants.engine.intrinsics import path_globs_to_digest, path_globs_to_paths, path_metadata_request
from pants.engine.rules import Rule, _uncacheable_rule, collect_rules, implicitly, rule
from pants.source.source_root import SourceRoot, SourceRootsRequest, get_source_roots
from pants.util.memo import memoized_property


@dataclass(frozen=True)
class SourcePaths:
    """A set of sources, under a single source root."""

    paths: tuple[Path, ...]
    source_root: SourceRoot

    def path_strs(self) -> tuple[str, ...]:
        return tuple(str(path) for path in self.paths)

    def filter_by_suffixes(self, suffixes: tuple[str, ...]) -> SourcePaths:
        suffixes_set = set(suffixes)
        return SourcePaths(
            tuple(path for path in self.paths if path.suffix in suffixes_set),
            self.source_root,
        )


@dataclass(frozen=True)
class CommonDir:
    path: Path | None


@rule
async def find_common_dir(source_paths: SourcePaths) -> CommonDir:
    if not source_paths.paths:  # We don't expect empty SourcePaths, but might as well be robust.
        return CommonDir(None)
    commonpath = os.path.commonpath(source_paths.paths)
    common_dir = commonpath
    if len(source_paths.paths) == 1:
        # If there is only one path, and it's a file, then commonpath will be that file, whereas
        # we want its enclosing dir. So detect that case.
        meta = await path_metadata_request(PathMetadataRequest(commonpath))
        # Chase any symlinks back to the final path they point to.
        while (
            meta.metadata
            and meta.metadata.kind == PathMetadataKind.SYMLINK
            and meta.metadata.symlink_target
        ):
            # NB: We don't `normpath` because eliminating `..` might change the meaning of the path
            #  if any of the intermediate directories are themselves symlinks.
            symlink_target = os.path.join(os.path.dirname(commonpath), meta.metadata.symlink_target)
            meta = await path_metadata_request(PathMetadataRequest(symlink_target))
        if meta.metadata and meta.metadata.kind == PathMetadataKind.FILE:
            common_dir = os.path.dirname(commonpath)
        else:
            common_dir = commonpath
    return CommonDir(Path(common_dir))


@dataclass(frozen=True)
class SourceDigest:
    digest: Digest


@rule
async def source_paths_to_digest(source_paths: SourcePaths) -> SourceDigest:
    source_digest = await path_globs_to_digest(
        PathGlobs(
            source_paths.path_strs(),
            glob_match_error_behavior=GlobMatchErrorBehavior.error,
            conjunction=GlobExpansionConjunction.all_match,
            description_of_origin="Input source paths",
        )
    )
    return SourceDigest(source_digest)


@dataclass(frozen=True)
class SourcePartition:
    """Access to source files and the config that goes with them."""

    _native_partition: PyNgSourcePartition
    _source_root: SourceRoot

    @memoized_property
    def source_paths(self) -> SourcePaths:
        return SourcePaths(
            tuple(Path(p) for p in self._native_partition.paths()), self._source_root
        )

    @memoized_property
    def options_reader(self) -> PyNgOptionsReader:
        return self._native_partition.options_reader()


class SourcePartitions(Collection[SourcePartition]):
    pass


# Uncacheable because we must get the most recent session value on each run.
@_uncacheable_rule
async def get_ng_options(session_values: SessionValues) -> PyNgOptions:
    return session_values[PyNgOptions]


# Uncacheable because we must recompute on each run.
@_uncacheable_rule
async def partition_sources(path_globs: PathGlobs) -> SourcePartitions:
    options = await get_ng_options(**implicitly())
    paths = await path_globs_to_paths(path_globs)
    # First partition by source root.
    source_roots = await get_source_roots(SourceRootsRequest.for_files(paths.files))
    root_to_paths = source_roots.root_to_paths()
    partitions: list[SourcePartition] = []
    for source_root, paths_in_partition in root_to_paths.items():
        # Then subpartition each of those by config.
        partitions.extend(
            SourcePartition(native_part, source_root)
            for native_part in options.partition_sources(
                tuple(str(path) for path in paths_in_partition)
            )
        )
    return SourcePartitions(tuple(partitions))


@rule
async def get_source_paths(partition: SourcePartition) -> SourcePaths:
    return partition.source_paths


def rules() -> tuple[Rule, ...]:
    return (*collect_rules(),)

1	# Copyright 2025 Pants project contributors (see CONTRIBUTORS.md).
2	# Licensed under the Apache License, Version 2.0 (see LICENSE).
3
4	from __future__ import annotations	7✔
5
6	import os	7✔
7	from dataclasses import dataclass	7✔
8	from pathlib import Path	7✔
9
10	from pants.base.glob_match_error_behavior import GlobMatchErrorBehavior	7✔
11	from pants.engine.collection import Collection	7✔
12	from pants.engine.fs import GlobExpansionConjunction, PathGlobs, PathMetadataRequest	7✔
13	from pants.engine.internals.native_engine import (	7✔
14	Digest,
15	PathMetadataKind,
16	PyNgOptions,
17	PyNgOptionsReader,
18	PyNgSourcePartition,
19	)
20	from pants.engine.internals.session import SessionValues	7✔
21	from pants.engine.intrinsics import path_globs_to_digest, path_globs_to_paths, path_metadata_request	7✔
22	from pants.engine.rules import Rule, _uncacheable_rule, collect_rules, implicitly, rule	7✔
23	from pants.source.source_root import SourceRoot, SourceRootsRequest, get_source_roots	7✔
24	from pants.util.memo import memoized_property	7✔
25
26
27	@dataclass(frozen=True)	7✔
28	class SourcePaths:	7✔
29	"""A set of sources, under a single source root."""
30
31	paths: tuple[Path, ...]	7✔
32	source_root: SourceRoot	7✔
33
34	def path_strs(self) -> tuple[str, ...]:	7✔
35	return tuple(str(path) for path in self.paths)	×
36
37	def filter_by_suffixes(self, suffixes: tuple[str, ...]) -> SourcePaths:	7✔
38	suffixes_set = set(suffixes)	1✔
39	return SourcePaths(	1✔
40	tuple(path for path in self.paths if path.suffix in suffixes_set),
41	self.source_root,
42	)
43
44
45	@dataclass(frozen=True)	7✔
46	class CommonDir:	7✔
47	path: Path \| None	7✔
48
49
50	@rule	7✔
51	async def find_common_dir(source_paths: SourcePaths) -> CommonDir:	7✔
52	if not source_paths.paths: # We don't expect empty SourcePaths, but might as well be robust.	1✔
53	return CommonDir(None)	×
54	commonpath = os.path.commonpath(source_paths.paths)	1✔
55	common_dir = commonpath	1✔
56	if len(source_paths.paths) == 1:	1✔
57	# If there is only one path, and it's a file, then commonpath will be that file, whereas
58	# we want its enclosing dir. So detect that case.
59	meta = await path_metadata_request(PathMetadataRequest(commonpath))	1✔
60	# Chase any symlinks back to the final path they point to.
61	while (	1✔
62	meta.metadata
63	and meta.metadata.kind == PathMetadataKind.SYMLINK
64	and meta.metadata.symlink_target
65	):
66	# NB: We don't `normpath` because eliminating `..` might change the meaning of the path
67	# if any of the intermediate directories are themselves symlinks.
68	symlink_target = os.path.join(os.path.dirname(commonpath), meta.metadata.symlink_target)	1✔
69	meta = await path_metadata_request(PathMetadataRequest(symlink_target))	1✔
70	if meta.metadata and meta.metadata.kind == PathMetadataKind.FILE:	1✔
71	common_dir = os.path.dirname(commonpath)	1✔
72	else:
73	common_dir = commonpath	1✔
74	return CommonDir(Path(common_dir))	1✔
75
76
77	@dataclass(frozen=True)	7✔
78	class SourceDigest:	7✔
79	digest: Digest	7✔
80
81
82	@rule	7✔
83	async def source_paths_to_digest(source_paths: SourcePaths) -> SourceDigest:	7✔
84	source_digest = await path_globs_to_digest(	×
85	PathGlobs(
86	source_paths.path_strs(),
87	glob_match_error_behavior=GlobMatchErrorBehavior.error,
88	conjunction=GlobExpansionConjunction.all_match,
89	description_of_origin="Input source paths",
90	)
91	)
92	return SourceDigest(source_digest)	×
93
94
95	@dataclass(frozen=True)	7✔
96	class SourcePartition:	7✔
97	"""Access to source files and the config that goes with them."""
98
99	_native_partition: PyNgSourcePartition	7✔
100	_source_root: SourceRoot	7✔
101
102	@memoized_property	7✔
103	def source_paths(self) -> SourcePaths:	7✔
104	return SourcePaths(	1✔
105	tuple(Path(p) for p in self._native_partition.paths()), self._source_root
106	)
107
108	@memoized_property	7✔
109	def options_reader(self) -> PyNgOptionsReader:	7✔
110	return self._native_partition.options_reader()	×
111
112
113	class SourcePartitions(Collection[SourcePartition]):	7✔
114	pass	7✔
115
116
117	# Uncacheable because we must get the most recent session value on each run.
118	@_uncacheable_rule	7✔
119	async def get_ng_options(session_values: SessionValues) -> PyNgOptions:	7✔
120	return session_values[PyNgOptions]	×
121
122
123	# Uncacheable because we must recompute on each run.
124	@_uncacheable_rule	7✔
125	async def partition_sources(path_globs: PathGlobs) -> SourcePartitions:	7✔
126	options = await get_ng_options(**implicitly())	1✔
127	paths = await path_globs_to_paths(path_globs)	1✔
128	# First partition by source root.
129	source_roots = await get_source_roots(SourceRootsRequest.for_files(paths.files))	1✔
130	root_to_paths = source_roots.root_to_paths()	1✔
131	partitions: list[SourcePartition] = []	1✔
132	for source_root, paths_in_partition in root_to_paths.items():	1✔
133	# Then subpartition each of those by config.
134	partitions.extend(	1✔
135	SourcePartition(native_part, source_root)
136	for native_part in options.partition_sources(
137	tuple(str(path) for path in paths_in_partition)
138	)
139	)
140	return SourcePartitions(tuple(partitions))	1✔
141
142
143	@rule	7✔
144	async def get_source_paths(partition: SourcePartition) -> SourcePaths:	7✔
145	return partition.source_paths	×
146
147
148	def rules() -> tuple[Rule, ...]:	7✔
149	return (*collect_rules(),)	7✔

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