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/subsystems/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	import os	×
UNCOV 8	import pkgutil	×
UNCOV 9	import shlex	×
UNCOV 10	from collections.abc import Iterable, Mapping	×
UNCOV 11	from dataclasses import dataclass	×
UNCOV 12	from pathlib import PurePath	×
UNCOV 13	from textwrap import dedent # noqa: PNT20	×
UNCOV 14	from typing import Any	×
15
UNCOV 16	import yaml	×
17
UNCOV 18	from pants.backend.helm.utils.yaml import FrozenYamlIndex	×
UNCOV 19	from pants.backend.python.subsystems.python_tool_base import PythonToolRequirementsBase	×
UNCOV 20	from pants.backend.python.target_types import EntryPoint	×
UNCOV 21	from pants.backend.python.util_rules import pex	×
UNCOV 22	from pants.backend.python.util_rules.pex import (	×
23	VenvPex,
24	VenvPexProcess,
25	create_venv_pex,
26	setup_venv_pex_process,
27	)
UNCOV 28	from pants.core.goals.run import RunFieldSet, RunRequest, generate_run_request	×
UNCOV 29	from pants.core.util_rules.system_binaries import CatBinary	×
UNCOV 30	from pants.engine.addresses import UnparsedAddressInputs	×
UNCOV 31	from pants.engine.engine_aware import EngineAwareParameter, EngineAwareReturnType	×
UNCOV 32	from pants.engine.fs import CreateDigest, Digest, FileContent	×
UNCOV 33	from pants.engine.internals.graph import find_valid_field_sets, resolve_targets	×
UNCOV 34	from pants.engine.internals.native_engine import MergeDigests	×
UNCOV 35	from pants.engine.intrinsics import create_digest, merge_digests	×
UNCOV 36	from pants.engine.rules import collect_rules, concurrently, implicitly, rule	×
UNCOV 37	from pants.engine.target import FieldSetsPerTargetRequest	×
UNCOV 38	from pants.util.frozendict import FrozenDict	×
UNCOV 39	from pants.util.logging import LogLevel	×
UNCOV 40	from pants.util.strutil import bullet_list, pluralize, softwrap	×
41
UNCOV 42	logger = logging.getLogger(__name__)	×
43
UNCOV 44	_HELM_POSTRENDERER_SOURCE = "post_renderer_main.py"	×
UNCOV 45	_HELM_POSTRENDERER_PACKAGE = "pants.backend.helm.subsystems"	×
46
47
UNCOV 48	class HelmPostRendererSubsystem(PythonToolRequirementsBase):	×
UNCOV 49	options_scope = "helm-post-renderer"	×
UNCOV 50	help_short = "Used perform modifications to the final output produced by Helm charts when they've been fully rendered."	×
51
UNCOV 52	default_requirements = [	×
53	"yamlpath>=3.6.0,<4",
54	"ruamel.yaml>=0.15.96,!=0.17.0,!=0.17.1,!=0.17.2,!=0.17.5,<=0.17.21",
55	]
56
UNCOV 57	register_interpreter_constraints = True	×
UNCOV 58	default_interpreter_constraints = ["CPython>=3.9,<3.14"]	×
59
UNCOV 60	default_lockfile_resource = (_HELM_POSTRENDERER_PACKAGE, "post_renderer.lock")	×
61
62
UNCOV 63	_HELM_POST_RENDERER_TOOL = "__pants_helm_post_renderer.py"	×
64
65
UNCOV 66	@dataclass(frozen=True)	×
UNCOV 67	class _HelmPostRendererTool:	×
UNCOV 68	pex: VenvPex	×
69
70
UNCOV 71	@rule(desc="Setup Helm post renderer binaries", level=LogLevel.DEBUG)	×
UNCOV 72	async def setup_post_renderer_tool(	×
73	post_renderer: HelmPostRendererSubsystem,
74	) -> _HelmPostRendererTool:
75	post_renderer_sources = pkgutil.get_data(_HELM_POSTRENDERER_PACKAGE, _HELM_POSTRENDERER_SOURCE)	×
76	if not post_renderer_sources:	×
77	raise ValueError(	×
78	f"Unable to find source to {_HELM_POSTRENDERER_SOURCE!r} in {_HELM_POSTRENDERER_PACKAGE}"
79	)
80
81	post_renderer_content = FileContent(	×
82	path=_HELM_POST_RENDERER_TOOL, content=post_renderer_sources, is_executable=True
83	)
84	post_renderer_digest = await create_digest(CreateDigest([post_renderer_content]))	×
85
86	post_renderer_pex = await create_venv_pex(	×
87	**implicitly(
88	post_renderer.to_pex_request(
89	main=EntryPoint(PurePath(post_renderer_content.path).stem),
90	sources=post_renderer_digest,
91	)
92	)
93	)
94	return _HelmPostRendererTool(post_renderer_pex)	×
95
96
UNCOV 97	HELM_POST_RENDERER_CFG_FILENAME = "post_renderer.cfg.yaml"	×
UNCOV 98	_HELM_POST_RENDERER_WRAPPER_SCRIPT = "post_renderer_wrapper.sh"	×
99
100
UNCOV 101	@dataclass(frozen=True)	×
UNCOV 102	class SetupHelmPostRenderer(EngineAwareParameter):	×
103	"""Request for a post-renderer process that will perform a series of replacements in the
104	generated files."""
105
UNCOV 106	replacements: FrozenYamlIndex[str]	×
UNCOV 107	description_of_origin: str	×
UNCOV 108	extra_post_renderers: UnparsedAddressInputs \| None = None	×
109
UNCOV 110	def debug_hint(self) -> str \| None:	×
111	return self.description_of_origin	×
112
113
UNCOV 114	@dataclass(frozen=True)	×
UNCOV 115	class HelmPostRenderer(EngineAwareReturnType):	×
UNCOV 116	exe: str	×
UNCOV 117	digest: Digest	×
UNCOV 118	immutable_input_digests: FrozenDict[str, Digest]	×
UNCOV 119	env: FrozenDict[str, str]	×
UNCOV 120	append_only_caches: FrozenDict[str, str]	×
UNCOV 121	description_of_origin: str	×
122
UNCOV 123	def __init__(	×
124	self,
125	*,
126	exe: str,
127	digest: Digest,
128	description_of_origin: str,
129	env: Mapping[str, str] \| None = None,
130	immutable_input_digests: Mapping[str, Digest] \| None = None,
131	append_only_caches: Mapping[str, str] \| None = None,
132	) -> None:
133	object.__setattr__(self, "exe", exe)	×
134	object.__setattr__(self, "digest", digest)	×
135	object.__setattr__(self, "description_of_origin", description_of_origin)	×
136	object.__setattr__(self, "env", FrozenDict(env or {}))	×
137	object.__setattr__(self, "append_only_caches", FrozenDict(append_only_caches or {}))	×
138	object.__setattr__(	×
139	self, "immutable_input_digests", FrozenDict(immutable_input_digests or {})
140	)
141
UNCOV 142	def level(self) -> LogLevel \| None:	×
143	return LogLevel.DEBUG	×
144
UNCOV 145	def message(self) -> str \| None:	×
146	return f"runnable {self.exe} for {self.description_of_origin} is ready."	×
147
UNCOV 148	def metadata(self) -> dict[str, Any] \| None:	×
149	return {	×
150	"exe": self.exe,
151	"env": self.env,
152	"append_only_caches": self.append_only_caches,
153	"description_of_origin": self.description_of_origin,
154	}
155
156
UNCOV 157	async def _resolve_post_renderers(	×
158	address_inputs: UnparsedAddressInputs,
159	) -> Iterable[RunRequest]:
160	logger.debug(	×
161	softwrap(
162	f"""
163	Resolving {pluralize(len(address_inputs.values), "post-renderer")} from {address_inputs.description_of_origin}:
164
165	{bullet_list(address_inputs.values, 5)}
166	"""
167	)
168	)
169
170	targets = await resolve_targets(**implicitly({address_inputs: UnparsedAddressInputs}))	×
171	field_sets_per_target = await find_valid_field_sets(	×
172	FieldSetsPerTargetRequest(RunFieldSet, targets),
173	**implicitly(),
174	)
175	return await concurrently(	×
176	generate_run_request(**implicitly({field_set: RunFieldSet}))
177	for field_set in field_sets_per_target.field_sets
178	)
179
180
UNCOV 181	@rule(desc="Configure Helm post-renderer", level=LogLevel.DEBUG)	×
UNCOV 182	async def setup_post_renderer_launcher(	×
183	request: SetupHelmPostRenderer,
184	post_renderer_tool: _HelmPostRendererTool,
185	cat_binary: CatBinary,
186	) -> HelmPostRenderer:
187	# Build post-renderer configuration file and create a digest containing it.
188	post_renderer_config = yaml.safe_dump(	×
189	request.replacements.to_json_dict(), explicit_start=True, sort_keys=True
190	)
191	post_renderer_cfg_digest = await create_digest(	×
192	CreateDigest(
193	[
194	FileContent(HELM_POST_RENDERER_CFG_FILENAME, post_renderer_config.encode("utf-8")),
195	]
196	)
197	)
198
199	# Generate a temporary PEX process that uses the previously created configuration file.
200	post_renderer_cfg_file = os.path.join(".", HELM_POST_RENDERER_CFG_FILENAME)	×
201	post_renderer_input_file = os.path.join(".", "__helm_stdout.yaml")	×
202	post_renderer_process = await setup_venv_pex_process(	×
203	VenvPexProcess(
204	post_renderer_tool.pex,
205	argv=[post_renderer_cfg_file, post_renderer_input_file],
206	input_digest=post_renderer_cfg_digest,
207	description="",
208	),
209	**implicitly(),
210	)
211
212	def shell_cmd(args: Iterable[str]) -> str:	×
213	return shlex.join(args)	×
214
215	# Build a shell wrapper script which will be the actual entry-point sent to Helm as the post-renderer.
216	# Extra post-renderers are plugged by piping the output of one into the next one in the order they
217	# have been defined.
218	extra_post_renderers = (	×
219	await _resolve_post_renderers(request.extra_post_renderers)
220	if request.extra_post_renderers
221	else []
222	)
223	post_renderer_process_cli = " \| ".join(	×
224	[
225	shell_cmd(post_renderer_process.argv),
226	*[shell_cmd(post_renderer.args) for post_renderer in extra_post_renderers],
227	]
228	)
229	logger.debug(	×
230	f'Using post-renderer pipeline "{post_renderer_process_cli}" in {request.description_of_origin}.'
231	)
232
233	postrenderer_wrapper_script = dedent(	×
234	f"""\
235	#!/bin/bash
236
237	# Output stdin into a file in disk
238	{cat_binary.path} <&0 > {post_renderer_input_file}
239
240	{post_renderer_process_cli}
241	"""
242	)
243	wrapper_digest = await create_digest(	×
244	CreateDigest(
245	[
246	FileContent(
247	_HELM_POST_RENDERER_WRAPPER_SCRIPT,
248	postrenderer_wrapper_script.encode("utf-8"),
249	is_executable=True,
250	),
251	]
252	),
253	)
254
255	# Combine all required settings for the internal and extra post-renderers
256	launcher_digest = await merge_digests(	×
257	MergeDigests(
258	[
259	wrapper_digest,
260	post_renderer_process.input_digest,
261	*[post_renderer.digest for post_renderer in extra_post_renderers],
262	]
263	),
264	)
265	launcher_env = {	×
266	**post_renderer_process.env,
267	**{
268	k: v
269	for post_renderer in extra_post_renderers
270	for k, v in post_renderer.extra_env.items()
271	},
272	}
273	launcher_append_only_caches = {	×
274	**post_renderer_process.append_only_caches,
275	**{
276	k: v
277	for post_renderer in extra_post_renderers
278	for k, v in (post_renderer.append_only_caches or {}).items()
279	},
280	}
281	launcher_immutable_input_digests = {	×
282	**post_renderer_process.immutable_input_digests,
283	**{
284	k: v
285	for post_renderer in extra_post_renderers
286	for k, v in (post_renderer.immutable_input_digests or {}).items()
287	},
288	}
289
290	return HelmPostRenderer(	×
291	exe=_HELM_POST_RENDERER_WRAPPER_SCRIPT,
292	digest=launcher_digest,
293	env=launcher_env,
294	append_only_caches=launcher_append_only_caches,
295	immutable_input_digests=launcher_immutable_input_digests,
296	description_of_origin=request.description_of_origin,
297	)
298
299
UNCOV 300	def rules():	×
UNCOV 301	return [	×
302	*collect_rules(),
303	*pex.rules(),
304	]

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