20632486505

Committed 01 Jan 2026 04:21AM UTC coverage: 43.231% (-37.1%) from 80.281%

Build # 20632486505

Build Type

Pull #22962

github

Committed by

web-flow

Commit Message

Merge 08d5c63b0 into f52ab6675

Pull Request Pull Request #22962: Bump the gha-deps group across 1 directory with 6 updates

Run Details

26122 of 60424 relevant lines covered (43.23%)

0.86 hits per line

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

37.8

/src/python/pants/backend/nfpm/util_rules/sandbox.py

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

from __future__ import annotations

import dataclasses
from dataclasses import dataclass
from enum import Enum, auto
from typing import cast

from pants.backend.nfpm.field_sets import NFPM_PACKAGE_FIELD_SET_TYPES, NfpmPackageFieldSet
from pants.backend.nfpm.fields.contents import (
    NfpmContentDirDstField,
    NfpmContentDstField,
    NfpmContentFileSourceField,
    NfpmContentSrcField,
    NfpmContentSymlinkDstField,
)
from pants.backend.nfpm.target_types import NfpmContentFile, NfpmPackageTarget
from pants.core.goals.package import (
    EnvironmentAwarePackageRequest,
    PackageFieldSet,
    TraverseIfNotPackageTarget,
    environment_aware_package,
)
from pants.core.util_rules.source_files import SourceFilesRequest, determine_source_files
from pants.core.util_rules.source_files import rules as source_files_rules
from pants.engine.fs import CreateDigest, DigestEntries, Directory, FileEntry, SymlinkEntry
from pants.engine.internals.graph import find_valid_field_sets, hydrate_sources
from pants.engine.internals.graph import transitive_targets as transitive_targets_get
from pants.engine.internals.native_engine import Digest, MergeDigests
from pants.engine.internals.selectors import concurrently
from pants.engine.intrinsics import create_digest, get_digest_entries, merge_digests
from pants.engine.rules import collect_rules, implicitly, rule
from pants.engine.target import (
    FieldSetsPerTargetRequest,
    GenerateSourcesRequest,
    HydrateSourcesRequest,
    SourcesField,
    Target,
    TransitiveTargets,
    TransitiveTargetsRequest,
)
from pants.engine.unions import UnionMembership
from pants.util.ordered_set import FrozenOrderedSet


class _DepCategory(Enum):
    ignore = auto()
    nfpm_content_from_dependency = auto()
    nfpm_content_from_source = auto()
    nfpm_package = auto()
    remaining = auto()

    @classmethod
    def for_target(
        cls,
        tgt: Target,
        field_set_type: type[NfpmPackageFieldSet],
    ) -> _DepCategory:
        # Assumption: this gets called with atomic targets not target generators. For example,
        # TransitiveTargets gets calculated AFTER target generation/expansion.
        if tgt.has_field(NfpmContentDirDstField) or tgt.has_field(NfpmContentSymlinkDstField):
            # NfpmContentDir and NfpmContentSymlink targets don't go in the sandbox.
            # They're only registered in the nfpm config.
            return _DepCategory.ignore
        if tgt.has_field(NfpmContentDstField):
            # an NfpmContentFile DOES need something in the sandbox

            # 'source' must be either None or a non-empty string.
            # If 'source' is None, the file comes from dependencies.
            if tgt[NfpmContentFileSourceField].value is None:
                # The file must be hydrated from one of the dependencies.
                return _DepCategory.nfpm_content_from_dependency
            # The file must be hydrated from the 'source' field
            return _DepCategory.nfpm_content_from_source

        for pkg_field_set_type in NFPM_PACKAGE_FIELD_SET_TYPES:
            if pkg_field_set_type.is_applicable(tgt):
                # we only respect nfpm package deps for the same packager
                # (For example, deb targets will ignore any deps on rpm targets)
                if pkg_field_set_type == field_set_type:
                    return _DepCategory.nfpm_package
                return _DepCategory.ignore

        return _DepCategory.remaining


