19015773527

Committed 02 Nov 2025 05:33PM UTC coverage: 17.872% (-62.4%) from 80.3%

Build # 19015773527

Build Type

Pull #22816

github

Committed by

web-flow

Commit Message

Merge a12d75757 into 6c024e162

Pull Request Pull Request #22816: Update Pants internal Python to 3.14

Run Details

4 of 5 new or added lines in 3 files covered. (80.0%)

28452 existing lines in 683 files now uncovered.

9831 of 55007 relevant lines covered (17.87%)

0.18 hits per line

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

0.0

/src/python/pants/backend/helm/resolve/fetch.py

1	# Copyright 2022 Pants project contributors (see CONTRIBUTORS.md).
2	# Licensed under the Apache License, Version 2.0 (see LICENSE).
3
UNCOV 4	from __future__ import annotations	×
5
UNCOV 6	import logging	×
UNCOV 7	import os	×
UNCOV 8	from dataclasses import dataclass	×
UNCOV 9	from typing import Any	×
10
UNCOV 11	from pants.backend.helm.resolve import artifacts	×
UNCOV 12	from pants.backend.helm.resolve.artifacts import (	×
13	HelmArtifact,
14	ResolvedHelmArtifact,
15	resolved_helm_artifact,
16	)
UNCOV 17	from pants.backend.helm.target_types import HelmArtifactFieldSet, HelmArtifactTarget	×
UNCOV 18	from pants.backend.helm.util_rules import tool	×
UNCOV 19	from pants.backend.helm.util_rules.tool import HelmProcess	×
UNCOV 20	from pants.engine.addresses import Address	×
UNCOV 21	from pants.engine.engine_aware import EngineAwareParameter, EngineAwareReturnType	×
UNCOV 22	from pants.engine.fs import (	×
23	CreateDigest,
24	DigestSubset,
25	Directory,
26	FileDigest,
27	PathGlobs,
28	RemovePrefix,
29	Snapshot,
30	)
UNCOV 31	from pants.engine.intrinsics import (	×
32	create_digest,
33	digest_subset_to_digest,
34	digest_to_snapshot,
35	remove_prefix,
36	)
UNCOV 37	from pants.engine.process import execute_process_or_raise	×
UNCOV 38	from pants.engine.rules import collect_rules, concurrently, implicitly, rule	×
UNCOV 39	from pants.engine.target import Target	×
UNCOV 40	from pants.util.logging import LogLevel	×
UNCOV 41	from pants.util.strutil import softwrap	×
42
UNCOV 43	logger = logging.getLogger(__name__)	×
44
45
UNCOV 46	class InvalidHelmArtifactTarget(Exception):	×
UNCOV 47	def __init__(self, target: Target) -> None:	×
48	super().__init__(	×
49	softwrap(
50	f"""
51	Can not fetch a Helm artifact for a `{target.alias}` target.
52
53	Helm artifacts are defined using target `{HelmArtifactTarget.alias}`.
54	"""
55	)
56	)
57
58
UNCOV 59	@dataclass(frozen=True)	×
UNCOV 60	class FetchHelmArtifactRequest(EngineAwareParameter):	×
UNCOV 61	field_set: HelmArtifactFieldSet	×
UNCOV 62	description_of_origin: str	×
63
UNCOV 64	@classmethod	×
UNCOV 65	def from_target(cls, target: Target, *, description_of_origin: str) -> FetchHelmArtifactRequest:	×
UNCOV 66	if not HelmArtifactFieldSet.is_applicable(target):	×
67	raise InvalidHelmArtifactTarget(target)	×
68
UNCOV 69	return cls(	×
70	field_set=HelmArtifactFieldSet.create(target),
71	description_of_origin=description_of_origin,
72	)
73
UNCOV 74	def debug_hint(self) -> str \| None:	×
75	return f"{self.field_set.address} from {self.description_of_origin}"	×
76
UNCOV 77	def metadata(self) -> dict[str, Any] \| None:	×
78	return {	×
79	"address": self.field_set.address.spec,
80	"description_of_origin": self.description_of_origin,
81	}
82
83
UNCOV 84	@dataclass(frozen=True)	×
UNCOV 85	class FetchedHelmArtifact(EngineAwareReturnType):	×
UNCOV 86	artifact: ResolvedHelmArtifact	×
UNCOV 87	snapshot: Snapshot	×
88
UNCOV 89	@property	×
UNCOV 90	def address(self) -> Address:	×
91	return self.artifact.address	×
92
UNCOV 93	def level(self) -> LogLevel \| None:	×
94	return LogLevel.DEBUG	×
95
UNCOV 96	def message(self) -> str \| None:	×
97	return softwrap(	×
98	f"""
99	Fetched Helm artifact '{self.artifact.name}' with version {self.artifact.version}
100	using URL: {self.artifact.chart_url}
101	"""
102	)
103
UNCOV 104	def metadata(self) -> dict[str, Any] \| None:	×
105	return {"artifact": self.artifact}	×
106
UNCOV 107	def artifacts(self) -> dict[str, FileDigest \| Snapshot] \| None:	×
108	return {"snapshot": self.snapshot}	×
109
110
UNCOV 111	def assemble_pull_target(resolved_artifact: ResolvedHelmArtifact) -> list[str]:	×
112	"""The target of `pull` need different args depending on whether it's an OCI Registry or a Helm
113	Chart Repository."""
114	is_oci = resolved_artifact.requirement.location.spec.startswith("oci://")	×
115	if is_oci:	×
116	return [f"{resolved_artifact.location_url}/{resolved_artifact.name}"]	×
117	else:
118	return [resolved_artifact.name, "--repo", resolved_artifact.location_url]	×
119
120
UNCOV 121	@rule(desc="Fetch Helm artifact", level=LogLevel.DEBUG)	×
UNCOV 122	async def fetch_helm_artifact(request: FetchHelmArtifactRequest) -> FetchedHelmArtifact:	×
123	download_prefix = "__downloads"	×
124
125	empty_download_digest, resolved_artifact = await concurrently(	×
126	create_digest(CreateDigest([Directory(download_prefix)])),
127	resolved_helm_artifact(
128	**implicitly({HelmArtifact.from_field_set(request.field_set): HelmArtifact})
129	),
130	)
131
132	download_result = await execute_process_or_raise(	×
133	**implicitly(
134	HelmProcess(
135	argv=[
136	"pull",
137	*assemble_pull_target(resolved_artifact),
138	"--version",
139	resolved_artifact.version,
140	"--destination",
141	download_prefix,
142	"--untar",
143	],
144	input_digest=empty_download_digest,
145	description=f"Pulling Helm Chart '{resolved_artifact.name}' with version {resolved_artifact.version}.",
146	output_directories=(download_prefix,),
147	level=LogLevel.DEBUG,
148	)
149	)
150	)
151
152	raw_output_digest = await remove_prefix(	×
153	RemovePrefix(download_result.output_digest, download_prefix)
154	)
155
156	# The download result will come with a tarball alongside the unzipped sources, pick the chart sources only.
157	artifact_sources_digest = await digest_subset_to_digest(	×
158	DigestSubset(raw_output_digest, PathGlobs([os.path.join(resolved_artifact.name, "**")]))
159	)
160	artifact_snapshot = await digest_to_snapshot(	×
161	**implicitly(RemovePrefix(artifact_sources_digest, resolved_artifact.name))
162	)
163
164	return FetchedHelmArtifact(artifact=resolved_artifact, snapshot=artifact_snapshot)	×
165
166
UNCOV 167	def rules():	×
UNCOV 168	return [collect_rules(), artifacts.rules(), *tool.rules()]	×

pantsbuild / pants / 19015773527

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