18917981582

Committed 29 Oct 2025 06:16PM UTC coverage: 79.819% (-0.2%) from 80.004%

Build # 18917981582

Build Type

Pull #22837

github

Committed by

web-flow

Commit Message

Merge 05341325b into 1a7da5c5e

Pull Request Pull Request #22837: Updated Treesitter dependencies

Run Details

77044 of 96523 relevant lines covered (79.82%)

3.05 hits per line

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

0.0

/src/python/pants/backend/helm/lint/trivy/rules.py

# Copyright 2024 Pants project contributors (see CONTRIBUTORS.md).
# Licensed under the Apache License, Version 2.0 (see LICENSE).
from abc import ABCMeta
from dataclasses import dataclass
from typing import Any

from pants.backend.helm.subsystems.post_renderer import setup_post_renderer_launcher
from pants.backend.helm.target_types import (
    HelmChartFieldSet,
    HelmChartTarget,
    HelmDeploymentFieldSet,
    HelmDeploymentTarget,
)
from pants.backend.helm.util_rules.post_renderer import HelmDeploymentPostRendererRequest
from pants.backend.helm.util_rules.renderer import (
    HelmDeploymentCmd,
    HelmDeploymentRequest,
    RenderedHelmFiles,
    RenderHelmChartRequest,
    render_helm_chart,
    run_renderer,
)
from pants.backend.tools.trivy.rules import RunTrivyRequest, run_trivy
from pants.backend.tools.trivy.subsystem import SkipTrivyField, Trivy
from pants.core.goals.lint import LintResult, LintTargetsRequest
from pants.core.util_rules.partitions import PartitionerType
from pants.engine.process import FallibleProcessResult
from pants.engine.rules import collect_rules, implicitly, rule
from pants.engine.target import FieldSet, Target
from pants.util.logging import LogLevel


class TrivyLintHelmRequest(LintTargetsRequest, metaclass=ABCMeta):
    tool_subsystem = Trivy  # type: ignore[assignment]


@dataclass(frozen=True)
class TrivyHelmFieldSet(FieldSet, metaclass=ABCMeta):
    @classmethod
    def opt_out(cls, tgt: Target) -> bool:
        return tgt.get(SkipTrivyField).value


@dataclass(frozen=True)
class RunTrivyOnHelmRequest:
    field_set: TrivyHelmFieldSet
    rendered_files: RenderedHelmFiles


@rule
async def run_trivy_on_helm(
    request: RunTrivyOnHelmRequest,
) -> FallibleProcessResult:
    """Run Trivy on Helm files, either a rendered Helm chart from a `helm_deployment` or a chart
    rendered from its defaults from a `helm_chart`"""

    r = await run_trivy(
        RunTrivyRequest(
            command="config",
            scanners=(),
            command_args=tuple(),
            target=".",  # the charts are rendered to the local directory
            input_digest=request.rendered_files.snapshot.digest,
            description=f"Run Trivy on Helm files for {request.field_set.address}",
        ),
        **implicitly(),
    )

    return r


@dataclass(frozen=True)
class TrivyLintHelmDeploymentFieldSet(HelmDeploymentFieldSet, TrivyHelmFieldSet):
    pass


class TrivyLintHelmDeploymentRequest(TrivyLintHelmRequest):
    field_set_type = TrivyLintHelmDeploymentFieldSet
    tool_subsystem = Trivy
    partitioner_type = PartitionerType.DEFAULT_ONE_PARTITION_PER_INPUT


@rule(desc="Lint Helm deployment with Trivy", level=LogLevel.DEBUG)
async def run_trivy_on_helm_deployment(
    request: TrivyLintHelmDeploymentRequest.Batch[TrivyLintHelmDeploymentFieldSet, Any],
) -> LintResult:
    assert len(request.elements) == 1, "not single element in partition"  # "Do we need to?"
    [field_set] = request.elements

    post_renderer = await setup_post_renderer_launcher(
        **implicitly(HelmDeploymentPostRendererRequest(field_set))
    )
    rendered_files = await run_renderer(
        **implicitly(
            HelmDeploymentRequest(
                field_set,
                cmd=HelmDeploymentCmd.RENDER,
                post_renderer=post_renderer,
                description=f"Evaluating Helm deployment files for {field_set.address}",
            )
        )
    )

    r = await run_trivy_on_helm(RunTrivyOnHelmRequest(field_set, rendered_files))

    return LintResult.create(request, r)


