24055979590

Committed 06 Apr 2026 11:17PM UTC coverage: 52.37% (-40.5%) from 92.908%

Build # 24055979590

Build Type

Pull #23225

github

Committed by

web-flow

Commit Message

Merge 67474653c into 542ca048d

Pull Request Pull Request #23225: Add --test-show-all-batch-targets to expose all targets in batched pytest

Coverage Stats

6 of 17 new or added lines in 2 files covered. (35.29%)

23030 existing lines in 605 files now uncovered.

31643 of 60422 relevant lines covered (52.37%)

1.05 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

# pants: infer-dep(post_renderer.lock*)

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

pantsbuild / pants / 24055979590

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