26342152999

Committed 23 May 2026 07:59PM UTC coverage: 91.165% (-1.6%) from 92.792%

Build # 26342152999

Build Type

push

github

Committed by

web-flow

Commit Message

Run Linux ARM CI on Depot runners (#23363)

RunsOn is deprecating their v2 stack, and rather than migrate
to v3 we should use the resources kindly donated by Depot.

GitHub also now has Linux ARM runners, should we need them.

Coverage Stats

87305 of 95766 relevant lines covered (91.16%)

3.87 hits per line

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

89.16

/src/python/pants/backend/helm/util_rules/sources.py

# Copyright 2022 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 pants.backend.helm.target_types import (
    HelmChartFieldSet,
    HelmChartMetaSourceField,
    HelmChartSourcesField,
)
from pants.core.target_types import FileSourceField, ResourceSourceField
from pants.core.util_rules import source_files
from pants.core.util_rules.source_files import (
    SourceFiles,
    SourceFilesRequest,
    determine_source_files,
)
from pants.engine.engine_aware import EngineAwareParameter
from pants.engine.fs import DigestSubset, MergeDigests, PathGlobs, Snapshot
from pants.engine.internals.graph import hydrate_sources, resolve_targets
from pants.engine.internals.native_engine import RemovePrefix
from pants.engine.intrinsics import digest_subset_to_digest, digest_to_snapshot
from pants.engine.rules import collect_rules, concurrently, implicitly, rule
from pants.engine.target import DependenciesRequest, HydrateSourcesRequest, SourcesField, Target


@dataclass(frozen=True)
class HelmChartRootRequest(EngineAwareParameter):
    source: HelmChartMetaSourceField

    def debug_hint(self) -> str | None:
        return self.source.address.spec


@dataclass(frozen=True)
class HelmChartRoot:
    path: str


@rule(desc="Detect Helm chart source root")
async def find_chart_source_root(request: HelmChartRootRequest) -> HelmChartRoot:
    source = await hydrate_sources(
        HydrateSourcesRequest(
            request.source, for_sources_types=[HelmChartMetaSourceField], enable_codegen=True
        ),
        **implicitly(),
    )
    assert len(source.snapshot.files) == 1

    return HelmChartRoot(os.path.dirname(source.snapshot.files[0]))


@dataclass(frozen=True)
class HelmChartSourceFilesRequest(EngineAwareParameter):
    field_set: HelmChartFieldSet
    include_resources: bool
    include_files: bool
    include_metadata: bool

    @classmethod
    def create(
        cls,
        target: Target,
        *,
        include_resources: bool = True,
        include_files: bool = False,
        include_metadata: bool = True,
    ) -> HelmChartSourceFilesRequest:
        return cls.for_field_set(
            HelmChartFieldSet.create(target),
            include_resources=include_resources,
            include_files=include_files,
            include_metadata=include_metadata,
        )

    @classmethod
    def for_field_set(
        cls,
        field_set: HelmChartFieldSet,
        *,
        include_resources: bool = True,
        include_files: bool = False,
        include_metadata: bool = True,
    ) -> HelmChartSourceFilesRequest:
        return cls(
            field_set=field_set,
            include_resources=include_resources,
            include_files=include_files,
            include_metadata=include_metadata,
        )

    @property
    def sources_fields(self) -> tuple[SourcesField, ...]:
        fields: list[SourcesField] = [self.field_set.sources]
        if self.include_metadata:
            fields.append(self.field_set.chart)
        return tuple(fields)

    @property
    def valid_sources_types(self) -> tuple[type[SourcesField], ...]:
        types: list[type[SourcesField]] = [HelmChartSourcesField]
        if self.include_metadata:
            types.append(HelmChartMetaSourceField)
        if self.include_resources:
            types.append(ResourceSourceField)
        if self.include_files:
            types.append(FileSourceField)
        return tuple(types)

    def debug_hint(self) -> str | None:
        return self.field_set.address.spec


@dataclass(frozen=True)
class HelmChartSourceFiles:
    snapshot: Snapshot
    unrooted_files: tuple[str, ...]


async def _strip_chart_source_root(
    source_files: SourceFiles, chart_root: HelmChartRoot
) -> Snapshot:
    if not source_files.snapshot.files:
        return source_files.snapshot

    if source_files.unrooted_files:
        rooted_files = set(source_files.snapshot.files) - set(source_files.unrooted_files)
        rooted_files_snapshot = await digest_to_snapshot(
            **implicitly(DigestSubset(source_files.snapshot.digest, PathGlobs(rooted_files)))
        )
    else:
        rooted_files_snapshot = source_files.snapshot

    resulting_snapshot = await digest_to_snapshot(
        **implicitly(RemovePrefix(rooted_files_snapshot.digest, chart_root.path))
    )
    if source_files.unrooted_files:
        # Add unrooted files back in
        unrooted_digest = await digest_subset_to_digest(
            DigestSubset(source_files.snapshot.digest, PathGlobs(source_files.unrooted_files)),
        )
        resulting_snapshot = await digest_to_snapshot(
            **implicitly(MergeDigests([resulting_snapshot.digest, unrooted_digest]))
        )

    return resulting_snapshot


