26342152999

Committed 23 May 2026 07:59PM UTC coverage: 91.165% (-1.6%) from 92.792%

Build # 26342152999

Build Type

push

github

Committed by

web-flow

Commit Message

Run Linux ARM CI on Depot runners (#23363)

RunsOn is deprecating their v2 stack, and rather than migrate
to v3 we should use the resources kindly donated by Depot.

GitHub also now has Linux ARM runners, should we need them.

Coverage Stats

87305 of 95766 relevant lines covered (91.16%)

3.87 hits per line

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

97.09

/src/python/pants/backend/helm/util_rules/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 dataclasses
import logging
import os
import re
from collections import defaultdict
from collections.abc import Iterable
from dataclasses import dataclass
from enum import Enum
from itertools import chain
from typing import Any

from pants.backend.helm.subsystems import post_renderer
from pants.backend.helm.subsystems.post_renderer import HelmPostRenderer
from pants.backend.helm.target_types import (
    HelmChartFieldSet,
    HelmDeploymentFieldSet,
    HelmDeploymentSourcesField,
)
from pants.backend.helm.util_rules import chart, tool
from pants.backend.helm.util_rules.chart import (
    FindHelmDeploymentChart,
    HelmChart,
    HelmChartRequest,
    find_chart_for_deployment,
    get_helm_chart,
)
from pants.backend.helm.util_rules.tool import HelmProcess, helm_process
from pants.core.util_rules.source_files import SourceFilesRequest, determine_source_files
from pants.engine.addresses import Address
from pants.engine.engine_aware import EngineAwareParameter, EngineAwareReturnType
from pants.engine.fs import (
    EMPTY_DIGEST,
    EMPTY_SNAPSHOT,
    CreateDigest,
    Digest,
    DigestSubset,
    Directory,
    FileContent,
    MergeDigests,
    PathGlobs,
    RemovePrefix,
    Snapshot,
)
from pants.engine.internals.native_engine import FileDigest
from pants.engine.intrinsics import create_digest, digest_to_snapshot, merge_digests
from pants.engine.process import (
    InteractiveProcess,
    ProcessCacheScope,
    ProcessResult,
    execute_process_or_raise,
)
from pants.engine.rules import collect_rules, concurrently, implicitly, rule
from pants.util.logging import LogLevel
from pants.util.strutil import pluralize, softwrap

logger = logging.getLogger(__name__)


class HelmDeploymentCmd(Enum):
    """Supported Helm rendering commands, for use when creating a `HelmDeploymentRenderer`."""

    UPGRADE = "upgrade"
    RENDER = "template"


@dataclass(frozen=True)
class HelmDeploymentRequest(EngineAwareParameter):
    field_set: HelmDeploymentFieldSet

    cmd: HelmDeploymentCmd
    description: str = dataclasses.field(compare=False)
    extra_argv: tuple[str, ...]
    post_renderer: HelmPostRenderer | None

    def __init__(
        self,
        field_set: HelmDeploymentFieldSet,
        *,
        cmd: HelmDeploymentCmd,
        description: str,
        extra_argv: Iterable[str] | None = None,
        post_renderer: HelmPostRenderer | None = None,
    ) -> None:
        object.__setattr__(self, "field_set", field_set)
        object.__setattr__(self, "cmd", cmd)
        object.__setattr__(self, "description", description)
        object.__setattr__(self, "extra_argv", tuple(extra_argv or ()))
        object.__setattr__(self, "post_renderer", post_renderer)

    def debug_hint(self) -> str | None:
        return self.field_set.address.spec

    def metadata(self) -> dict[str, Any] | None:
        return {
            "cmd": self.cmd.value,
            "address": self.field_set.address,
            "description": self.description,
            "extra_argv": self.extra_argv,
            "post_renderer": self.post_renderer,
        }


@dataclass(frozen=True)
class _HelmDeploymentProcessWrapper(EngineAwareParameter, EngineAwareReturnType):
    """Intermediate representation of a `HelmProcess` that will produce a fully rendered set of
    manifests from a given chart.

    The encapsulated `process` will be side-effecting depending on the `cmd` that was originally requested.

    This is meant to only be used internally by this module.
    """

    chart: HelmChart
    cmd: HelmDeploymentCmd
    process: HelmProcess
    address: Address
    output_directory: str | None

    @property
    def is_side_effect(self) -> bool:
        return self.cmd != HelmDeploymentCmd.RENDER

    @property
    def uses_post_renderer(self) -> bool:
        if self.output_directory:
            return False
        return True

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

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

    def message(self) -> str | None:
        msg = softwrap(
            f"""
            Built deployment process for {self.address} using chart {self.chart.address}
            with{"out" if not self.output_directory else ""} a post-renderer stage
            """
        )
        if self.output_directory:
            msg += f" and output directory: {self.output_directory}."
        else:
            msg += " and output to stdout."
        return msg

    def metadata(self) -> dict[str, Any] | None:
        meta = {
            "address": self.address.spec,
            "chart": self.chart,
            "process": self.process,
        }

        if self.output_directory:
            meta["output_directory"] = self.output_directory

        return meta


@dataclass(frozen=True)
class RenderHelmChartRequest(EngineAwareParameter):
    field_set: HelmChartFieldSet
    release_name: str | None = None

    def debug_hint(self) -> str:
        return self.field_set.address.spec


@dataclass(frozen=True)
class RenderedHelmFiles(EngineAwareReturnType):
    address: Address
    chart: HelmChart
    snapshot: Snapshot
    post_processed: bool

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

    def message(self) -> str | None:
        return softwrap(
            f"""
            Generated {pluralize(len(self.snapshot.files), "file")} from deployment {self.address}
            using chart {self.chart.address}.
            """
        )

    def artifacts(self) -> dict[str, FileDigest | Snapshot] | None:
        return {"content": self.snapshot}

    def metadata(self) -> dict[str, Any] | None:
        return {
            "address": self.address.spec,
            "chart": self.chart,
            "post_processed": self.post_processed,
        }

    def cacheable(self) -> bool:
        # When using post-renderers it may not be safe to cache the generated files as the final result
        # may contain secrets or other kind of sensitive information.
        return not self.post_processed


async def _sort_value_file_names_for_evaluation(
    address: Address,
    *,
    sources_field: HelmDeploymentSourcesField,
    value_files_snapshot: Snapshot,
    prefix: str,
) -> list[str]:
    """Sorts the list of files in `value_files_snapshot` alphabetically but grouping them in the
    order in which they have been given in the `sources_field` field glob patterns."""

    base_path = address.spec_path
    result: list[str] = []

    if not sources_field.value:
        result = list(value_files_snapshot.files)
        result.sort()
    else:
        # Break the list of filenames in subsets that follow the order given in the `sources` field
        subset_snapshots = await concurrently(
            digest_to_snapshot(
                **implicitly(
                    DigestSubset(
                        value_files_snapshot.digest,
                        PathGlobs([os.path.join(base_path, glob_pattern)]),
                    )
                )
            )
            for glob_pattern in sources_field.globs
        )
        sources_subsets = [set(snapshot.files) for snapshot in subset_snapshots]

        def minimise_and_sort_subset(input_subset: set[str]) -> list[str]:
            result: set[str] = input_subset
            for subset in sources_subsets:
                if subset == input_subset:
                    continue

                if result.issuperset(subset):
                    result = result.difference(subset)

            result_as_list = list(result)
            result_as_list.sort()
            return result_as_list

        result = list(
            chain.from_iterable([minimise_and_sort_subset(subset) for subset in sources_subsets])
        )

    logger.debug(
        softwrap(
            f"""Value files for {address} would be evaluated using the following order:

            {", ".join(result)}
            """
        )
    )

    return [os.path.join(prefix, filename) for filename in result]


@rule(desc="Prepare Helm deployment renderer")
async def setup_render_helm_deployment_process(
    request: HelmDeploymentRequest,
) -> _HelmDeploymentProcessWrapper:
    value_files_prefix = "__values"
    chart, value_files = await concurrently(
        find_chart_for_deployment(FindHelmDeploymentChart(request.field_set)),
        determine_source_files(
            SourceFilesRequest(
                sources_fields=[request.field_set.sources],
                for_sources_types=[HelmDeploymentSourcesField],
                enable_codegen=True,
            )
        ),
    )

    logger.debug(f"Using Helm chart {chart.address} in deployment {request.field_set.address}.")

    output_dir = None
    output_digest = EMPTY_DIGEST
    output_directories = None
    if not request.post_renderer:
        output_dir = "__out"
        output_digest = await create_digest(CreateDigest([Directory(output_dir)]))
        output_directories = [output_dir]

    # Sort the list of file names following a consistent ordering
    sorted_value_files = await _sort_value_file_names_for_evaluation(
        request.field_set.address,
        sources_field=request.field_set.sources,
        value_files_snapshot=value_files.snapshot,
        prefix=value_files_prefix,
    )

    # Digests to be used as an input into the renderer process.
    input_digests = [output_digest]

    # Additional process values in case a post_renderer has been requested.
    env: dict[str, str] = {}
    immutable_input_digests: dict[str, Digest] = {
        **chart.immutable_input_digests,
        value_files_prefix: value_files.snapshot.digest,
    }
    append_only_caches: dict[str, str] = {}
    if request.post_renderer:
        logger.debug(f"Using post-renderer stage in deployment {request.field_set.address}")
        input_digests.append(request.post_renderer.digest)
        env.update(request.post_renderer.env)
        immutable_input_digests.update(request.post_renderer.immutable_input_digests)
        append_only_caches.update(request.post_renderer.append_only_caches)

    merged_digests = await merge_digests(MergeDigests(input_digests))

    inline_values = request.field_set.values.value
    release_name = (
        request.field_set.release_name.value
        or request.field_set.address.target_name.replace("_", "-")
    )

    def maybe_escape_string_value(value: str) -> str:
        if re.findall("\\s+", value):
            return f'"{value}"'
        return value

    # If using a post-renderer we are only going to keep the process result cached in
    # memory to prevent storing in disk, either locally or remotely, secrets or other
    # sensitive values that may have been added in by the post-renderer.
    process_cache = (
        ProcessCacheScope.PER_RESTART_SUCCESSFUL
        if request.post_renderer
        else ProcessCacheScope.SUCCESSFUL
    )

    process = HelmProcess(
        argv=[
            request.cmd.value,
            release_name,
            chart.name,
            *(
                ("--description", f'"{request.field_set.description.value}"')
                if request.field_set.description.value
                else ()
            ),
            *(
                ("--namespace", request.field_set.namespace.value)
                if request.field_set.namespace.value
                else ()
            ),
            *(("--skip-crds",) if request.field_set.skip_crds.value else ()),
            *(("--no-hooks",) if request.field_set.no_hooks.value else ()),
            *(("--output-dir", output_dir) if output_dir else ()),
            *(("--enable-dns",) if request.field_set.enable_dns.value else ()),
            *(
                ("--post-renderer", os.path.join(".", request.post_renderer.exe))
                if request.post_renderer
                else ()
            ),
            *(("--values", ",".join(sorted_value_files)) if sorted_value_files else ()),
            *chain.from_iterable(
                (
                    ("--set", f"{key}={maybe_escape_string_value(value)}")
                    for key, value in inline_values.items()
                )
                if inline_values
                else ()
            ),
            *request.extra_argv,
        ],
        extra_env=env,
        extra_immutable_input_digests=immutable_input_digests,
        extra_append_only_caches=append_only_caches,
        description=request.description,
        level=LogLevel.DEBUG if request.cmd == HelmDeploymentCmd.RENDER else LogLevel.INFO,
        input_digest=merged_digests,
        output_directories=output_directories,
        cache_scope=process_cache,
    )

    return _HelmDeploymentProcessWrapper(
        cmd=request.cmd,
        chart=chart,
        process=process,
        address=request.field_set.address,
        output_directory=output_dir,
    )


_YAML_FILE_SEPARATOR = "---"
_HELM_OUTPUT_FILE_MARKER = "# Source: "


@rule(desc="Render Helm deployment", level=LogLevel.DEBUG)
async def run_renderer(process_wrapper: _HelmDeploymentProcessWrapper) -> RenderedHelmFiles:
    assert not process_wrapper.is_side_effect

    def file_content(file_name: str, lines: Iterable[str]) -> FileContent:
        sanitised_lines = list(lines)
        if len(sanitised_lines) == 0:
            return FileContent(file_name, b"")
        if sanitised_lines[len(sanitised_lines) - 1] == _YAML_FILE_SEPARATOR:
            sanitised_lines = sanitised_lines[:-1]
        if sanitised_lines[0] != _YAML_FILE_SEPARATOR:
            sanitised_lines = [_YAML_FILE_SEPARATOR, *sanitised_lines]

        content = "\n".join(sanitised_lines) + "\n"
        return FileContent(file_name, content.encode("utf-8"))

    def parse_renderer_output(result: ProcessResult) -> list[FileContent]:
        rendered_files_contents = result.stdout.decode("utf-8")
        rendered_files: dict[str, list[str]] = defaultdict(list)

        curr_file_name = None
        for line in rendered_files_contents.splitlines():
            if not line:
                continue

            if line.startswith(_HELM_OUTPUT_FILE_MARKER):
                curr_file_name = line[len(_HELM_OUTPUT_FILE_MARKER) :]

            if not curr_file_name:
                continue

            rendered_files[curr_file_name].append(line)

        return [file_content(file_name, lines) for file_name, lines in rendered_files.items()]

    logger.debug(f"Rendering Helm files for {process_wrapper.address}")
    result = await execute_process_or_raise(**implicitly({process_wrapper.process: HelmProcess}))

    output_snapshot = EMPTY_SNAPSHOT
    if not process_wrapper.output_directory:
        logger.debug(
            f"Parsing Helm rendered files from the process' output of {process_wrapper.address}."
        )
        output_snapshot = await digest_to_snapshot(
            **implicitly(CreateDigest(parse_renderer_output(result)))
        )
    else:
        logger.debug(
            f"Obtaining Helm rendered files from the process' output directory of {process_wrapper.address}."
        )
        output_snapshot = await digest_to_snapshot(
            **implicitly(RemovePrefix(result.output_digest, process_wrapper.output_directory))
        )

    return RenderedHelmFiles(
        address=process_wrapper.address,
        chart=process_wrapper.chart,
        snapshot=output_snapshot,
        post_processed=process_wrapper.uses_post_renderer,
    )


@rule
async def materialize_deployment_process_wrapper_into_interactive_process(
    process_wrapper: _HelmDeploymentProcessWrapper,
) -> InteractiveProcess:
    assert process_wrapper.is_side_effect

    process = await helm_process(process_wrapper.process, **implicitly())
    return InteractiveProcess.from_process(process)


@rule
async def render_helm_chart(request: RenderHelmChartRequest) -> RenderedHelmFiles:
    output_dir = "__out"
    chart, empty_output = await concurrently(
        get_helm_chart(HelmChartRequest(request.field_set), **implicitly()),
        create_digest(CreateDigest([Directory(output_dir)])),
    )

    release_name = request.release_name or request.field_set.address.target_name.replace("_", "-")

    result = await execute_process_or_raise(
        **implicitly(
            HelmProcess(
                argv=[
                    "template",
                    release_name,
                    chart.name,
                    *(
                        ("--description", f'"{request.field_set.description.value}"')
                        if request.field_set.description.value
                        else ()
                    ),
                    "--output-dir",
                    output_dir,
                ],
                description=f"Rendering chart {request.field_set.address}",
                input_digest=empty_output,
                extra_immutable_input_digests=chart.immutable_input_digests,
                output_directories=(output_dir,),
                level=LogLevel.DEBUG,
            )
        )
    )

    output_snapshot = await digest_to_snapshot(
        **implicitly(RemovePrefix(result.output_digest, output_dir))
    )
    return RenderedHelmFiles(
        address=request.field_set.address,
        chart=chart,
        snapshot=output_snapshot,
        post_processed=False,
    )


def rules():
    return [*collect_rules(), *chart.rules(), *tool.rules(), *post_renderer.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	6✔
5
6	import dataclasses	6✔
7	import logging	6✔
8	import os	6✔
9	import re	6✔
10	from collections import defaultdict	6✔
11	from collections.abc import Iterable	6✔
12	from dataclasses import dataclass	6✔
13	from enum import Enum	6✔
14	from itertools import chain	6✔
15	from typing import Any	6✔
16
17	from pants.backend.helm.subsystems import post_renderer	6✔
18	from pants.backend.helm.subsystems.post_renderer import HelmPostRenderer	6✔
19	from pants.backend.helm.target_types import (	6✔
20	HelmChartFieldSet,
21	HelmDeploymentFieldSet,
22	HelmDeploymentSourcesField,
23	)
24	from pants.backend.helm.util_rules import chart, tool	6✔
25	from pants.backend.helm.util_rules.chart import (	6✔
26	FindHelmDeploymentChart,
27	HelmChart,
28	HelmChartRequest,
29	find_chart_for_deployment,
30	get_helm_chart,
31	)
32	from pants.backend.helm.util_rules.tool import HelmProcess, helm_process	6✔
33	from pants.core.util_rules.source_files import SourceFilesRequest, determine_source_files	6✔
34	from pants.engine.addresses import Address	6✔
35	from pants.engine.engine_aware import EngineAwareParameter, EngineAwareReturnType	6✔
36	from pants.engine.fs import (	6✔
37	EMPTY_DIGEST,
38	EMPTY_SNAPSHOT,
39	CreateDigest,
40	Digest,
41	DigestSubset,
42	Directory,
43	FileContent,
44	MergeDigests,
45	PathGlobs,
46	RemovePrefix,
47	Snapshot,
48	)
49	from pants.engine.internals.native_engine import FileDigest	6✔
50	from pants.engine.intrinsics import create_digest, digest_to_snapshot, merge_digests	6✔
51	from pants.engine.process import (	6✔
52	InteractiveProcess,
53	ProcessCacheScope,
54	ProcessResult,
55	execute_process_or_raise,
56	)
57	from pants.engine.rules import collect_rules, concurrently, implicitly, rule	6✔
58	from pants.util.logging import LogLevel	6✔
59	from pants.util.strutil import pluralize, softwrap	6✔
60
61	logger = logging.getLogger(__name__)	6✔
62
63
64	class HelmDeploymentCmd(Enum):	6✔
65	"""Supported Helm rendering commands, for use when creating a `HelmDeploymentRenderer`."""
66
67	UPGRADE = "upgrade"	6✔
68	RENDER = "template"	6✔
69
70
71	@dataclass(frozen=True)	6✔
72	class HelmDeploymentRequest(EngineAwareParameter):	6✔
73	field_set: HelmDeploymentFieldSet	6✔
74
75	cmd: HelmDeploymentCmd	6✔
76	description: str = dataclasses.field(compare=False)	6✔
77	extra_argv: tuple[str, ...]	6✔
78	post_renderer: HelmPostRenderer \| None	6✔
79
80	def __init__(	6✔
81	self,
82	field_set: HelmDeploymentFieldSet,
83	*,
84	cmd: HelmDeploymentCmd,
85	description: str,
86	extra_argv: Iterable[str] \| None = None,
87	post_renderer: HelmPostRenderer \| None = None,
88	) -> None:
89	object.__setattr__(self, "field_set", field_set)	6✔
90	object.__setattr__(self, "cmd", cmd)	6✔
91	object.__setattr__(self, "description", description)	6✔
92	object.__setattr__(self, "extra_argv", tuple(extra_argv or ()))	6✔
93	object.__setattr__(self, "post_renderer", post_renderer)	6✔
94
95	def debug_hint(self) -> str \| None:	6✔
96	return self.field_set.address.spec	6✔
97
98	def metadata(self) -> dict[str, Any] \| None:	6✔
99	return {	6✔
100	"cmd": self.cmd.value,
101	"address": self.field_set.address,
102	"description": self.description,
103	"extra_argv": self.extra_argv,
104	"post_renderer": self.post_renderer,
105	}
106
107
108	@dataclass(frozen=True)	6✔
109	class _HelmDeploymentProcessWrapper(EngineAwareParameter, EngineAwareReturnType):	6✔
110	"""Intermediate representation of a `HelmProcess` that will produce a fully rendered set of
111	manifests from a given chart.
112
113	The encapsulated `process` will be side-effecting depending on the `cmd` that was originally requested.
114
115	This is meant to only be used internally by this module.
116	"""
117
118	chart: HelmChart	6✔
119	cmd: HelmDeploymentCmd	6✔
120	process: HelmProcess	6✔
121	address: Address	6✔
122	output_directory: str \| None	6✔
123
124	@property	6✔
125	def is_side_effect(self) -> bool:	6✔
126	return self.cmd != HelmDeploymentCmd.RENDER	6✔
127
128	@property	6✔
129	def uses_post_renderer(self) -> bool:	6✔
130	if self.output_directory:	6✔
131	return False	6✔
132	return True	5✔
133
134	def debug_hint(self) -> str \| None:	6✔
135	return self.address.spec	×
136
137	def level(self) -> LogLevel \| None:	6✔
138	return LogLevel.DEBUG	6✔
139
140	def message(self) -> str \| None:	6✔
141	msg = softwrap(	6✔
142	f"""
143	Built deployment process for {self.address} using chart {self.chart.address}
144	with{"out" if not self.output_directory else ""} a post-renderer stage
145	"""
146	)
147	if self.output_directory:	6✔
148	msg += f" and output directory: {self.output_directory}."	6✔
149	else:
150	msg += " and output to stdout."	5✔
151	return msg	6✔
152
153	def metadata(self) -> dict[str, Any] \| None:	6✔
154	meta = {	6✔
155	"address": self.address.spec,
156	"chart": self.chart,
157	"process": self.process,
158	}
159
160	if self.output_directory:	6✔
161	meta["output_directory"] = self.output_directory	6✔
162
163	return meta	6✔
164
165
166	@dataclass(frozen=True)	6✔
167	class RenderHelmChartRequest(EngineAwareParameter):	6✔
168	field_set: HelmChartFieldSet	6✔
169	release_name: str \| None = None	6✔
170
171	def debug_hint(self) -> str:	6✔
172	return self.field_set.address.spec	×
173
174
175	@dataclass(frozen=True)	6✔
176	class RenderedHelmFiles(EngineAwareReturnType):	6✔
177	address: Address	6✔
178	chart: HelmChart	6✔
179	snapshot: Snapshot	6✔
180	post_processed: bool	6✔
181
182	def level(self) -> LogLevel \| None:	6✔
183	return LogLevel.DEBUG	6✔
184
185	def message(self) -> str \| None:	6✔
186	return softwrap(	6✔
187	f"""
188	Generated {pluralize(len(self.snapshot.files), "file")} from deployment {self.address}
189	using chart {self.chart.address}.
190	"""
191	)
192
193	def artifacts(self) -> dict[str, FileDigest \| Snapshot] \| None:	6✔
194	return {"content": self.snapshot}	6✔
195
196	def metadata(self) -> dict[str, Any] \| None:	6✔
197	return {	6✔
198	"address": self.address.spec,
199	"chart": self.chart,
200	"post_processed": self.post_processed,
201	}
202
203	def cacheable(self) -> bool:	6✔
204	# When using post-renderers it may not be safe to cache the generated files as the final result
205	# may contain secrets or other kind of sensitive information.
206	return not self.post_processed	6✔
207
208
209	async def _sort_value_file_names_for_evaluation(	6✔
210	address: Address,
211	*,
212	sources_field: HelmDeploymentSourcesField,
213	value_files_snapshot: Snapshot,
214	prefix: str,
215	) -> list[str]:
216	"""Sorts the list of files in `value_files_snapshot` alphabetically but grouping them in the
217	order in which they have been given in the `sources_field` field glob patterns."""
218
219	base_path = address.spec_path	6✔
220	result: list[str] = []	6✔
221
222	if not sources_field.value:	6✔
223	result = list(value_files_snapshot.files)	×
224	result.sort()	×
225	else:
226	# Break the list of filenames in subsets that follow the order given in the `sources` field
227	subset_snapshots = await concurrently(	6✔
228	digest_to_snapshot(
229	**implicitly(
230	DigestSubset(
231	value_files_snapshot.digest,
232	PathGlobs([os.path.join(base_path, glob_pattern)]),
233	)
234	)
235	)
236	for glob_pattern in sources_field.globs
237	)
238	sources_subsets = [set(snapshot.files) for snapshot in subset_snapshots]	6✔
239
240	def minimise_and_sort_subset(input_subset: set[str]) -> list[str]:	6✔
241	result: set[str] = input_subset	6✔
242	for subset in sources_subsets:	6✔
243	if subset == input_subset:	6✔
244	continue	6✔
245
246	if result.issuperset(subset):	3✔
247	result = result.difference(subset)	3✔
248
249	result_as_list = list(result)	6✔
250	result_as_list.sort()	6✔
251	return result_as_list	6✔
252
253	result = list(	6✔
254	chain.from_iterable([minimise_and_sort_subset(subset) for subset in sources_subsets])
255	)
256
257	logger.debug(	6✔
258	softwrap(
259	f"""Value files for {address} would be evaluated using the following order:
260
261	{", ".join(result)}
262	"""
263	)
264	)
265
266	return [os.path.join(prefix, filename) for filename in result]	6✔
267
268
269	@rule(desc="Prepare Helm deployment renderer")	6✔
270	async def setup_render_helm_deployment_process(	6✔
271	request: HelmDeploymentRequest,
272	) -> _HelmDeploymentProcessWrapper:
273	value_files_prefix = "__values"	6✔
274	chart, value_files = await concurrently(	6✔
275	find_chart_for_deployment(FindHelmDeploymentChart(request.field_set)),
276	determine_source_files(
277	SourceFilesRequest(
278	sources_fields=[request.field_set.sources],
279	for_sources_types=[HelmDeploymentSourcesField],
280	enable_codegen=True,
281	)
282	),
283	)
284
285	logger.debug(f"Using Helm chart {chart.address} in deployment {request.field_set.address}.")	6✔
286
287	output_dir = None	6✔
288	output_digest = EMPTY_DIGEST	6✔
289	output_directories = None	6✔
290	if not request.post_renderer:	6✔
291	output_dir = "__out"	6✔
292	output_digest = await create_digest(CreateDigest([Directory(output_dir)]))	6✔
293	output_directories = [output_dir]	6✔
294
295	# Sort the list of file names following a consistent ordering
296	sorted_value_files = await _sort_value_file_names_for_evaluation(	6✔
297	request.field_set.address,
298	sources_field=request.field_set.sources,
299	value_files_snapshot=value_files.snapshot,
300	prefix=value_files_prefix,
301	)
302
303	# Digests to be used as an input into the renderer process.
304	input_digests = [output_digest]	6✔
305
306	# Additional process values in case a post_renderer has been requested.
307	env: dict[str, str] = {}	6✔
308	immutable_input_digests: dict[str, Digest] = {	6✔
309	**chart.immutable_input_digests,
310	value_files_prefix: value_files.snapshot.digest,
311	}
312	append_only_caches: dict[str, str] = {}	6✔
313	if request.post_renderer:	6✔
314	logger.debug(f"Using post-renderer stage in deployment {request.field_set.address}")	5✔
315	input_digests.append(request.post_renderer.digest)	5✔
316	env.update(request.post_renderer.env)	5✔
317	immutable_input_digests.update(request.post_renderer.immutable_input_digests)	5✔
318	append_only_caches.update(request.post_renderer.append_only_caches)	5✔
319
320	merged_digests = await merge_digests(MergeDigests(input_digests))	6✔
321
322	inline_values = request.field_set.values.value	6✔
323	release_name = (	6✔
324	request.field_set.release_name.value
325	or request.field_set.address.target_name.replace("_", "-")
326	)
327
328	def maybe_escape_string_value(value: str) -> str:	6✔
329	if re.findall("\\s+", value):	3✔
330	return f'"{value}"'	×
331	return value	3✔
332
333	# If using a post-renderer we are only going to keep the process result cached in
334	# memory to prevent storing in disk, either locally or remotely, secrets or other
335	# sensitive values that may have been added in by the post-renderer.
336	process_cache = (	6✔
337	ProcessCacheScope.PER_RESTART_SUCCESSFUL
338	if request.post_renderer
339	else ProcessCacheScope.SUCCESSFUL
340	)
341
342	process = HelmProcess(	6✔
343	argv=[
344	request.cmd.value,
345	release_name,
346	chart.name,
347	*(
348	("--description", f'"{request.field_set.description.value}"')
349	if request.field_set.description.value
350	else ()
351	),
352	*(
353	("--namespace", request.field_set.namespace.value)
354	if request.field_set.namespace.value
355	else ()
356	),
357	*(("--skip-crds",) if request.field_set.skip_crds.value else ()),
358	*(("--no-hooks",) if request.field_set.no_hooks.value else ()),
359	*(("--output-dir", output_dir) if output_dir else ()),
360	*(("--enable-dns",) if request.field_set.enable_dns.value else ()),
361	*(
362	("--post-renderer", os.path.join(".", request.post_renderer.exe))
363	if request.post_renderer
364	else ()
365	),
366	*(("--values", ",".join(sorted_value_files)) if sorted_value_files else ()),
367	*chain.from_iterable(
368	(
369	("--set", f"{key}={maybe_escape_string_value(value)}")
370	for key, value in inline_values.items()
371	)
372	if inline_values
373	else ()
374	),
375	*request.extra_argv,
376	],
377	extra_env=env,
378	extra_immutable_input_digests=immutable_input_digests,
379	extra_append_only_caches=append_only_caches,
380	description=request.description,
381	level=LogLevel.DEBUG if request.cmd == HelmDeploymentCmd.RENDER else LogLevel.INFO,
382	input_digest=merged_digests,
383	output_directories=output_directories,
384	cache_scope=process_cache,
385	)
386
387	return _HelmDeploymentProcessWrapper(	6✔
388	cmd=request.cmd,
389	chart=chart,
390	process=process,
391	address=request.field_set.address,
392	output_directory=output_dir,
393	)
394
395
396	_YAML_FILE_SEPARATOR = "---"	6✔
397	_HELM_OUTPUT_FILE_MARKER = "# Source: "	6✔
398
399
400	@rule(desc="Render Helm deployment", level=LogLevel.DEBUG)	6✔
401	async def run_renderer(process_wrapper: _HelmDeploymentProcessWrapper) -> RenderedHelmFiles:	6✔
402	assert not process_wrapper.is_side_effect	6✔
403
404	def file_content(file_name: str, lines: Iterable[str]) -> FileContent:	6✔
405	sanitised_lines = list(lines)	5✔
406	if len(sanitised_lines) == 0:	5✔
407	return FileContent(file_name, b"")	×
408	if sanitised_lines[len(sanitised_lines) - 1] == _YAML_FILE_SEPARATOR:	5✔
409	sanitised_lines = sanitised_lines[:-1]	3✔
410	if sanitised_lines[0] != _YAML_FILE_SEPARATOR:	5✔
411	sanitised_lines = [_YAML_FILE_SEPARATOR, *sanitised_lines]	5✔
412
413	content = "\n".join(sanitised_lines) + "\n"	5✔
414	return FileContent(file_name, content.encode("utf-8"))	5✔
415
416	def parse_renderer_output(result: ProcessResult) -> list[FileContent]:	6✔
417	rendered_files_contents = result.stdout.decode("utf-8")	5✔
418	rendered_files: dict[str, list[str]] = defaultdict(list)	5✔
419
420	curr_file_name = None	5✔
421	for line in rendered_files_contents.splitlines():	5✔
422	if not line:	5✔
423	continue	1✔
424
425	if line.startswith(_HELM_OUTPUT_FILE_MARKER):	5✔
426	curr_file_name = line[len(_HELM_OUTPUT_FILE_MARKER) :]	5✔
427
428	if not curr_file_name:	5✔
429	continue	5✔
430
431	rendered_files[curr_file_name].append(line)	5✔
432
433	return [file_content(file_name, lines) for file_name, lines in rendered_files.items()]	5✔
434
435	logger.debug(f"Rendering Helm files for {process_wrapper.address}")	6✔
436	result = await execute_process_or_raise(**implicitly({process_wrapper.process: HelmProcess}))	6✔
437
438	output_snapshot = EMPTY_SNAPSHOT	6✔
439	if not process_wrapper.output_directory:	6✔
440	logger.debug(	5✔
441	f"Parsing Helm rendered files from the process' output of {process_wrapper.address}."
442	)
443	output_snapshot = await digest_to_snapshot(	5✔
444	**implicitly(CreateDigest(parse_renderer_output(result)))
445	)
446	else:
447	logger.debug(	6✔
448	f"Obtaining Helm rendered files from the process' output directory of {process_wrapper.address}."
449	)
450	output_snapshot = await digest_to_snapshot(	6✔
451	**implicitly(RemovePrefix(result.output_digest, process_wrapper.output_directory))
452	)
453
454	return RenderedHelmFiles(	6✔
455	address=process_wrapper.address,
456	chart=process_wrapper.chart,
457	snapshot=output_snapshot,
458	post_processed=process_wrapper.uses_post_renderer,
459	)
460
461
462	@rule	6✔
463	async def materialize_deployment_process_wrapper_into_interactive_process(	6✔
464	process_wrapper: _HelmDeploymentProcessWrapper,
465	) -> InteractiveProcess:
466	assert process_wrapper.is_side_effect	1✔
467
468	process = await helm_process(process_wrapper.process, **implicitly())	1✔
469	return InteractiveProcess.from_process(process)	1✔
470
471
472	@rule	6✔
473	async def render_helm_chart(request: RenderHelmChartRequest) -> RenderedHelmFiles:	6✔
474	output_dir = "__out"	3✔
475	chart, empty_output = await concurrently(	3✔
476	get_helm_chart(HelmChartRequest(request.field_set), **implicitly()),
477	create_digest(CreateDigest([Directory(output_dir)])),
478	)
479
480	release_name = request.release_name or request.field_set.address.target_name.replace("_", "-")	3✔
481
482	result = await execute_process_or_raise(	3✔
483	**implicitly(
484	HelmProcess(
485	argv=[
486	"template",
487	release_name,
488	chart.name,
489	*(
490	("--description", f'"{request.field_set.description.value}"')
491	if request.field_set.description.value
492	else ()
493	),
494	"--output-dir",
495	output_dir,
496	],
497	description=f"Rendering chart {request.field_set.address}",
498	input_digest=empty_output,
499	extra_immutable_input_digests=chart.immutable_input_digests,
500	output_directories=(output_dir,),
501	level=LogLevel.DEBUG,
502	)
503	)
504	)
505
506	output_snapshot = await digest_to_snapshot(	3✔
507	**implicitly(RemovePrefix(result.output_digest, output_dir))
508	)
509	return RenderedHelmFiles(	3✔
510	address=request.field_set.address,
511	chart=chart,
512	snapshot=output_snapshot,
513	post_processed=False,
514	)
515
516
517	def rules():	6✔
518	return [collect_rules(), chart.rules(), tool.rules(), post_renderer.rules()]	6✔

pantsbuild / pants / 26342152999

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