22285099215

Committed 22 Feb 2026 08:52PM UTC coverage: 75.854% (-17.1%) from 92.936%

Build # 22285099215

Build Type

Pull #23121

github

Committed by

web-flow

Commit Message

Merge c7299df9c into ba8359840

Pull Request Pull Request #23121: fix issue with optional fields in dependency validator

Run Details

28 of 29 new or added lines in 2 files covered. (96.55%)

11174 existing lines in 400 files now uncovered.

53694 of 70786 relevant lines covered (75.85%)

1.88 hits per line

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

72.18

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

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

from __future__ import annotations

import dataclasses
import logging
from collections.abc import Iterable
from dataclasses import dataclass
from typing import Any

from pants.backend.helm.dependency_inference import chart as chart_inference
from pants.backend.helm.resolve import fetch
from pants.backend.helm.resolve.artifacts import ResolvedHelmArtifact
from pants.backend.helm.resolve.fetch import FetchedHelmArtifact, FetchHelmArtifactRequest
from pants.backend.helm.subsystems.helm import HelmSubsystem
from pants.backend.helm.target_types import (
    HelmArtifactFieldSet,
    HelmChartFieldSet,
    HelmChartTarget,
    HelmDeploymentFieldSet,
)
from pants.backend.helm.util_rules import chart_metadata, sources
from pants.backend.helm.util_rules.chart_metadata import (
    HELM_CHART_METADATA_FILENAMES,
    HelmChartDependency,
    HelmChartMetadata,
    ParseHelmChartMetadataDigest,
    parse_chart_metadata_from_digest,
    parse_chart_metadata_from_field,
    render_chart_metadata,
)
from pants.backend.helm.util_rules.sources import HelmChartSourceFilesRequest, get_helm_source_files
from pants.engine.addresses import Address
from pants.engine.engine_aware import EngineAwareParameter, EngineAwareReturnType
from pants.engine.fs import (
    EMPTY_DIGEST,
    AddPrefix,
    Digest,
    DigestSubset,
    FileDigest,
    MergeDigests,
    PathGlobs,
    Snapshot,
)
from pants.engine.internals.build_files import resolve_address
from pants.engine.internals.graph import resolve_target, resolve_targets
from pants.engine.intrinsics import (
    add_prefix,
    digest_subset_to_digest,
    digest_to_snapshot,
    merge_digests,
)
from pants.engine.rules import collect_rules, concurrently, implicitly, rule
from pants.engine.target import DependenciesRequest, Target, WrappedTargetRequest
from pants.util.frozendict import FrozenDict
from pants.util.logging import LogLevel
from pants.util.ordered_set import OrderedSet
from pants.util.strutil import pluralize, softwrap

logger = logging.getLogger(__name__)


class InvalidHelmChartTarget(ValueError):
    def __init__(self, target: Target) -> None:
        super().__init__(f"The target {target.address} is not a `{HelmChartTarget.alias}`.")


@dataclass(frozen=True)
class HelmChart(EngineAwareReturnType):
    address: Address
    info: HelmChartMetadata
    snapshot: Snapshot
    artifact: ResolvedHelmArtifact | None = None

    @property
    def name(self) -> str:
        return self.info.name

    @property
    def description(self) -> str:
        return self.name

    @property
    def immutable_input_digests(self) -> FrozenDict[str, Digest]:
        return FrozenDict({self.name: self.snapshot.digest})

    def level(self) -> LogLevel | None:
        return LogLevel.DEBUG

    def message(self) -> str | None:
        msg = f"Built Helm chart '{self.info.name}' from target at {self.address}"
        if self.artifact:
            msg += f" (resolved using URL {self.artifact.chart_url})."
        return msg

    def metadata(self) -> dict[str, Any] | None:
        meta: dict[str, Any] = {"name": self.info.name, "address": self.address.spec}
        if self.artifact:
            meta["artifact"] = self.artifact
        return meta

    def artifacts(self) -> dict[str, FileDigest | Snapshot] | None:
        return {"snapshot": self.snapshot}