@dataclass(frozen=True)
class TrivyLintHelmChartFieldSet(HelmChartFieldSet, TrivyHelmFieldSet):
    pass


class TrivyLintHelmChartRequest(TrivyLintHelmRequest):
    field_set_type = TrivyLintHelmChartFieldSet
    tool_subsystem = Trivy
    partitioner_type = PartitionerType.DEFAULT_ONE_PARTITION_PER_INPUT


@rule(desc="Lint Helm chart with Trivy", level=LogLevel.DEBUG)
async def run_trivy_on_helm_chart(
    request: TrivyLintHelmChartRequest.Batch[TrivyLintHelmChartFieldSet, Any],
) -> LintResult:
    assert len(request.elements) == 1, "not single element in partition"  # "Do we need to?"
    [field_set] = request.elements

    rendered_files: RenderedHelmFiles = await render_helm_chart(RenderHelmChartRequest(field_set))
    r = await run_trivy_on_helm(RunTrivyOnHelmRequest(field_set, rendered_files))

    return LintResult.create(request, r)


def rules():
    return (
        *collect_rules(),
        *TrivyLintHelmDeploymentRequest.rules(),
        *TrivyLintHelmChartRequest.rules(),
        HelmDeploymentTarget.register_plugin_field(SkipTrivyField),
        HelmChartTarget.register_plugin_field(SkipTrivyField),
    )

