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/goals/lint.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
from dataclasses import dataclass

from pants.backend.helm.subsystems.helm import HelmSubsystem
from pants.backend.helm.target_types import (
    HelmChartFieldSet,
    HelmChartLintQuietField,
    HelmChartLintStrictField,
    HelmSkipLintField,
)
from pants.backend.helm.util_rules import tool
from pants.backend.helm.util_rules.chart import HelmChart, HelmChartRequest, get_helm_chart
from pants.backend.helm.util_rules.tool import HelmProcess
from pants.core.goals.lint import LintResult, LintTargetsRequest, Partitions
from pants.core.util_rules.partitions import Partition
from pants.engine.intrinsics import execute_process
from pants.engine.rules import collect_rules, concurrently, implicitly, rule
from pants.util.logging import LogLevel

logger = logging.getLogger(__name__)


@dataclass(frozen=True)
class HelmLintFieldSet(HelmChartFieldSet):
    lint_strict: HelmChartLintStrictField
    lint_quiet: HelmChartLintQuietField
    skip_lint: HelmSkipLintField


class HelmLintRequest(LintTargetsRequest):
    field_set_type = HelmLintFieldSet
    tool_subsystem = HelmSubsystem  # type: ignore[assignment]


@rule
async def partition_helm_lint(
    request: HelmLintRequest.PartitionRequest[HelmLintFieldSet],
) -> Partitions[HelmLintFieldSet, HelmChart]:
    field_sets = tuple(
        field_set for field_set in request.field_sets if not field_set.skip_lint.value
    )
    charts = await concurrently(
        get_helm_chart(HelmChartRequest(field_set), **implicitly()) for field_set in field_sets
    )
    return Partitions(
        Partition((field_set,), chart) for chart, field_set in zip(charts, field_sets)
    )


@rule(desc="Lint Helm charts", level=LogLevel.DEBUG)
async def run_helm_lint(
    request: HelmLintRequest.Batch[HelmLintFieldSet, HelmChart],
    helm_subsystem: HelmSubsystem,
) -> LintResult:
    assert len(request.elements) == 1
    assert request.partition_metadata is not None

    field_set = request.elements[0]
    chart = request.partition_metadata

    argv = ["lint", chart.name]

    strict = field_set.lint_strict.value or helm_subsystem.lint_strict
    if strict:
        argv.append("--strict")
    quiet = field_set.lint_quiet.value or helm_subsystem.lint_quiet
    if quiet:
        argv.append("--quiet")

    process_result = await execute_process(
        **implicitly(
            HelmProcess(
                argv,
                extra_immutable_input_digests=chart.immutable_input_digests,
                description=f"Linting chart: {chart.info.name}",
            )
        )
    )
    return LintResult.create(request, process_result)


def rules():
    return [*collect_rules(), *tool.rules(), *HelmLintRequest.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
6	import logging	×
7	from dataclasses import dataclass	×
8
9	from pants.backend.helm.subsystems.helm import HelmSubsystem	×
10	from pants.backend.helm.target_types import (	×
11	HelmChartFieldSet,
12	HelmChartLintQuietField,
13	HelmChartLintStrictField,
14	HelmSkipLintField,
15	)
16	from pants.backend.helm.util_rules import tool	×
17	from pants.backend.helm.util_rules.chart import HelmChart, HelmChartRequest, get_helm_chart	×
18	from pants.backend.helm.util_rules.tool import HelmProcess	×
19	from pants.core.goals.lint import LintResult, LintTargetsRequest, Partitions	×
20	from pants.core.util_rules.partitions import Partition	×
21	from pants.engine.intrinsics import execute_process	×
22	from pants.engine.rules import collect_rules, concurrently, implicitly, rule	×
23	from pants.util.logging import LogLevel	×
24
25	logger = logging.getLogger(__name__)	×
26
27
28	@dataclass(frozen=True)	×
29	class HelmLintFieldSet(HelmChartFieldSet):	×
30	lint_strict: HelmChartLintStrictField	×
31	lint_quiet: HelmChartLintQuietField	×
32	skip_lint: HelmSkipLintField	×
33
34
35	class HelmLintRequest(LintTargetsRequest):	×
36	field_set_type = HelmLintFieldSet	×
37	tool_subsystem = HelmSubsystem # type: ignore[assignment]	×
38
39
40	@rule	×
41	async def partition_helm_lint(	×
42	request: HelmLintRequest.PartitionRequest[HelmLintFieldSet],
43	) -> Partitions[HelmLintFieldSet, HelmChart]:
44	field_sets = tuple(	×
45	field_set for field_set in request.field_sets if not field_set.skip_lint.value
46	)
47	charts = await concurrently(	×
48	get_helm_chart(HelmChartRequest(field_set), **implicitly()) for field_set in field_sets
49	)
50	return Partitions(	×
51	Partition((field_set,), chart) for chart, field_set in zip(charts, field_sets)
52	)
53
54
55	@rule(desc="Lint Helm charts", level=LogLevel.DEBUG)	×
56	async def run_helm_lint(	×
57	request: HelmLintRequest.Batch[HelmLintFieldSet, HelmChart],
58	helm_subsystem: HelmSubsystem,
59	) -> LintResult:
60	assert len(request.elements) == 1	×
61	assert request.partition_metadata is not None	×
62
63	field_set = request.elements[0]	×
64	chart = request.partition_metadata	×
65
66	argv = ["lint", chart.name]	×
67
68	strict = field_set.lint_strict.value or helm_subsystem.lint_strict	×
69	if strict:	×
70	argv.append("--strict")	×
71	quiet = field_set.lint_quiet.value or helm_subsystem.lint_quiet	×
72	if quiet:	×
73	argv.append("--quiet")	×
74
75	process_result = await execute_process(	×
76	**implicitly(
77	HelmProcess(
78	argv,
79	extra_immutable_input_digests=chart.immutable_input_digests,
80	description=f"Linting chart: {chart.info.name}",
81	)
82	)
83	)
84	return LintResult.create(request, process_result)	×
85
86
87	def rules():	×
88	return [collect_rules(), tool.rules(), *HelmLintRequest.rules()]	×

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