@dataclass(frozen=True)
class HelmChartRequest(EngineAwareParameter):
    field_set: HelmChartFieldSet

    @classmethod
    def from_target(cls, target: Target) -> HelmChartRequest:
        if not HelmChartFieldSet.is_applicable(target):
            raise InvalidHelmChartTarget(target)
        return cls(HelmChartFieldSet.create(target))

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

    def metadata(self) -> dict[str, Any] | None:
        return {"address": self.field_set.address.spec}


@rule(desc="Compile third parth Helm chart", level=LogLevel.DEBUG)
async def create_chart_from_artifact(fetched_artifact: FetchedHelmArtifact) -> HelmChart:
    metadata = await parse_chart_metadata_from_digest(
        ParseHelmChartMetadataDigest(
            fetched_artifact.snapshot.digest,
            description_of_origin=f"the `helm_artifact` {fetched_artifact.address}",
        )
    )
    return HelmChart(
        fetched_artifact.address,
        metadata,
        fetched_artifact.snapshot,
        artifact=fetched_artifact.artifact,
    )


async def _merge_subchart_digests(charts: Iterable[HelmChart]) -> Digest:
    prefixed_chart_digests = await concurrently(
        add_prefix(AddPrefix(chart.snapshot.digest, chart.name)) for chart in charts
    )
    merged_digests = await merge_digests(MergeDigests(prefixed_chart_digests))
    return await add_prefix(AddPrefix(merged_digests, "charts"))


async def _find_charts_by_targets(
    targets: Iterable[Target], *, description_of_origin: str
) -> tuple[HelmChart, ...]:
    # These `Get`s are not ported yet to call-by-name due to the rule cycle with`get_helm_chart`
    # which calls this function.
    requests = [
        *(
            get_helm_chart(HelmChartRequest.from_target(target), **implicitly())
            for target in targets
            if HelmChartFieldSet.is_applicable(target)
        ),
        *(
            create_chart_from_artifact(
                **implicitly(
                    {
                        FetchHelmArtifactRequest.from_target(
                            target,
                            description_of_origin=description_of_origin,
                        ): FetchHelmArtifactRequest
                    }
                )
            )
            for target in targets
            if HelmArtifactFieldSet.is_applicable(target)
        ),
    ]
    return await concurrently(requests)


