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/dependency_inference/deployment.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 fnmatch	×
UNCOV 7	import logging	×
UNCOV 8	import re	×
UNCOV 9	from dataclasses import dataclass, field	×
UNCOV 10	from pathlib import PurePath	×
UNCOV 11	from typing import Any	×
12
UNCOV 13	from pants.backend.docker.target_types import AllDockerImageTargets	×
UNCOV 14	from pants.backend.docker.target_types import rules as docker_target_types_rules	×
UNCOV 15	from pants.backend.docker.utils import image_ref_regexp	×
UNCOV 16	from pants.backend.helm.dependency_inference.subsystem import (	×
17	HelmInferSubsystem,
18	UnownedDependencyError,
19	UnownedHelmDependencyUsage,
20	)
UNCOV 21	from pants.backend.helm.subsystems import k8s_parser	×
UNCOV 22	from pants.backend.helm.subsystems.k8s_parser import ParseKubeManifestRequest, parse_kube_manifest	×
UNCOV 23	from pants.backend.helm.target_types import HelmDeploymentFieldSet	×
UNCOV 24	from pants.backend.helm.target_types import rules as helm_target_types_rules	×
UNCOV 25	from pants.backend.helm.util_rules import renderer	×
UNCOV 26	from pants.backend.helm.util_rules.renderer import (	×
27	HelmDeploymentCmd,
28	HelmDeploymentRequest,
29	run_renderer,
30	)
UNCOV 31	from pants.backend.helm.utils.yaml import FrozenYamlIndex, MutableYamlIndex	×
UNCOV 32	from pants.build_graph.address import MaybeAddress	×
UNCOV 33	from pants.engine.addresses import Address	×
UNCOV 34	from pants.engine.engine_aware import EngineAwareParameter, EngineAwareReturnType	×
UNCOV 35	from pants.engine.fs import FileEntry	×
UNCOV 36	from pants.engine.internals.build_files import maybe_resolve_address, resolve_address	×
UNCOV 37	from pants.engine.internals.graph import determine_explicitly_provided_dependencies	×
UNCOV 38	from pants.engine.internals.native_engine import AddressInput, AddressParseException	×
UNCOV 39	from pants.engine.intrinsics import get_digest_entries	×
UNCOV 40	from pants.engine.rules import collect_rules, concurrently, implicitly, rule	×
UNCOV 41	from pants.engine.target import DependenciesRequest, InferDependenciesRequest, InferredDependencies	×
UNCOV 42	from pants.engine.unions import UnionRule	×
UNCOV 43	from pants.util.logging import LogLevel	×
UNCOV 44	from pants.util.ordered_set import FrozenOrderedSet, OrderedSet	×
UNCOV 45	from pants.util.strutil import pluralize, softwrap	×
46
UNCOV 47	logger = logging.getLogger(__name__)	×
48
49
UNCOV 50	@dataclass(frozen=True)	×
UNCOV 51	class AnalyseHelmDeploymentRequest(EngineAwareParameter):	×
UNCOV 52	field_set: HelmDeploymentFieldSet	×
53
UNCOV 54	def debug_hint(self) -> str \| None:	×
55	return self.field_set.address.spec	×
56
57
UNCOV 58	@dataclass(frozen=True)	×
UNCOV 59	class HelmDeploymentReport(EngineAwareReturnType):	×
UNCOV 60	address: Address	×
UNCOV 61	image_refs: FrozenYamlIndex[str]	×
62
UNCOV 63	@property	×
UNCOV 64	def all_image_refs(self) -> FrozenOrderedSet[str]:	×
UNCOV 65	return FrozenOrderedSet(self.image_refs.values())	×
66
UNCOV 67	def level(self) -> LogLevel \| None:	×
68	return LogLevel.DEBUG	×
69
UNCOV 70	def metadata(self) -> dict[str, Any] \| None:	×
71	return {"address": self.address, "image_refs": self.image_refs}	×
72
73
UNCOV 74	@rule(desc="Analyse Helm deployment", level=LogLevel.DEBUG)	×
UNCOV 75	async def analyse_deployment(request: AnalyseHelmDeploymentRequest) -> HelmDeploymentReport:	×
76	rendered_deployment = await run_renderer(	×
77	**implicitly(
78	HelmDeploymentRequest(
79	cmd=HelmDeploymentCmd.RENDER,
80	field_set=request.field_set,
81	description=f"Rendering Helm deployment {request.field_set.address}",
82	)
83	)
84	)
85
86	rendered_entries = await get_digest_entries(rendered_deployment.snapshot.digest)	×
87	parsed_manifests = await concurrently(	×
88	parse_kube_manifest(ParseKubeManifestRequest(file=entry), **implicitly())
89	for entry in rendered_entries
90	if isinstance(entry, FileEntry)
91	)
92
93	# Build YAML index of Docker image refs for future processing during dependency inference or post-rendering.
94	image_refs_index: MutableYamlIndex[str] = MutableYamlIndex()	×
95	for manifest in parsed_manifests:	×
96	for entry in manifest.found_image_refs:	×
97	image_refs_index.insert(	×
98	file_path=PurePath(manifest.filename),
99	document_index=entry.document_index,
100	yaml_path=entry.path,
101	item=entry.unparsed_image_ref,
102	)
103
104	return HelmDeploymentReport(	×
105	address=request.field_set.address, image_refs=image_refs_index.frozen()
106	)
107
108
UNCOV 109	@dataclass(frozen=True)	×
UNCOV 110	class FirstPartyHelmDeploymentMappingRequest(EngineAwareParameter):	×
UNCOV 111	field_set: HelmDeploymentFieldSet	×
112
UNCOV 113	def debug_hint(self) -> str \| None:	×
114	return self.field_set.address.spec	×
115
116
UNCOV 117	@dataclass(frozen=True)	×
UNCOV 118	class FirstPartyHelmDeploymentMapping:	×
119	"""A mapping between `helm_deployment` target addresses and tuples made up of a Docker image
120	reference and a `docker_image` target address.
121
122	The tuples of Docker image references and addresses are stored in a YAML index so we can track
123	the locations in which the Docker image refs appear in the deployment files.
124	"""
125
UNCOV 126	address: Address	×
UNCOV 127	indexed_docker_addresses: FrozenYamlIndex[tuple[str, Address]]	×
128
129
UNCOV 130	@dataclass(frozen=True)	×
UNCOV 131	class _FirstPartyHelmDeploymentMappingRequest(EngineAwareParameter):	×
UNCOV 132	field_set: HelmDeploymentFieldSet	×
133
134
UNCOV 135	@rule	×
UNCOV 136	async def _first_party_helm_deployment_mapping(	×
137	request: _FirstPartyHelmDeploymentMappingRequest,
138	docker_targets: AllDockerImageTargets,
139	helm_infer: HelmInferSubsystem,
140	) -> FirstPartyHelmDeploymentMapping:
141	deployment_report = await analyse_deployment(AnalyseHelmDeploymentRequest(request.field_set))	×
142
143	def image_ref_to_address_input(image_ref: str) -> tuple[str, AddressInput] \| None:	×
144	try:	×
145	return image_ref, AddressInput.parse(	×
146	image_ref,
147	description_of_origin=f"the helm_deployment at {request.field_set.address}",
148	relative_to=request.field_set.address.spec_path,
149	)
150	except AddressParseException:	×
151	return None	×
152
153	indexed_address_inputs = deployment_report.image_refs.transform_values(	×
154	image_ref_to_address_input
155	)
156	maybe_addresses = await concurrently(	×
157	maybe_resolve_address(ai) for _, ai in indexed_address_inputs.values()
158	)
159
160	docker_target_addresses = {tgt.address for tgt in docker_targets}	×
161	maybe_addresses_by_ref = {	×
162	ref: maybe_addr
163	for ((ref, _), maybe_addr) in zip(indexed_address_inputs.values(), maybe_addresses)
164	}
165
166	resolver = ImageReferenceResolver(helm_infer, maybe_addresses_by_ref, docker_target_addresses)	×
167
168	indexed_docker_addresses = indexed_address_inputs.transform_values(	×
169	lambda image_ref_ai: resolver.image_ref_to_actual_address(image_ref_ai[0])
170	)
171
172	resolver.report_errors()	×
173	return FirstPartyHelmDeploymentMapping(	×
174	address=request.field_set.address,
175	indexed_docker_addresses=indexed_docker_addresses,
176	)
177
178
UNCOV 179	@rule	×
UNCOV 180	async def first_party_helm_deployment_mapping(	×
181	request: FirstPartyHelmDeploymentMappingRequest,
182	helm_infer: HelmInferSubsystem,
183	) -> FirstPartyHelmDeploymentMapping:
184	if not helm_infer.deployment_dependencies:	×
185	return FirstPartyHelmDeploymentMapping(	×
186	request.field_set.address,
187	FrozenYamlIndex.empty(),
188	)
189	# Use a small proxy rule to make sure we don't calculate AllDockerImageTargets
190	# if `[helm-infer].deployment_dependencies` is set to true.
191	return await _first_party_helm_deployment_mapping(	×
192	_FirstPartyHelmDeploymentMappingRequest(field_set=request.field_set), **implicitly()
193	)
194
195
UNCOV 196	@dataclass	×
UNCOV 197	class ImageReferenceResolver:	×
198	"""Attempt to resolve images to their references.
199
200	Errors are stored internally and are surfaced by a call to `report_errors`, so we can report all
201	problems and avoid only raising 1 per run.
202	"""
203
UNCOV 204	helm_infer: HelmInferSubsystem	×
UNCOV 205	maybe_addresses_by_ref: dict[str, MaybeAddress]	×
UNCOV 206	docker_target_addresses: set[Address]	×
207
UNCOV 208	errors: list[str] = field(default_factory=list)	×
209
UNCOV 210	def image_ref_to_actual_address(self, image_ref: str) -> tuple[str, Address] \| None:	×
211	"""Attempt to resolve an image reference to its Pants Address or the docker image."""
UNCOV 212	maybe_addr = self.maybe_addresses_by_ref.get(image_ref)	×
UNCOV 213	if not maybe_addr:	×
214	return None	×
UNCOV 215	if not isinstance(maybe_addr.val, Address):	×
216	# obviously intended to be a Pants target
UNCOV 217	if (	×
218	image_ref.startswith("//")
219	or image_ref.startswith("./")
220	or image_ref.startswith(":")
221	):
UNCOV 222	message = f"`{image_ref}` was supplied but the docker_image target at `{maybe_addr.val}` does not exist."	×
UNCOV 223	self._handle_missing_docker_image(message)	×
UNCOV 224	return None	×
225	# explicit 3rd party
UNCOV 226	elif self._image_ref_is_known_external(image_ref):	×
UNCOV 227	return None	×
228	else:
UNCOV 229	message = f"""\	×
230	`{image_ref}` was supplied, but Pants cannot determine
231	whether this should be a target's address or a 3rd-party dependency.
232	If this should be an external image, add `{image_ref}` to `[{HelmInferSubsystem.options_scope}].external_docker_images`
233	If this should be a target address, use an absolute path instead (possibly `//{image_ref}`).
234	"""
UNCOV 235	self._handle_missing_docker_image(message)	×
UNCOV 236	return None	×
237
UNCOV 238	if maybe_addr.val not in self.docker_target_addresses:	×
UNCOV 239	message = f"The address `{image_ref}` was supplied, but the target at `{maybe_addr.val}` is not a docker_image target."	×
UNCOV 240	self._handle_missing_docker_image(message)	×
UNCOV 241	return None	×
UNCOV 242	return image_ref, maybe_addr.val	×
243
UNCOV 244	def _image_ref_is_known_external(self, image_ref: str) -> bool:	×
UNCOV 245	parsed = re.match(image_ref_regexp, image_ref.strip("\"'"))	×
UNCOV 246	if not parsed:	×
247	return False	×
UNCOV 248	if parsed.group("registry"):	×
UNCOV 249	image_name = parsed.group("registry") + "/" + parsed.group("repository")	×
250	else:
UNCOV 251	image_name = parsed.group("repository")	×
252
UNCOV 253	return any(	×
254	(fnmatch.fnmatch(image_name, pattern) or fnmatch.fnmatch(image_ref, pattern))
255	for pattern in self.helm_infer.external_docker_images
256	)
257
UNCOV 258	def _handle_missing_docker_image(self, message):	×
UNCOV 259	self.errors.append(message)	×
260
UNCOV 261	def report_errors(self):	×
262	"""Raise all gathered errors.
263
264	Invoke this method after all resolving is done.
265	"""
UNCOV 266	if not self.errors:	×
UNCOV 267	return	×
268
UNCOV 269	message = "\n".join(	×
270	[
271	"Error resolving Docker image dependency of a Helm chart.",
272	*self.errors,
273	f"The behavior for unowned imports can also be set with the `[{HelmInferSubsystem.options_scope}].unowned_dependency_behavior`",
274	]
275	)
UNCOV 276	if self.helm_infer.unowned_dependency_behavior == UnownedHelmDependencyUsage.RaiseError:	×
UNCOV 277	raise UnownedDependencyError(message)	×
278	elif self.helm_infer.unowned_dependency_behavior == UnownedHelmDependencyUsage.LogWarning:	×
279	logging.warning(message)	×
280	else:
281	return	×
282
283
UNCOV 284	class InferHelmDeploymentDependenciesRequest(InferDependenciesRequest):	×
UNCOV 285	infer_from = HelmDeploymentFieldSet	×
286
287
UNCOV 288	@rule(desc="Find the dependencies needed by a Helm deployment")	×
UNCOV 289	async def inject_deployment_dependencies(	×
290	request: InferHelmDeploymentDependenciesRequest,
291	infer_subsystem: HelmInferSubsystem,
292	) -> InferredDependencies:
293	get_address = resolve_address(	×
294	**implicitly({request.field_set.chart.to_address_input(): AddressInput})
295	)
296	get_explicit_deps = determine_explicitly_provided_dependencies(	×
297	**implicitly(DependenciesRequest(request.field_set.dependencies))
298	)
299
300	chart_address, explicitly_provided_deps, mapping = await concurrently(	×
301	get_address,
302	get_explicit_deps,
303	first_party_helm_deployment_mapping(
304	FirstPartyHelmDeploymentMappingRequest(request.field_set), **implicitly()
305	),
306	)
307
308	dependencies: OrderedSet[Address] = OrderedSet()	×
309	dependencies.add(chart_address)	×
310
311	for imager_ref, candidate_address in mapping.indexed_docker_addresses.values():	×
312	matches = frozenset([candidate_address]).difference(explicitly_provided_deps.includes)	×
313	explicitly_provided_deps.maybe_warn_of_ambiguous_dependency_inference(	×
314	matches,
315	request.field_set.address,
316	context=softwrap(
317	f"""
318	The Helm deployment {request.field_set.address} declares
319	{imager_ref} as Docker image reference
320	"""
321	),
322	import_reference="manifest",
323	)
324
325	maybe_disambiguated = explicitly_provided_deps.disambiguated(matches)	×
326	if maybe_disambiguated:	×
327	dependencies.add(maybe_disambiguated)	×
328
329	logging.debug(	×
330	f"Found {pluralize(len(dependencies), 'dependency')} for target {request.field_set.address}"
331	)
332	return InferredDependencies(dependencies)	×
333
334
UNCOV 335	def rules():	×
UNCOV 336	return [	×
337	*collect_rules(),
338	*renderer.rules(),
339	*k8s_parser.rules(),
340	*helm_target_types_rules(),
341	*docker_target_types_rules(),
342	UnionRule(InferDependenciesRequest, InferHelmDeploymentDependenciesRequest),
343	]

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