@dataclass(frozen=True)
class _NfpmSortedDeps:
    nfpm_content_from_dependency_targets: tuple[NfpmContentFile, ...]
    nfpm_content_from_source_targets: tuple[NfpmContentFile, ...]
    nfpm_package_targets: tuple[NfpmPackageTarget, ...]
    package_targets: tuple[Target, ...]
    remaining_targets: tuple[Target, ...]

    @classmethod
    def sort(
        cls,
        field_set: NfpmPackageFieldSet,
        transitive_targets: TransitiveTargets,
        union_membership: UnionMembership,
    ) -> _NfpmSortedDeps:
        package_field_set_types = (
            FrozenOrderedSet(union_membership.get(PackageFieldSet)) - NFPM_PACKAGE_FIELD_SET_TYPES
        )

        nfpm_content_from_dependency_targets: list[NfpmContentFile] = []
        nfpm_content_from_source_targets: list[NfpmContentFile] = []
        nfpm_package_targets: list[NfpmPackageTarget] = []
        package_targets: list[Target] = []
        remaining_targets: list[Target] = []

        for tgt in transitive_targets.dependencies:
            category = _DepCategory.for_target(tgt, type(field_set))
            if category == _DepCategory.ignore:
                continue
            elif category == _DepCategory.nfpm_content_from_dependency:
                nfpm_content_from_dependency_targets.append(cast(NfpmContentFile, tgt))
            elif category == _DepCategory.nfpm_content_from_source:
                nfpm_content_from_source_targets.append(cast(NfpmContentFile, tgt))
            elif category == _DepCategory.nfpm_package:
                nfpm_package_targets.append(cast(NfpmPackageTarget, tgt))
            elif category == _DepCategory.remaining:
                is_package = False
                for field_set_type in package_field_set_types:
                    if field_set_type.is_applicable(tgt):
                        is_package = True
                        package_targets.append(tgt)
                        break
                if not is_package:
                    remaining_targets.append(tgt)
            else:
                raise ValueError(f"Please file a bug report--got unknown _DepCategory: {category}")

        return cls(
            nfpm_content_from_dependency_targets=tuple(nfpm_content_from_dependency_targets),
            nfpm_content_from_source_targets=tuple(nfpm_content_from_source_targets),
            nfpm_package_targets=tuple(nfpm_package_targets),
            package_targets=tuple(package_targets),
            remaining_targets=tuple(remaining_targets),
        )


@dataclass(frozen=True)
class NfpmContentSandboxRequest:
    field_set: NfpmPackageFieldSet


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