@rule(desc="Compile Helm chart", level=LogLevel.DEBUG)
async def get_helm_chart(request: HelmChartRequest, subsystem: HelmSubsystem) -> HelmChart:
    dependencies, source_files, chart_info = await concurrently(
        resolve_targets(**implicitly(DependenciesRequest(request.field_set.dependencies))),
        get_helm_source_files(
            HelmChartSourceFilesRequest.for_field_set(
                request.field_set,
                include_metadata=False,
                include_resources=True,
                include_files=True,
            )
        ),
        parse_chart_metadata_from_field(request.field_set.chart),
    )

    if request.field_set.version.value:
        chart_info = dataclasses.replace(chart_info, version=request.field_set.version.value)

    subcharts_digest = EMPTY_DIGEST
    subcharts = await _find_charts_by_targets(
        dependencies, description_of_origin=f"the `helm_chart` {request.field_set.address}"
    )
    if subcharts:
        logger.debug(
            softwrap(
                f"""
                Found {pluralize(len(subcharts), "subchart")} as direct dependencies
                on Helm chart at: {request.field_set.address}.
                """
            )
        )

        subcharts_digest = await _merge_subchart_digests(subcharts)

        # Update subchart dependencies in the metadata and re-render it.
        remotes = subsystem.remotes()
        subchart_map: dict[str, HelmChart] = {chart.info.name: chart for chart in subcharts}
        updated_dependencies: OrderedSet[HelmChartDependency] = OrderedSet()
        for dep in chart_info.dependencies:
            updated_dep = dep

            if not dep.repository and remotes.default_registry:
                # If the dependency hasn't specified a repository, then we choose the registry with the 'default' alias.
                default_remote = remotes.default_registry
                updated_dep = dataclasses.replace(updated_dep, repository=default_remote.address)
            elif dep.repository and dep.repository.startswith("@"):
                remote = next(remotes.get(dep.repository))
                updated_dep = dataclasses.replace(updated_dep, repository=remote.address)

            if dep.name in subchart_map:
                updated_dep = dataclasses.replace(
                    updated_dep, version=subchart_map[dep.name].info.version
                )

            updated_dependencies.add(updated_dep)

        # Include the explicitly provided subchats in the set of dependencies if not already present.
        updated_dependencies_names = {dep.name for dep in updated_dependencies}
        remaining_subcharts = [
            chart for chart in subcharts if chart.info.name not in updated_dependencies_names
        ]
        for chart in remaining_subcharts:
            if chart.artifact:
                dependency = HelmChartDependency(
                    name=chart.artifact.name,
                    version=chart.artifact.version,
                    repository=chart.artifact.location_url,
                )
            else:
                dependency = HelmChartDependency(name=chart.info.name, version=chart.info.version)
            updated_dependencies.add(dependency)

        # Update metadata with the information about charts' dependencies.
        chart_info = dataclasses.replace(chart_info, dependencies=tuple(updated_dependencies))

    # Re-render the Chart.yaml file with the updated dependencies.
    metadata_digest, sources_without_metadata = await concurrently(
        render_chart_metadata(chart_info),
        digest_subset_to_digest(
            DigestSubset(
                source_files.snapshot.digest,
                PathGlobs(
                    ["**/*", *(f"!**/{filename}" for filename in HELM_CHART_METADATA_FILENAMES)]
                ),
            )
        ),
    )

    # Merge all digests that conform chart's content.
    chart_snapshot = await digest_to_snapshot(
        **implicitly(MergeDigests([metadata_digest, sources_without_metadata, subcharts_digest]))
    )

    return HelmChart(address=request.field_set.address, info=chart_info, snapshot=chart_snapshot)


@dataclass(frozen=True)
class FindHelmDeploymentChart(EngineAwareParameter):
    field_set: HelmDeploymentFieldSet

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


@rule(desc="Find Helm deployment's chart", level=LogLevel.DEBUG)
async def find_chart_for_deployment(request: FindHelmDeploymentChart) -> HelmChart:
    address_input = request.field_set.chart.to_address_input()
    address = await resolve_address(**implicitly(address_input))
    wrapped_target = await resolve_target(
        WrappedTargetRequest(address, address_input.description_of_origin), **implicitly()
    )

    found_charts = await _find_charts_by_targets(
        [wrapped_target.target],
        description_of_origin=f"the `helm_deployment` {request.field_set.address}",
    )
    assert len(found_charts) == 1
    return found_charts[0]