1	# Copyright 2024 Pants project contributors (see CONTRIBUTORS.md).
2	# Licensed under the Apache License, Version 2.0 (see LICENSE).
3	from abc import ABCMeta	×
4	from dataclasses import dataclass	×
5	from typing import Any	×
6
7	from pants.backend.helm.subsystems.post_renderer import setup_post_renderer_launcher	×
8	from pants.backend.helm.target_types import (	×
9	HelmChartFieldSet,
10	HelmChartTarget,
11	HelmDeploymentFieldSet,
12	HelmDeploymentTarget,
13	)
14	from pants.backend.helm.util_rules.post_renderer import HelmDeploymentPostRendererRequest	×
15	from pants.backend.helm.util_rules.renderer import (	×
16	HelmDeploymentCmd,
17	HelmDeploymentRequest,
18	RenderedHelmFiles,
19	RenderHelmChartRequest,
20	render_helm_chart,
21	run_renderer,
22	)
23	from pants.backend.tools.trivy.rules import RunTrivyRequest, run_trivy	×
24	from pants.backend.tools.trivy.subsystem import SkipTrivyField, Trivy	×
25	from pants.core.goals.lint import LintResult, LintTargetsRequest	×
26	from pants.core.util_rules.partitions import PartitionerType	×
27	from pants.engine.process import FallibleProcessResult	×
28	from pants.engine.rules import collect_rules, implicitly, rule	×
29	from pants.engine.target import FieldSet, Target	×
30	from pants.util.logging import LogLevel	×
31
32
33	class TrivyLintHelmRequest(LintTargetsRequest, metaclass=ABCMeta):	×
34	tool_subsystem = Trivy # type: ignore[assignment]	×
35
36
37	@dataclass(frozen=True)	×
38	class TrivyHelmFieldSet(FieldSet, metaclass=ABCMeta):	×
39	@classmethod	×
40	def opt_out(cls, tgt: Target) -> bool:	×
41	return tgt.get(SkipTrivyField).value	×
42
43
44	@dataclass(frozen=True)	×
45	class RunTrivyOnHelmRequest:	×
46	field_set: TrivyHelmFieldSet	×
47	rendered_files: RenderedHelmFiles	×
48
49
50	@rule	×
51	async def run_trivy_on_helm(	×
52	request: RunTrivyOnHelmRequest,
53	) -> FallibleProcessResult:
54	"""Run Trivy on Helm files, either a rendered Helm chart from a `helm_deployment` or a chart
55	rendered from its defaults from a `helm_chart`"""
56
57	r = await run_trivy(	×
58	RunTrivyRequest(
59	command="config",
60	scanners=(),
61	command_args=tuple(),
62	target=".", # the charts are rendered to the local directory
63	input_digest=request.rendered_files.snapshot.digest,
64	description=f"Run Trivy on Helm files for {request.field_set.address}",
65	),
66	**implicitly(),
67	)
68
69	return r	×
70
71
72	@dataclass(frozen=True)	×
73	class TrivyLintHelmDeploymentFieldSet(HelmDeploymentFieldSet, TrivyHelmFieldSet):	×
74	pass	×
75
76
77	class TrivyLintHelmDeploymentRequest(TrivyLintHelmRequest):	×
78	field_set_type = TrivyLintHelmDeploymentFieldSet	×
79	tool_subsystem = Trivy	×
80	partitioner_type = PartitionerType.DEFAULT_ONE_PARTITION_PER_INPUT	×
81
82
83	@rule(desc="Lint Helm deployment with Trivy", level=LogLevel.DEBUG)	×
84	async def run_trivy_on_helm_deployment(	×
85	request: TrivyLintHelmDeploymentRequest.Batch[TrivyLintHelmDeploymentFieldSet, Any],
86	) -> LintResult:
87	assert len(request.elements) == 1, "not single element in partition" # "Do we need to?"	×
88	[field_set] = request.elements	×
89
90	post_renderer = await setup_post_renderer_launcher(	×
91	**implicitly(HelmDeploymentPostRendererRequest(field_set))
92	)
93	rendered_files = await run_renderer(	×
94	**implicitly(
95	HelmDeploymentRequest(
96	field_set,
97	cmd=HelmDeploymentCmd.RENDER,
98	post_renderer=post_renderer,
99	description=f"Evaluating Helm deployment files for {field_set.address}",
100	)
101	)
102	)
103
104	r = await run_trivy_on_helm(RunTrivyOnHelmRequest(field_set, rendered_files))	×
105
106	return LintResult.create(request, r)	×
107
108
109	@dataclass(frozen=True)	×
110	class TrivyLintHelmChartFieldSet(HelmChartFieldSet, TrivyHelmFieldSet):	×
111	pass	×
112
113
114	class TrivyLintHelmChartRequest(TrivyLintHelmRequest):	×
115	field_set_type = TrivyLintHelmChartFieldSet	×
116	tool_subsystem = Trivy	×
117	partitioner_type = PartitionerType.DEFAULT_ONE_PARTITION_PER_INPUT	×
118
119
120	@rule(desc="Lint Helm chart with Trivy", level=LogLevel.DEBUG)	×
121	async def run_trivy_on_helm_chart(	×
122	request: TrivyLintHelmChartRequest.Batch[TrivyLintHelmChartFieldSet, Any],
123	) -> LintResult:
124	assert len(request.elements) == 1, "not single element in partition" # "Do we need to?"	×
125	[field_set] = request.elements	×
126
127	rendered_files: RenderedHelmFiles = await render_helm_chart(RenderHelmChartRequest(field_set))	×
128	r = await run_trivy_on_helm(RunTrivyOnHelmRequest(field_set, rendered_files))	×
129
130	return LintResult.create(request, r)	×
131
132
133	def rules():	×
134	return (	×
135	*collect_rules(),
136	*TrivyLintHelmDeploymentRequest.rules(),
137	*TrivyLintHelmChartRequest.rules(),
138	HelmDeploymentTarget.register_plugin_field(SkipTrivyField),
139	HelmChartTarget.register_plugin_field(SkipTrivyField),
140	)

pantsbuild / pants / 18917981582

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