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/util_rules/post_renderer.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	from dataclasses import dataclass	×
UNCOV 8	from itertools import chain	×
UNCOV 9	from typing import Any	×
10
UNCOV 11	from pants.backend.docker.goals.package_image import DockerPackageFieldSet	×
UNCOV 12	from pants.backend.docker.subsystems import dockerfile_parser	×
UNCOV 13	from pants.backend.docker.subsystems.docker_options import DockerOptions	×
UNCOV 14	from pants.backend.docker.target_types import (	×
15	DockerImageTags,
16	DockerImageTagsRequest,
17	get_docker_image_tags,
18	)
UNCOV 19	from pants.backend.docker.util_rules import (	×
20	docker_binary,
21	docker_build_args,
22	docker_build_context,
23	docker_build_env,
24	dockerfile,
25	)
UNCOV 26	from pants.backend.docker.util_rules.docker_build_context import (	×
27	DockerBuildContext,
28	DockerBuildContextRequest,
29	create_docker_build_context,
30	)
UNCOV 31	from pants.backend.helm.dependency_inference.deployment import (	×
32	FirstPartyHelmDeploymentMappingRequest,
33	first_party_helm_deployment_mapping,
34	)
UNCOV 35	from pants.backend.helm.subsystems import post_renderer	×
UNCOV 36	from pants.backend.helm.subsystems.post_renderer import SetupHelmPostRenderer	×
UNCOV 37	from pants.backend.helm.target_types import HelmDeploymentFieldSet	×
UNCOV 38	from pants.engine.addresses import Address, Addresses	×
UNCOV 39	from pants.engine.engine_aware import EngineAwareParameter	×
UNCOV 40	from pants.engine.internals.graph import resolve_target, resolve_targets	×
UNCOV 41	from pants.engine.rules import collect_rules, concurrently, implicitly, rule	×
UNCOV 42	from pants.engine.target import WrappedTargetRequest	×
UNCOV 43	from pants.engine.unions import UnionMembership	×
UNCOV 44	from pants.util.logging import LogLevel	×
UNCOV 45	from pants.util.strutil import bullet_list, softwrap	×
46
UNCOV 47	logger = logging.getLogger(__name__)	×
48
49
UNCOV 50	@dataclass(frozen=True)	×
UNCOV 51	class HelmDeploymentPostRendererRequest(EngineAwareParameter):	×
UNCOV 52	field_set: HelmDeploymentFieldSet	×
53
UNCOV 54	def debug_hint(self) -> str \| None:	×
55	return self.field_set.address.spec	×
56
UNCOV 57	def metadata(self) -> dict[str, Any] \| None:	×
58	return {"address": self.field_set.address.spec}	×
59
60
UNCOV 61	async def _obtain_custom_image_tags(	×
62	address: Address, union_membership: UnionMembership
63	) -> DockerImageTags:
64	wrapped_target = await resolve_target(	×
65	WrappedTargetRequest(address, description_of_origin="<infallible>"), **implicitly()
66	)
67
68	image_tags_requests = union_membership.get(DockerImageTagsRequest)	×
69	found_image_tags = await concurrently(	×
70	get_docker_image_tags(
71	**implicitly({image_tags_request_cls(wrapped_target.target): DockerImageTagsRequest})
72	)
73	for image_tags_request_cls in image_tags_requests
74	if image_tags_request_cls.is_applicable(wrapped_target.target)
75	)
76
77	return DockerImageTags(chain.from_iterable(found_image_tags))	×
78
79
UNCOV 80	@rule(desc="Prepare Helm deployment post-renderer", level=LogLevel.DEBUG)	×
UNCOV 81	async def prepare_post_renderer_for_helm_deployment(	×
82	request: HelmDeploymentPostRendererRequest,
83	union_membership: UnionMembership,
84	docker_options: DockerOptions,
85	) -> SetupHelmPostRenderer:
86	mapping = await first_party_helm_deployment_mapping(	×
87	FirstPartyHelmDeploymentMappingRequest(request.field_set), **implicitly()
88	)
89
90	docker_addresses = [addr for _, addr in mapping.indexed_docker_addresses.values()]	×
91
92	logger.debug(	×
93	softwrap(
94	f"""
95	Resolving Docker image references for targets:
96
97	{bullet_list([addr.spec for addr in docker_addresses])}
98	"""
99	)
100	)
101	docker_contexts = await concurrently(	×
102	create_docker_build_context(
103	DockerBuildContextRequest(
104	address=addr,
105	build_upstream_images=False,
106	),
107	**implicitly(),
108	)
109	for addr in docker_addresses
110	)
111
112	docker_targets = await resolve_targets(**implicitly(Addresses(docker_addresses)))	×
113	field_sets = [DockerPackageFieldSet.create(tgt) for tgt in docker_targets]	×
114
115	async def resolve_docker_image_ref(address: Address, context: DockerBuildContext) -> str \| None:	×
116	docker_field_sets = [fs for fs in field_sets if fs.address == address]	×
117	if not docker_field_sets:	×
118	return None	×
119
120	additional_image_tags = await _obtain_custom_image_tags(address, union_membership)	×
121
122	docker_field_set = docker_field_sets[0]	×
123	image_refs = docker_field_set.image_refs(	×
124	default_repository=docker_options.default_repository,
125	registries=docker_options.registries(),
126	interpolation_context=context.interpolation_context,
127	additional_tags=tuple(additional_image_tags),
128	)
129
130	# Choose first non-latest image reference found, or fallback to 'latest'.
131	found_ref: str \| None = None	×
132	fallback_ref: str \| None = None	×
133	for registry in image_refs:	×
134	for tag in registry.tags:	×
135	ref = tag.full_name	×
136	if ref.endswith(":latest"):	×
137	fallback_ref = ref	×
138	else:
139	found_ref = ref	×
140	break	×
141
142	resolved_ref = found_ref or fallback_ref	×
143	if resolved_ref:	×
144	logger.debug(f"Resolved Docker image ref '{resolved_ref}' for address {address}.")	×
145	else:
146	logger.warning(f"Could not resolve a valid image ref for Docker target {address}.")	×
147
148	return resolved_ref	×
149
150	docker_addr_ref_mapping = {	×
151	addr: await resolve_docker_image_ref(addr, ctx)
152	for addr, ctx in zip(docker_addresses, docker_contexts)
153	}
154
155	def find_replacement(value: tuple[str, Address]) -> str \| None:	×
156	_, addr = value	×
157	return docker_addr_ref_mapping.get(addr)	×
158
159	replacements = mapping.indexed_docker_addresses.transform_values(find_replacement)	×
160
161	return SetupHelmPostRenderer(	×
162	replacements, description_of_origin=f"the `helm_deployment` {request.field_set.address}"
163	)
164
165
UNCOV 166	def rules():	×
UNCOV 167	return [	×
168	*collect_rules(),
169	*docker_binary.rules(),
170	*docker_build_args.rules(),
171	*docker_build_context.rules(),
172	*docker_build_env.rules(),
173	*dockerfile.rules(),
174	*dockerfile_parser.rules(),
175	*post_renderer.rules(),
176	]

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