@rule
async def populate_nfpm_content_sandbox(
    request: NfpmContentSandboxRequest, union_membership: UnionMembership
) -> NfpmContentSandbox:
    transitive_targets = await transitive_targets_get(
        TransitiveTargetsRequest(
            [request.field_set.address],
            should_traverse_deps_predicate=TraverseIfNotPackageTarget(
                roots=[request.field_set.address],
                union_membership=union_membership,
            ),
        ),
        **implicitly(),
    )

    deps = _NfpmSortedDeps.sort(request.field_set, transitive_targets, union_membership)

    # 1. Build packages for deps that are (non-nfpm) Packages

    package_field_sets_per_target = await find_valid_field_sets(
        FieldSetsPerTargetRequest(PackageFieldSet, deps.package_targets), **implicitly()
    )
    packages = await concurrently(
        environment_aware_package(EnvironmentAwarePackageRequest(field_set))
        for field_set in package_field_sets_per_target.field_sets
    )

    # 2. Hydrate 'source' fields for nfpm_content_file targets.

    nfpm_content_source_fields_to_relocate: list[
        tuple[NfpmContentFileSourceField, NfpmContentSrcField]
    ] = []
    nfpm_content_source_fields: list[SourcesField] = []

    for nfpm_content_tgt in deps.nfpm_content_from_source_targets:
        source = nfpm_content_tgt[NfpmContentFileSourceField]
        src = nfpm_content_tgt[NfpmContentSrcField]
        # If 'src' is empty, it defaults to the content target's 'source'.
        if src.value and source.value != src.value:
            nfpm_content_source_fields_to_relocate.append((source, src))
        else:
            nfpm_content_source_fields.append(source)

    hydrated_sources_to_relocate = await concurrently(
        hydrate_sources(HydrateSourcesRequest(field), **implicitly())
        for field, _ in nfpm_content_source_fields_to_relocate
    )
    relocated_source_entries = await concurrently(
        get_digest_entries(hydrated.snapshot.digest) for hydrated in hydrated_sources_to_relocate
    )
    moved_entries: list[FileEntry | SymlinkEntry | Directory] = []
    digest_entries: DigestEntries
    for digest_entries, (source, src) in zip(
        relocated_source_entries, nfpm_content_source_fields_to_relocate
    ):
        for entry in digest_entries:
            if isinstance(entry, FileEntry) and entry.path == source.value:
                new_path = src.value
                if new_path is None:
                    raise ValueError("unexpected None")

                moved_entries.append(dataclasses.replace(entry, path=new_path))
            else:
                moved_entries.append(entry)

    nfpm_content_relocated_sources_digest, nfpm_content_sources = await concurrently(
        create_digest(CreateDigest(moved_entries)),
        # nfpm_content_file sources are simply files -- no codegen required.
        # anything more involved (like downloading http_source()) should use 'dependencies' instead
        # (for example, depend on a 'file(source=http_source(...))' target to download something).
        determine_source_files(SourceFilesRequest(nfpm_content_source_fields)),
    )

    # 3. Hydrate sources from 'dependencies' fields for nfpm_content_file targets,
    # which should all be accounted for in the transitive targets in deps.remaining_targets.

    # This involves doing as much codegen as possible (based on export_codegen goal).
    codegen_inputs_to_outputs = [
        (req.input, req.output) for req in union_membership.get(GenerateSourcesRequest)
    ]
    codegen_sources_fields_with_output = []
    for tgt in deps.remaining_targets:
        if not tgt.has_field(SourcesField):
            continue
        sources_field = tgt[SourcesField]
        found = False
        for input_type, output_type in codegen_inputs_to_outputs:
            if isinstance(sources_field, input_type):
                codegen_sources_fields_with_output.append((sources_field, output_type))
                found = True
        # make sure to include anything where codegen doesn't apply
        if not found:
            codegen_sources_fields_with_output.append((sources_field, type(sources_field)))
    hydrated_dep_sources = await concurrently(
        hydrate_sources(
            HydrateSourcesRequest(
                sources,
                for_sources_types=(output_type,),
                enable_codegen=True,
            ),
            **implicitly(),
        )
        for sources, output_type in codegen_sources_fields_with_output
    )

    # I would love to cleanly support relocations to 'src' from 'dependencies' files.
    # But, I don't see any clean approaches to identify which package, generated file,
    # or workspace file needs to be relocated to 'src'.
    # TODO: handle relocations from 'dependencies' to 'src'
    #       deps.nfpm_content_from_dependency_targets

    # This should include at least all files in 'src' fields of nfpm_content_file targets.
    # Other dependency files aren't required since nFPM will ignore anything not configured.
    sandbox_digest = await merge_digests(
        MergeDigests(
            [
                *(package.digest for package in packages),
                # nfpm_content_file 'src' from 'source' field
                nfpm_content_relocated_sources_digest,
                nfpm_content_sources.snapshot.digest,
                # nfpm_content_file 'src' from 'dependencies' field
                *(hydrated.snapshot.digest for hydrated in hydrated_dep_sources),
            ]
        )
    )

    return NfpmContentSandbox(sandbox_digest)


def rules():
    return [
        *source_files_rules(),
        *collect_rules(),
    ]