@rule
async def get_helm_source_files(request: HelmChartSourceFilesRequest) -> HelmChartSourceFiles:
    chart_root, dependencies = await concurrently(
        find_chart_source_root(HelmChartRootRequest(request.field_set.chart)),
        resolve_targets(**implicitly(DependenciesRequest(request.field_set.dependencies))),
    )

    source_files, original_sources = await concurrently(
        determine_source_files(
            SourceFilesRequest(
                sources_fields=[
                    *request.sources_fields,
                    *(
                        tgt.get(SourcesField)
                        for tgt in dependencies
                        if not HelmChartFieldSet.is_applicable(tgt)
                    ),
                ],
                for_sources_types=request.valid_sources_types,
                enable_codegen=True,
            )
        ),
        determine_source_files(
            SourceFilesRequest([request.field_set.sources], enable_codegen=False)
        ),
    )

    stripped_source_files = await _strip_chart_source_root(source_files, chart_root)
    stripped_original_sources = await _strip_chart_source_root(original_sources, chart_root)

    all_files_snapshot = await digest_to_snapshot(
        **implicitly(MergeDigests([stripped_source_files.digest, stripped_original_sources.digest]))
    )
    return HelmChartSourceFiles(
        snapshot=all_files_snapshot,
        unrooted_files=(*source_files.unrooted_files, *original_sources.unrooted_files),
    )


def rules():
    return [*collect_rules(), *source_files.rules()]

