22523112068

Committed 28 Feb 2026 03:01PM UTC coverage: 90.325% (-2.6%) from 92.93%

Build # 22523112068

Build Type

push

github

Committed by

web-flow

Commit Message

Prepare 2.32.0.dev3 (#23148)

Run Details

82731 of 91593 relevant lines covered (90.32%)

3.28 hits per line

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

94.39

/src/python/pants/backend/helm/subsystems/post_renderer.py

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

from __future__ import annotations

import logging
import os
import pkgutil
import shlex
from collections.abc import Iterable, Mapping
from dataclasses import dataclass
from pathlib import PurePath
from textwrap import dedent  # noqa: PNT20
from typing import Any

import yaml

from pants.backend.helm.utils.yaml import FrozenYamlIndex
from pants.backend.python.subsystems.python_tool_base import PythonToolRequirementsBase
from pants.backend.python.target_types import EntryPoint
from pants.backend.python.util_rules import pex
from pants.backend.python.util_rules.pex import (
    VenvPex,
    VenvPexProcess,
    create_venv_pex,
    setup_venv_pex_process,
)
from pants.core.goals.run import RunFieldSet, RunRequest, generate_run_request
from pants.core.util_rules.system_binaries import CatBinary
from pants.engine.addresses import UnparsedAddressInputs
from pants.engine.engine_aware import EngineAwareParameter, EngineAwareReturnType
from pants.engine.fs import CreateDigest, Digest, FileContent
from pants.engine.internals.graph import find_valid_field_sets, resolve_targets
from pants.engine.internals.native_engine import MergeDigests
from pants.engine.intrinsics import create_digest, merge_digests
from pants.engine.rules import collect_rules, concurrently, implicitly, rule
from pants.engine.target import FieldSetsPerTargetRequest
from pants.util.frozendict import FrozenDict
from pants.util.logging import LogLevel
from pants.util.strutil import bullet_list, pluralize, softwrap

logger = logging.getLogger(__name__)

_HELM_POSTRENDERER_SOURCE = "post_renderer_main.py"
_HELM_POSTRENDERER_PACKAGE = "pants.backend.helm.subsystems"


class HelmPostRendererSubsystem(PythonToolRequirementsBase):
    options_scope = "helm-post-renderer"
    help_short = "Used perform modifications to the final output produced by Helm charts when they've been fully rendered."

    default_requirements = [
        "yamlpath>=3.6.0,<4",
        "ruamel.yaml>=0.15.96,!=0.17.0,!=0.17.1,!=0.17.2,!=0.17.5,<=0.17.21",
    ]

    register_interpreter_constraints = True

    default_lockfile_resource = (_HELM_POSTRENDERER_PACKAGE, "post_renderer.lock")


_HELM_POST_RENDERER_TOOL = "__pants_helm_post_renderer.py"


@dataclass(frozen=True)
class _HelmPostRendererTool:
    pex: VenvPex


@rule(desc="Setup Helm post renderer binaries", level=LogLevel.DEBUG)
async def setup_post_renderer_tool(
    post_renderer: HelmPostRendererSubsystem,
) -> _HelmPostRendererTool:
    post_renderer_sources = pkgutil.get_data(_HELM_POSTRENDERER_PACKAGE, _HELM_POSTRENDERER_SOURCE)
    if not post_renderer_sources:
        raise ValueError(
            f"Unable to find source to {_HELM_POSTRENDERER_SOURCE!r} in {_HELM_POSTRENDERER_PACKAGE}"
        )

    post_renderer_content = FileContent(
        path=_HELM_POST_RENDERER_TOOL, content=post_renderer_sources, is_executable=True
    )
    post_renderer_digest = await create_digest(CreateDigest([post_renderer_content]))

    post_renderer_pex = await create_venv_pex(
        **implicitly(
            post_renderer.to_pex_request(
                main=EntryPoint(PurePath(post_renderer_content.path).stem),
                sources=post_renderer_digest,
            )
        )
    )
    return _HelmPostRendererTool(post_renderer_pex)


HELM_POST_RENDERER_CFG_FILENAME = "post_renderer.cfg.yaml"
_HELM_POST_RENDERER_WRAPPER_SCRIPT = "post_renderer_wrapper.sh"


@dataclass(frozen=True)
class SetupHelmPostRenderer(EngineAwareParameter):
    """Request for a post-renderer process that will perform a series of replacements in the
    generated files."""

    replacements: FrozenYamlIndex[str]
    description_of_origin: str
    extra_post_renderers: UnparsedAddressInputs | None = None

    def debug_hint(self) -> str | None:
        return self.description_of_origin


@dataclass(frozen=True)
class HelmPostRenderer(EngineAwareReturnType):
    exe: str
    digest: Digest
    immutable_input_digests: FrozenDict[str, Digest]
    env: FrozenDict[str, str]
    append_only_caches: FrozenDict[str, str]
    description_of_origin: str

    def __init__(
        self,
        *,
        exe: str,
        digest: Digest,
        description_of_origin: str,
        env: Mapping[str, str] | None = None,
        immutable_input_digests: Mapping[str, Digest] | None = None,
        append_only_caches: Mapping[str, str] | None = None,
    ) -> None:
        object.__setattr__(self, "exe", exe)
        object.__setattr__(self, "digest", digest)
        object.__setattr__(self, "description_of_origin", description_of_origin)
        object.__setattr__(self, "env", FrozenDict(env or {}))
        object.__setattr__(self, "append_only_caches", FrozenDict(append_only_caches or {}))
        object.__setattr__(
            self, "immutable_input_digests", FrozenDict(immutable_input_digests or {})
        )

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

    def message(self) -> str | None:
        return f"runnable {self.exe} for {self.description_of_origin} is ready."

    def metadata(self) -> dict[str, Any] | None:
        return {
            "exe": self.exe,
            "env": self.env,
            "append_only_caches": self.append_only_caches,
            "description_of_origin": self.description_of_origin,
        }


async def _resolve_post_renderers(
    address_inputs: UnparsedAddressInputs,
) -> Iterable[RunRequest]:
    logger.debug(
        softwrap(
            f"""
            Resolving {pluralize(len(address_inputs.values), "post-renderer")} from {address_inputs.description_of_origin}:

            {bullet_list(address_inputs.values, 5)}
            """
        )
    )

    targets = await resolve_targets(**implicitly({address_inputs: UnparsedAddressInputs}))
    field_sets_per_target = await find_valid_field_sets(
        FieldSetsPerTargetRequest(RunFieldSet, targets),
        **implicitly(),
    )
    return await concurrently(
        generate_run_request(**implicitly({field_set: RunFieldSet}))
        for field_set in field_sets_per_target.field_sets
    )


@rule(desc="Configure Helm post-renderer", level=LogLevel.DEBUG)
async def setup_post_renderer_launcher(
    request: SetupHelmPostRenderer,
    post_renderer_tool: _HelmPostRendererTool,
    cat_binary: CatBinary,
) -> HelmPostRenderer:
    # Build post-renderer configuration file and create a digest containing it.
    post_renderer_config = yaml.safe_dump(
        request.replacements.to_json_dict(), explicit_start=True, sort_keys=True
    )
    post_renderer_cfg_digest = await create_digest(
        CreateDigest(
            [
                FileContent(HELM_POST_RENDERER_CFG_FILENAME, post_renderer_config.encode("utf-8")),
            ]
        )
    )

    # Generate a temporary PEX process that uses the previously created configuration file.
    post_renderer_cfg_file = os.path.join(".", HELM_POST_RENDERER_CFG_FILENAME)
    post_renderer_input_file = os.path.join(".", "__helm_stdout.yaml")
    post_renderer_process = await setup_venv_pex_process(
        VenvPexProcess(
            post_renderer_tool.pex,
            argv=[post_renderer_cfg_file, post_renderer_input_file],
            input_digest=post_renderer_cfg_digest,
            description="",
        ),
        **implicitly(),
    )

    def shell_cmd(args: Iterable[str]) -> str:
        return shlex.join(args)

    # Build a shell wrapper script which will be the actual entry-point sent to Helm as the post-renderer.
    # Extra post-renderers are plugged by piping the output of one into the next one in the order they
    # have been defined.
    extra_post_renderers = (
        await _resolve_post_renderers(request.extra_post_renderers)
        if request.extra_post_renderers
        else []
    )
    post_renderer_process_cli = " | ".join(
        [
            shell_cmd(post_renderer_process.argv),
            *[shell_cmd(post_renderer.args) for post_renderer in extra_post_renderers],
        ]
    )
    logger.debug(
        f'Using post-renderer pipeline "{post_renderer_process_cli}" in {request.description_of_origin}.'
    )

    postrenderer_wrapper_script = dedent(
        f"""\
        #!/bin/bash

        # Output stdin into a file in disk
        {cat_binary.path} <&0 > {post_renderer_input_file}

        {post_renderer_process_cli}
        """
    )
    wrapper_digest = await create_digest(
        CreateDigest(
            [
                FileContent(
                    _HELM_POST_RENDERER_WRAPPER_SCRIPT,
                    postrenderer_wrapper_script.encode("utf-8"),
                    is_executable=True,
                ),
            ]
        ),
    )

    # Combine all required settings for the internal and extra post-renderers
    launcher_digest = await merge_digests(
        MergeDigests(
            [
                wrapper_digest,
                post_renderer_process.input_digest,
                *[post_renderer.digest for post_renderer in extra_post_renderers],
            ]
        ),
    )
    launcher_env = {
        **post_renderer_process.env,
        **{
            k: v
            for post_renderer in extra_post_renderers
            for k, v in post_renderer.extra_env.items()
        },
    }
    launcher_append_only_caches = {
        **post_renderer_process.append_only_caches,
        **{
            k: v
            for post_renderer in extra_post_renderers
            for k, v in (post_renderer.append_only_caches or {}).items()
        },
    }
    launcher_immutable_input_digests = {
        **post_renderer_process.immutable_input_digests,
        **{
            k: v
            for post_renderer in extra_post_renderers
            for k, v in (post_renderer.immutable_input_digests or {}).items()
        },
    }

    return HelmPostRenderer(
        exe=_HELM_POST_RENDERER_WRAPPER_SCRIPT,
        digest=launcher_digest,
        env=launcher_env,
        append_only_caches=launcher_append_only_caches,
        immutable_input_digests=launcher_immutable_input_digests,
        description_of_origin=request.description_of_origin,
    )


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

pantsbuild / pants / 22523112068

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