1	# Copyright 2023 Pants project contributors (see CONTRIBUTORS.md).
2	# Licensed under the Apache License, Version 2.0 (see LICENSE).
3
4	from __future__ import annotations	2✔
5
6	import dataclasses	2✔
7	from dataclasses import dataclass	2✔
8	from enum import Enum, auto	2✔
9	from typing import cast	2✔
10
11	from pants.backend.nfpm.field_sets import NFPM_PACKAGE_FIELD_SET_TYPES, NfpmPackageFieldSet	2✔
12	from pants.backend.nfpm.fields.contents import (	2✔
13	NfpmContentDirDstField,
14	NfpmContentDstField,
15	NfpmContentFileSourceField,
16	NfpmContentSrcField,
17	NfpmContentSymlinkDstField,
18	)
19	from pants.backend.nfpm.target_types import NfpmContentFile, NfpmPackageTarget	2✔
20	from pants.core.goals.package import (	2✔
21	EnvironmentAwarePackageRequest,
22	PackageFieldSet,
23	TraverseIfNotPackageTarget,
24	environment_aware_package,
25	)
26	from pants.core.util_rules.source_files import SourceFilesRequest, determine_source_files	2✔
27	from pants.core.util_rules.source_files import rules as source_files_rules	2✔
28	from pants.engine.fs import CreateDigest, DigestEntries, Directory, FileEntry, SymlinkEntry	2✔
29	from pants.engine.internals.graph import find_valid_field_sets, hydrate_sources	2✔
30	from pants.engine.internals.graph import transitive_targets as transitive_targets_get	2✔
31	from pants.engine.internals.native_engine import Digest, MergeDigests	2✔
32	from pants.engine.internals.selectors import concurrently	2✔
33	from pants.engine.intrinsics import create_digest, get_digest_entries, merge_digests	2✔
34	from pants.engine.rules import collect_rules, implicitly, rule	2✔
35	from pants.engine.target import (	2✔
36	FieldSetsPerTargetRequest,
37	GenerateSourcesRequest,
38	HydrateSourcesRequest,
39	SourcesField,
40	Target,
41	TransitiveTargets,
42	TransitiveTargetsRequest,
43	)
44	from pants.engine.unions import UnionMembership	2✔
45	from pants.util.ordered_set import FrozenOrderedSet	2✔
46
47
48	class _DepCategory(Enum):	2✔
49	ignore = auto()	2✔
50	nfpm_content_from_dependency = auto()	2✔
51	nfpm_content_from_source = auto()	2✔
52	nfpm_package = auto()	2✔
53	remaining = auto()	2✔
54
55	@classmethod	2✔
56	def for_target(	2✔
57	cls,
58	tgt: Target,
59	field_set_type: type[NfpmPackageFieldSet],
60	) -> _DepCategory:
61	# Assumption: this gets called with atomic targets not target generators. For example,
62	# TransitiveTargets gets calculated AFTER target generation/expansion.
63	if tgt.has_field(NfpmContentDirDstField) or tgt.has_field(NfpmContentSymlinkDstField):	×
64	# NfpmContentDir and NfpmContentSymlink targets don't go in the sandbox.
65	# They're only registered in the nfpm config.
66	return _DepCategory.ignore	×
67	if tgt.has_field(NfpmContentDstField):	×
68	# an NfpmContentFile DOES need something in the sandbox
69
70	# 'source' must be either None or a non-empty string.
71	# If 'source' is None, the file comes from dependencies.
72	if tgt[NfpmContentFileSourceField].value is None:	×
73	# The file must be hydrated from one of the dependencies.
74	return _DepCategory.nfpm_content_from_dependency	×
75	# The file must be hydrated from the 'source' field
76	return _DepCategory.nfpm_content_from_source	×
77
78	for pkg_field_set_type in NFPM_PACKAGE_FIELD_SET_TYPES:	×
79	if pkg_field_set_type.is_applicable(tgt):	×
80	# we only respect nfpm package deps for the same packager
81	# (For example, deb targets will ignore any deps on rpm targets)
82	if pkg_field_set_type == field_set_type:	×
83	return _DepCategory.nfpm_package	×
84	return _DepCategory.ignore	×
85
86	return _DepCategory.remaining	×
87
88
89	@dataclass(frozen=True)	2✔
90	class _NfpmSortedDeps:	2✔
91	nfpm_content_from_dependency_targets: tuple[NfpmContentFile, ...]	2✔
92	nfpm_content_from_source_targets: tuple[NfpmContentFile, ...]	2✔
93	nfpm_package_targets: tuple[NfpmPackageTarget, ...]	2✔
94	package_targets: tuple[Target, ...]	2✔
95	remaining_targets: tuple[Target, ...]	2✔
96
97	@classmethod	2✔
98	def sort(	2✔
99	cls,
100	field_set: NfpmPackageFieldSet,
101	transitive_targets: TransitiveTargets,
102	union_membership: UnionMembership,
103	) -> _NfpmSortedDeps:
104	package_field_set_types = (	×
105	FrozenOrderedSet(union_membership.get(PackageFieldSet)) - NFPM_PACKAGE_FIELD_SET_TYPES
106	)
107
108	nfpm_content_from_dependency_targets: list[NfpmContentFile] = []	×
109	nfpm_content_from_source_targets: list[NfpmContentFile] = []	×
110	nfpm_package_targets: list[NfpmPackageTarget] = []	×
111	package_targets: list[Target] = []	×
112	remaining_targets: list[Target] = []	×
113
114	for tgt in transitive_targets.dependencies:	×
115	category = _DepCategory.for_target(tgt, type(field_set))	×
116	if category == _DepCategory.ignore:	×
117	continue	×
118	elif category == _DepCategory.nfpm_content_from_dependency:	×
119	nfpm_content_from_dependency_targets.append(cast(NfpmContentFile, tgt))	×
120	elif category == _DepCategory.nfpm_content_from_source:	×
121	nfpm_content_from_source_targets.append(cast(NfpmContentFile, tgt))	×
122	elif category == _DepCategory.nfpm_package:	×
123	nfpm_package_targets.append(cast(NfpmPackageTarget, tgt))	×
124	elif category == _DepCategory.remaining:	×
125	is_package = False	×
126	for field_set_type in package_field_set_types:	×
127	if field_set_type.is_applicable(tgt):	×
128	is_package = True	×
129	package_targets.append(tgt)	×
130	break	×
131	if not is_package:	×
132	remaining_targets.append(tgt)	×
133	else:
134	raise ValueError(f"Please file a bug report--got unknown _DepCategory: {category}")	×
135
136	return cls(	×
137	nfpm_content_from_dependency_targets=tuple(nfpm_content_from_dependency_targets),
138	nfpm_content_from_source_targets=tuple(nfpm_content_from_source_targets),
139	nfpm_package_targets=tuple(nfpm_package_targets),
140	package_targets=tuple(package_targets),
141	remaining_targets=tuple(remaining_targets),
142	)
143
144
145	@dataclass(frozen=True)	2✔
146	class NfpmContentSandboxRequest:	2✔
147	field_set: NfpmPackageFieldSet	2✔
148
149
150	@dataclass(frozen=True)	2✔
151	class NfpmContentSandbox:	2✔
152	digest: Digest	2✔
153
154
155	@rule	2✔
156	async def populate_nfpm_content_sandbox(	2✔
157	request: NfpmContentSandboxRequest, union_membership: UnionMembership
158	) -> NfpmContentSandbox:
159	transitive_targets = await transitive_targets_get(	×
160	TransitiveTargetsRequest(
161	[request.field_set.address],
162	should_traverse_deps_predicate=TraverseIfNotPackageTarget(
163	roots=[request.field_set.address],
164	union_membership=union_membership,
165	),
166	),
167	**implicitly(),
168	)
169
170	deps = _NfpmSortedDeps.sort(request.field_set, transitive_targets, union_membership)	×
171
172	# 1. Build packages for deps that are (non-nfpm) Packages
173
174	package_field_sets_per_target = await find_valid_field_sets(	×
175	FieldSetsPerTargetRequest(PackageFieldSet, deps.package_targets), **implicitly()
176	)
177	packages = await concurrently(	×
178	environment_aware_package(EnvironmentAwarePackageRequest(field_set))
179	for field_set in package_field_sets_per_target.field_sets
180	)
181
182	# 2. Hydrate 'source' fields for nfpm_content_file targets.
183
184	nfpm_content_source_fields_to_relocate: list[	×
185	tuple[NfpmContentFileSourceField, NfpmContentSrcField]
186	] = []
187	nfpm_content_source_fields: list[SourcesField] = []	×
188
189	for nfpm_content_tgt in deps.nfpm_content_from_source_targets:	×
190	source = nfpm_content_tgt[NfpmContentFileSourceField]	×
191	src = nfpm_content_tgt[NfpmContentSrcField]	×
192	# If 'src' is empty, it defaults to the content target's 'source'.
193	if src.value and source.value != src.value:	×
194	nfpm_content_source_fields_to_relocate.append((source, src))	×
195	else:
196	nfpm_content_source_fields.append(source)	×
197
198	hydrated_sources_to_relocate = await concurrently(	×
199	hydrate_sources(HydrateSourcesRequest(field), **implicitly())
200	for field, _ in nfpm_content_source_fields_to_relocate
201	)
202	relocated_source_entries = await concurrently(	×
203	get_digest_entries(hydrated.snapshot.digest) for hydrated in hydrated_sources_to_relocate
204	)
205	moved_entries: list[FileEntry \| SymlinkEntry \| Directory] = []	×
206	digest_entries: DigestEntries
207	for digest_entries, (source, src) in zip(	×
208	relocated_source_entries, nfpm_content_source_fields_to_relocate
209	):
210	for entry in digest_entries:	×
211	if isinstance(entry, FileEntry) and entry.path == source.value:	×
212	new_path = src.value	×
213	if new_path is None:	×
214	raise ValueError("unexpected None")	×
215
216	moved_entries.append(dataclasses.replace(entry, path=new_path))	×
217	else:
218	moved_entries.append(entry)	×
219
220	nfpm_content_relocated_sources_digest, nfpm_content_sources = await concurrently(	×
221	create_digest(CreateDigest(moved_entries)),
222	# nfpm_content_file sources are simply files -- no codegen required.
223	# anything more involved (like downloading http_source()) should use 'dependencies' instead
224	# (for example, depend on a 'file(source=http_source(...))' target to download something).
225	determine_source_files(SourceFilesRequest(nfpm_content_source_fields)),
226	)
227
228	# 3. Hydrate sources from 'dependencies' fields for nfpm_content_file targets,
229	# which should all be accounted for in the transitive targets in deps.remaining_targets.
230
231	# This involves doing as much codegen as possible (based on export_codegen goal).
232	codegen_inputs_to_outputs = [	×
233	(req.input, req.output) for req in union_membership.get(GenerateSourcesRequest)
234	]
235	codegen_sources_fields_with_output = []	×
236	for tgt in deps.remaining_targets:	×
237	if not tgt.has_field(SourcesField):	×
238	continue	×
239	sources_field = tgt[SourcesField]	×
240	found = False	×
241	for input_type, output_type in codegen_inputs_to_outputs:	×
242	if isinstance(sources_field, input_type):	×
243	codegen_sources_fields_with_output.append((sources_field, output_type))	×
244	found = True	×
245	# make sure to include anything where codegen doesn't apply
246	if not found:	×
247	codegen_sources_fields_with_output.append((sources_field, type(sources_field)))	×
248	hydrated_dep_sources = await concurrently(	×
249	hydrate_sources(
250	HydrateSourcesRequest(
251	sources,
252	for_sources_types=(output_type,),
253	enable_codegen=True,
254	),
255	**implicitly(),
256	)
257	for sources, output_type in codegen_sources_fields_with_output
258	)
259
260	# I would love to cleanly support relocations to 'src' from 'dependencies' files.
261	# But, I don't see any clean approaches to identify which package, generated file,
262	# or workspace file needs to be relocated to 'src'.
263	# TODO: handle relocations from 'dependencies' to 'src'
264	# deps.nfpm_content_from_dependency_targets
265
266	# This should include at least all files in 'src' fields of nfpm_content_file targets.
267	# Other dependency files aren't required since nFPM will ignore anything not configured.
268	sandbox_digest = await merge_digests(	×
269	MergeDigests(
270	[
271	*(package.digest for package in packages),
272	# nfpm_content_file 'src' from 'source' field
273	nfpm_content_relocated_sources_digest,
274	nfpm_content_sources.snapshot.digest,
275	# nfpm_content_file 'src' from 'dependencies' field
276	*(hydrated.snapshot.digest for hydrated in hydrated_dep_sources),
277	]
278	)
279	)
280
281	return NfpmContentSandbox(sandbox_digest)	×
282
283
284	def rules():	2✔
285	return [	2✔
286	*source_files_rules(),
287	*collect_rules(),
288	]

pantsbuild / pants / 20632486505

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