1	# Copyright 2022 Pants project contributors (see CONTRIBUTORS.md).
2	# Licensed under the Apache License, Version 2.0 (see LICENSE).
3
4	from __future__ import annotations	9✔
5
6	import os	9✔
7	from dataclasses import dataclass	9✔
8
9	from pants.backend.helm.target_types import (	9✔
10	HelmChartFieldSet,
11	HelmChartMetaSourceField,
12	HelmChartSourcesField,
13	)
14	from pants.core.target_types import FileSourceField, ResourceSourceField	9✔
15	from pants.core.util_rules import source_files	9✔
16	from pants.core.util_rules.source_files import (	9✔
17	SourceFiles,
18	SourceFilesRequest,
19	determine_source_files,
20	)
21	from pants.engine.engine_aware import EngineAwareParameter	9✔
22	from pants.engine.fs import DigestSubset, MergeDigests, PathGlobs, Snapshot	9✔
23	from pants.engine.internals.graph import hydrate_sources, resolve_targets	9✔
24	from pants.engine.internals.native_engine import RemovePrefix	9✔
25	from pants.engine.intrinsics import digest_subset_to_digest, digest_to_snapshot	9✔
26	from pants.engine.rules import collect_rules, concurrently, implicitly, rule	9✔
27	from pants.engine.target import DependenciesRequest, HydrateSourcesRequest, SourcesField, Target	9✔
28
29
30	@dataclass(frozen=True)	9✔
31	class HelmChartRootRequest(EngineAwareParameter):	9✔
32	source: HelmChartMetaSourceField	9✔
33
34	def debug_hint(self) -> str \| None:	9✔
35	return self.source.address.spec	×
36
37
38	@dataclass(frozen=True)	9✔
39	class HelmChartRoot:	9✔
40	path: str	9✔
41
42
43	@rule(desc="Detect Helm chart source root")	9✔
44	async def find_chart_source_root(request: HelmChartRootRequest) -> HelmChartRoot:	9✔
45	source = await hydrate_sources(	7✔
46	HydrateSourcesRequest(
47	request.source, for_sources_types=[HelmChartMetaSourceField], enable_codegen=True
48	),
49	**implicitly(),
50	)
51	assert len(source.snapshot.files) == 1	7✔
52
53	return HelmChartRoot(os.path.dirname(source.snapshot.files[0]))	7✔
54
55
56	@dataclass(frozen=True)	9✔
57	class HelmChartSourceFilesRequest(EngineAwareParameter):	9✔
58	field_set: HelmChartFieldSet	9✔
59	include_resources: bool	9✔
60	include_files: bool	9✔
61	include_metadata: bool	9✔
62
63	@classmethod	9✔
64	def create(	9✔
65	cls,
66	target: Target,
67	*,
68	include_resources: bool = True,
69	include_files: bool = False,
70	include_metadata: bool = True,
71	) -> HelmChartSourceFilesRequest:
72	return cls.for_field_set(	×
73	HelmChartFieldSet.create(target),
74	include_resources=include_resources,
75	include_files=include_files,
76	include_metadata=include_metadata,
77	)
78
79	@classmethod	9✔
80	def for_field_set(	9✔
81	cls,
82	field_set: HelmChartFieldSet,
83	*,
84	include_resources: bool = True,
85	include_files: bool = False,
86	include_metadata: bool = True,
87	) -> HelmChartSourceFilesRequest:
88	return cls(	7✔
89	field_set=field_set,
90	include_resources=include_resources,
91	include_files=include_files,
92	include_metadata=include_metadata,
93	)
94
95	@property	9✔
96	def sources_fields(self) -> tuple[SourcesField, ...]:	9✔
97	fields: list[SourcesField] = [self.field_set.sources]	7✔
98	if self.include_metadata:	7✔
99	fields.append(self.field_set.chart)	×
100	return tuple(fields)	7✔
101
102	@property	9✔
103	def valid_sources_types(self) -> tuple[type[SourcesField], ...]:	9✔
104	types: list[type[SourcesField]] = [HelmChartSourcesField]	7✔
105	if self.include_metadata:	7✔
106	types.append(HelmChartMetaSourceField)	×
107	if self.include_resources:	7✔
108	types.append(ResourceSourceField)	7✔
109	if self.include_files:	7✔
110	types.append(FileSourceField)	7✔
111	return tuple(types)	7✔
112
113	def debug_hint(self) -> str \| None:	9✔
114	return self.field_set.address.spec	×
115
116
117	@dataclass(frozen=True)	9✔
118	class HelmChartSourceFiles:	9✔
119	snapshot: Snapshot	9✔
120	unrooted_files: tuple[str, ...]	9✔
121
122
123	async def _strip_chart_source_root(	9✔
124	source_files: SourceFiles, chart_root: HelmChartRoot
125	) -> Snapshot:
126	if not source_files.snapshot.files:	7✔
127	return source_files.snapshot	1✔
128
129	if source_files.unrooted_files:	7✔
130	rooted_files = set(source_files.snapshot.files) - set(source_files.unrooted_files)	×
131	rooted_files_snapshot = await digest_to_snapshot(	×
132	**implicitly(DigestSubset(source_files.snapshot.digest, PathGlobs(rooted_files)))
133	)
134	else:
135	rooted_files_snapshot = source_files.snapshot	7✔
136
137	resulting_snapshot = await digest_to_snapshot(	7✔
138	**implicitly(RemovePrefix(rooted_files_snapshot.digest, chart_root.path))
139	)
140	if source_files.unrooted_files:	7✔
141	# Add unrooted files back in
142	unrooted_digest = await digest_subset_to_digest(	×
143	DigestSubset(source_files.snapshot.digest, PathGlobs(source_files.unrooted_files)),
144	)
145	resulting_snapshot = await digest_to_snapshot(	×
146	**implicitly(MergeDigests([resulting_snapshot.digest, unrooted_digest]))
147	)
148
149	return resulting_snapshot	7✔
150
151
152	@rule	9✔
153	async def get_helm_source_files(request: HelmChartSourceFilesRequest) -> HelmChartSourceFiles:	9✔
154	chart_root, dependencies = await concurrently(	7✔
155	find_chart_source_root(HelmChartRootRequest(request.field_set.chart)),
156	resolve_targets(**implicitly(DependenciesRequest(request.field_set.dependencies))),
157	)
158
159	source_files, original_sources = await concurrently(	7✔
160	determine_source_files(
161	SourceFilesRequest(
162	sources_fields=[
163	*request.sources_fields,
164	*(
165	tgt.get(SourcesField)
166	for tgt in dependencies
167	if not HelmChartFieldSet.is_applicable(tgt)
168	),
169	],
170	for_sources_types=request.valid_sources_types,
171	enable_codegen=True,
172	)
173	),
174	determine_source_files(
175	SourceFilesRequest([request.field_set.sources], enable_codegen=False)
176	),
177	)
178
179	stripped_source_files = await _strip_chart_source_root(source_files, chart_root)	7✔
180	stripped_original_sources = await _strip_chart_source_root(original_sources, chart_root)	7✔
181
182	all_files_snapshot = await digest_to_snapshot(	7✔
183	**implicitly(MergeDigests([stripped_source_files.digest, stripped_original_sources.digest]))
184	)
185	return HelmChartSourceFiles(	7✔
186	snapshot=all_files_snapshot,
187	unrooted_files=(source_files.unrooted_files, original_sources.unrooted_files),
188	)
189
190
191	def rules():	9✔
192	return [collect_rules(), source_files.rules()]	7✔

pantsbuild / pants / 26342152999

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