def rules():
    return [
        *collect_rules(),
        *sources.rules(),
        *chart_metadata.rules(),
        *chart_inference.rules(),
        *fetch.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	4✔
5
6	import dataclasses	4✔
7	import logging	4✔
8	from collections.abc import Iterable	4✔
9	from dataclasses import dataclass	4✔
10	from typing import Any	4✔
11
12	from pants.backend.helm.dependency_inference import chart as chart_inference	4✔
13	from pants.backend.helm.resolve import fetch	4✔
14	from pants.backend.helm.resolve.artifacts import ResolvedHelmArtifact	4✔
15	from pants.backend.helm.resolve.fetch import FetchedHelmArtifact, FetchHelmArtifactRequest	4✔
16	from pants.backend.helm.subsystems.helm import HelmSubsystem	4✔
17	from pants.backend.helm.target_types import (	4✔
18	HelmArtifactFieldSet,
19	HelmChartFieldSet,
20	HelmChartTarget,
21	HelmDeploymentFieldSet,
22	)
23	from pants.backend.helm.util_rules import chart_metadata, sources	4✔
24	from pants.backend.helm.util_rules.chart_metadata import (	4✔
25	HELM_CHART_METADATA_FILENAMES,
26	HelmChartDependency,
27	HelmChartMetadata,
28	ParseHelmChartMetadataDigest,
29	parse_chart_metadata_from_digest,
30	parse_chart_metadata_from_field,
31	render_chart_metadata,
32	)
33	from pants.backend.helm.util_rules.sources import HelmChartSourceFilesRequest, get_helm_source_files	4✔
34	from pants.engine.addresses import Address	4✔
35	from pants.engine.engine_aware import EngineAwareParameter, EngineAwareReturnType	4✔
36	from pants.engine.fs import (	4✔
37	EMPTY_DIGEST,
38	AddPrefix,
39	Digest,
40	DigestSubset,
41	FileDigest,
42	MergeDigests,
43	PathGlobs,
44	Snapshot,
45	)
46	from pants.engine.internals.build_files import resolve_address	4✔
47	from pants.engine.internals.graph import resolve_target, resolve_targets	4✔
48	from pants.engine.intrinsics import (	4✔
49	add_prefix,
50	digest_subset_to_digest,
51	digest_to_snapshot,
52	merge_digests,
53	)
54	from pants.engine.rules import collect_rules, concurrently, implicitly, rule	4✔
55	from pants.engine.target import DependenciesRequest, Target, WrappedTargetRequest	4✔
56	from pants.util.frozendict import FrozenDict	4✔
57	from pants.util.logging import LogLevel	4✔
58	from pants.util.ordered_set import OrderedSet	4✔
59	from pants.util.strutil import pluralize, softwrap	4✔
60
61	logger = logging.getLogger(__name__)	4✔
62
63
64	class InvalidHelmChartTarget(ValueError):	4✔
65	def __init__(self, target: Target) -> None:	4✔
66	super().__init__(f"The target {target.address} is not a `{HelmChartTarget.alias}`.")	×
67
68
69	@dataclass(frozen=True)	4✔
70	class HelmChart(EngineAwareReturnType):	4✔
71	address: Address	4✔
72	info: HelmChartMetadata	4✔
73	snapshot: Snapshot	4✔
74	artifact: ResolvedHelmArtifact \| None = None	4✔
75
76	@property	4✔
77	def name(self) -> str:	4✔
78	return self.info.name	4✔
79
80	@property	4✔
81	def description(self) -> str:	4✔
82	return self.name	1✔
83
84	@property	4✔
85	def immutable_input_digests(self) -> FrozenDict[str, Digest]:	4✔
86	return FrozenDict({self.name: self.snapshot.digest})	4✔
87
88	def level(self) -> LogLevel \| None:	4✔
89	return LogLevel.DEBUG	4✔
90
91	def message(self) -> str \| None:	4✔
92	msg = f"Built Helm chart '{self.info.name}' from target at {self.address}"	4✔
93	if self.artifact:	4✔
UNCOV 94	msg += f" (resolved using URL {self.artifact.chart_url})."	×
95	return msg	4✔
96
97	def metadata(self) -> dict[str, Any] \| None:	4✔
98	meta: dict[str, Any] = {"name": self.info.name, "address": self.address.spec}	4✔
99	if self.artifact:	4✔
UNCOV 100	meta["artifact"] = self.artifact	×
101	return meta	4✔
102
103	def artifacts(self) -> dict[str, FileDigest \| Snapshot] \| None:	4✔
104	return {"snapshot": self.snapshot}	4✔
105
106
107	@dataclass(frozen=True)	4✔
108	class HelmChartRequest(EngineAwareParameter):	4✔
109	field_set: HelmChartFieldSet	4✔
110
111	@classmethod	4✔
112	def from_target(cls, target: Target) -> HelmChartRequest:	4✔
113	if not HelmChartFieldSet.is_applicable(target):	3✔
114	raise InvalidHelmChartTarget(target)	×
115	return cls(HelmChartFieldSet.create(target))	3✔
116
117	def debug_hint(self) -> str \| None:	4✔
UNCOV 118	return self.field_set.address.spec	×
119
120	def metadata(self) -> dict[str, Any] \| None:	4✔
UNCOV 121	return {"address": self.field_set.address.spec}	×
122
123
124	@rule(desc="Compile third parth Helm chart", level=LogLevel.DEBUG)	4✔
125	async def create_chart_from_artifact(fetched_artifact: FetchedHelmArtifact) -> HelmChart:	4✔
UNCOV 126	metadata = await parse_chart_metadata_from_digest(	×
127	ParseHelmChartMetadataDigest(
128	fetched_artifact.snapshot.digest,
129	description_of_origin=f"the `helm_artifact` {fetched_artifact.address}",
130	)
131	)
UNCOV 132	return HelmChart(	×
133	fetched_artifact.address,
134	metadata,
135	fetched_artifact.snapshot,
136	artifact=fetched_artifact.artifact,
137	)
138
139
140	async def _merge_subchart_digests(charts: Iterable[HelmChart]) -> Digest:	4✔
UNCOV 141	prefixed_chart_digests = await concurrently(	×
142	add_prefix(AddPrefix(chart.snapshot.digest, chart.name)) for chart in charts
143	)
UNCOV 144	merged_digests = await merge_digests(MergeDigests(prefixed_chart_digests))	×
UNCOV 145	return await add_prefix(AddPrefix(merged_digests, "charts"))	×
146
147
148	async def _find_charts_by_targets(	4✔
149	targets: Iterable[Target], *, description_of_origin: str
150	) -> tuple[HelmChart, ...]:
151	# These `Get`s are not ported yet to call-by-name due to the rule cycle with`get_helm_chart`
152	# which calls this function.
153	requests = [	4✔
154	*(
155	get_helm_chart(HelmChartRequest.from_target(target), **implicitly())
156	for target in targets
157	if HelmChartFieldSet.is_applicable(target)
158	),
159	*(
160	create_chart_from_artifact(
161	**implicitly(
162	{
163	FetchHelmArtifactRequest.from_target(
164	target,
165	description_of_origin=description_of_origin,
166	): FetchHelmArtifactRequest
167	}
168	)
169	)
170	for target in targets
171	if HelmArtifactFieldSet.is_applicable(target)
172	),
173	]
174	return await concurrently(requests)	4✔
175
176
177	@rule(desc="Compile Helm chart", level=LogLevel.DEBUG)	4✔
178	async def get_helm_chart(request: HelmChartRequest, subsystem: HelmSubsystem) -> HelmChart:	4✔
179	dependencies, source_files, chart_info = await concurrently(	4✔
180	resolve_targets(**implicitly(DependenciesRequest(request.field_set.dependencies))),
181	get_helm_source_files(
182	HelmChartSourceFilesRequest.for_field_set(
183	request.field_set,
184	include_metadata=False,
185	include_resources=True,
186	include_files=True,
187	)
188	),
189	parse_chart_metadata_from_field(request.field_set.chart),
190	)
191
192	if request.field_set.version.value:	4✔
UNCOV 193	chart_info = dataclasses.replace(chart_info, version=request.field_set.version.value)	×
194
195	subcharts_digest = EMPTY_DIGEST	4✔
196	subcharts = await _find_charts_by_targets(	4✔
197	dependencies, description_of_origin=f"the `helm_chart` {request.field_set.address}"
198	)
199	if subcharts:	4✔
UNCOV 200	logger.debug(	×
201	softwrap(
202	f"""
203	Found {pluralize(len(subcharts), "subchart")} as direct dependencies
204	on Helm chart at: {request.field_set.address}.
205	"""
206	)
207	)
208
UNCOV 209	subcharts_digest = await _merge_subchart_digests(subcharts)	×
210
211	# Update subchart dependencies in the metadata and re-render it.
UNCOV 212	remotes = subsystem.remotes()	×
UNCOV 213	subchart_map: dict[str, HelmChart] = {chart.info.name: chart for chart in subcharts}	×
UNCOV 214	updated_dependencies: OrderedSet[HelmChartDependency] = OrderedSet()	×
UNCOV 215	for dep in chart_info.dependencies:	×
UNCOV 216	updated_dep = dep	×
217
UNCOV 218	if not dep.repository and remotes.default_registry:	×
219	# If the dependency hasn't specified a repository, then we choose the registry with the 'default' alias.
220	default_remote = remotes.default_registry	×
221	updated_dep = dataclasses.replace(updated_dep, repository=default_remote.address)	×
UNCOV 222	elif dep.repository and dep.repository.startswith("@"):	×
223	remote = next(remotes.get(dep.repository))	×
224	updated_dep = dataclasses.replace(updated_dep, repository=remote.address)	×
225
UNCOV 226	if dep.name in subchart_map:	×
UNCOV 227	updated_dep = dataclasses.replace(	×
228	updated_dep, version=subchart_map[dep.name].info.version
229	)
230
UNCOV 231	updated_dependencies.add(updated_dep)	×
232
233	# Include the explicitly provided subchats in the set of dependencies if not already present.
UNCOV 234	updated_dependencies_names = {dep.name for dep in updated_dependencies}	×
UNCOV 235	remaining_subcharts = [	×
236	chart for chart in subcharts if chart.info.name not in updated_dependencies_names
237	]
UNCOV 238	for chart in remaining_subcharts:	×
UNCOV 239	if chart.artifact:	×
UNCOV 240	dependency = HelmChartDependency(	×
241	name=chart.artifact.name,
242	version=chart.artifact.version,
243	repository=chart.artifact.location_url,
244	)
245	else:
UNCOV 246	dependency = HelmChartDependency(name=chart.info.name, version=chart.info.version)	×
UNCOV 247	updated_dependencies.add(dependency)	×
248
249	# Update metadata with the information about charts' dependencies.
UNCOV 250	chart_info = dataclasses.replace(chart_info, dependencies=tuple(updated_dependencies))	×
251
252	# Re-render the Chart.yaml file with the updated dependencies.
253	metadata_digest, sources_without_metadata = await concurrently(	4✔
254	render_chart_metadata(chart_info),
255	digest_subset_to_digest(
256	DigestSubset(
257	source_files.snapshot.digest,
258	PathGlobs(
259	["*/", (f"!*/{filename}" for filename in HELM_CHART_METADATA_FILENAMES)]
260	),
261	)
262	),
263	)
264
265	# Merge all digests that conform chart's content.
266	chart_snapshot = await digest_to_snapshot(	4✔
267	**implicitly(MergeDigests([metadata_digest, sources_without_metadata, subcharts_digest]))
268	)
269
270	return HelmChart(address=request.field_set.address, info=chart_info, snapshot=chart_snapshot)	4✔
271
272
273	@dataclass(frozen=True)	4✔
274	class FindHelmDeploymentChart(EngineAwareParameter):	4✔
275	field_set: HelmDeploymentFieldSet	4✔
276
277	def debug_hint(self) -> str \| None:	4✔
UNCOV 278	return self.field_set.address.spec	×
279
280
281	@rule(desc="Find Helm deployment's chart", level=LogLevel.DEBUG)	4✔
282	async def find_chart_for_deployment(request: FindHelmDeploymentChart) -> HelmChart:	4✔
283	address_input = request.field_set.chart.to_address_input()	3✔
284	address = await resolve_address(**implicitly(address_input))	3✔
285	wrapped_target = await resolve_target(	3✔
286	WrappedTargetRequest(address, address_input.description_of_origin), **implicitly()
287	)
288
289	found_charts = await _find_charts_by_targets(	3✔
290	[wrapped_target.target],
291	description_of_origin=f"the `helm_deployment` {request.field_set.address}",
292	)
293	assert len(found_charts) == 1	3✔
294	return found_charts[0]	3✔
295
296
297	def rules():	4✔
298	return [	4✔
299	*collect_rules(),
300	*sources.rules(),
301	*chart_metadata.rules(),
302	*chart_inference.rules(),
303	*fetch.rules(),
304	]

pantsbuild / pants / 22285099215

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