21648319458

Committed 03 Feb 2026 09:21PM UTC coverage: 80.296% (+0.02%) from 80.278%

Build # 21648319458

Build Type

push

github

Committed by

web-flow

Commit Message

Skip Preemptive Helm (#23054)

Implements [preemptive
skipping](https://github.com/pantsbuild/pants/pull/23052) for helm

Run Details

51 of 53 new or added lines in 2 files covered. (96.23%)

1 existing line in 1 file now uncovered.

78517 of 97785 relevant lines covered (80.3%)

3.36 hits per line

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

98.98

/src/python/pants/backend/helm/goals/publish_test.py

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

from __future__ import annotations

from collections.abc import Iterable
from textwrap import dedent
from typing import cast
from unittest.mock import Mock

import pytest

from pants.backend.helm.goals import publish
from pants.backend.helm.goals.package import BuiltHelmArtifact, HelmPackageFieldSet
from pants.backend.helm.goals.publish import (
    HelmPublishFieldSet,
    PublishHelmChartRequest,
    PublishHelmChartSkipRequest,
    check_if_skip_push,
)
from pants.backend.helm.subsystems.helm import HelmSubsystem
from pants.backend.helm.target_types import HelmChartTarget
from pants.backend.helm.util_rules import tool
from pants.backend.helm.util_rules.chart import HelmChart, HelmChartRequest
from pants.backend.helm.util_rules.chart_metadata import HelmChartMetadata
from pants.backend.helm.util_rules.tool import HelmBinary
from pants.core.goals.package import BuiltPackage
from pants.core.goals.publish import (
    CheckSkipResult,
    PublishOutputData,
    PublishPackages,
    PublishProcesses,
)
from pants.core.util_rules import external_tool
from pants.engine.addresses import Address
from pants.engine.fs import EMPTY_DIGEST
from pants.engine.process import InteractiveProcess
from pants.testutil.option_util import create_subsystem
from pants.testutil.process_util import process_assertion
from pants.testutil.rule_runner import QueryRule, RuleRunner, run_rule_with_mocks


@pytest.fixture
def rule_runner() -> RuleRunner:
    return RuleRunner(
        rules=[
            *external_tool.rules(),
            *publish.rules(),
            *tool.rules(),
            QueryRule(PublishProcesses, [PublishHelmChartRequest]),
            QueryRule(HelmBinary, []),
        ],
        target_types=[HelmChartTarget],
    )


def _build(metadata: HelmChartMetadata) -> tuple[BuiltPackage, ...]:
    return (
        BuiltPackage(
            EMPTY_DIGEST, (BuiltHelmArtifact.create(f"{metadata.artifact_name}.tgz", metadata),)
        ),
    )


def _run_publish(
    rule_runner: RuleRunner,
    address: Address,
    metadata: HelmChartMetadata,
    *,
    registries: dict | None = None,
    default_repo: str | None = None,
) -> tuple[PublishProcesses, HelmBinary]:
    opts: dict[str, str] = {}
    opts.setdefault("--helm-registries", "{}")

    if registries:
        opts["--helm-registries"] = repr(registries)
    if default_repo:
        opts["--helm-default-registry-repository"] = default_repo

    rule_runner.set_options([f"{key}={value}" for key, value in opts.items()])

    target = cast(HelmChartTarget, rule_runner.get_target(address))
    field_set = HelmPublishFieldSet.create(target)
    packages = _build(metadata)

    result = rule_runner.request(PublishProcesses, [field_set._request(packages)])
    helm = rule_runner.request(HelmBinary, [])
    return result, helm


def assert_publish(
    publish: PublishPackages,
    expect_names: tuple[str, ...],
    expect_description: str | None,
    expect_process,
) -> None:
    assert publish.names == expect_names
    assert publish.description == expect_description
    if expect_process:
        assert publish.process
        assert isinstance(publish.process, InteractiveProcess)
        expect_process(publish.process.process)
    else:
        assert publish.process is None


def _declare_targets(rule_runner: RuleRunner) -> None:
    rule_runner.write_files(
        {
            "src/missing-registries/BUILD": """helm_chart()""",
            "src/skip-push/BUILD": """helm_chart(skip_push=True)""",
            "src/registries/BUILD": dedent(
                """\
                helm_chart(registries=["@internal", "oci://www.example.com/external"])
                """
            ),
            "src/repository/BUILD": dedent(
                """\
                helm_chart(registries=["oci://www.example.com/external"], repository="mycharts")
                """
            ),
        }
    )


def get_publish_output_data(target: Address, registries: Iterable[str]) -> PublishOutputData:
    return PublishOutputData(
        {
            "publisher": "helm",
            "target": target,
            "registries": tuple(registries),
        }
    )


MISSING_REGISTRIES_ADDRESS = Address("src/missing-registries")
SKIP_PUSH_ADDRESS = Address("src/skip-push")
REGISTRIES_ADDRESS = Address("src/registries")
REPOSITORY_ADDRESS = Address("src/repository")


@pytest.mark.parametrize(
    ["address", "default_registry", "expected"],
    [
        pytest.param(
            REGISTRIES_ADDRESS,
            False,
            CheckSkipResult.no_skip(),
            id="no_skip_has_registries_and_skip_push_false",
        ),
        pytest.param(
            MISSING_REGISTRIES_ADDRESS,
            False,
            CheckSkipResult.skip(
                names=["missing-registries-0.1.0"],
                description="(by missing `registries` on src/missing-registries:missing-registries)",
                data=get_publish_output_data(
                    MISSING_REGISTRIES_ADDRESS, ["<ALL DEFAULT HELM REGISTRIES>"]
                ),
            ),
            id="skip_missing_registries",
        ),
        pytest.param(
            SKIP_PUSH_ADDRESS,
            True,
            CheckSkipResult.skip(
                names=["skip-push-0.1.0"],
                description="(by `skip_push` on src/skip-push:skip-push)",
                data=get_publish_output_data(SKIP_PUSH_ADDRESS, ["<ALL DEFAULT HELM REGISTRIES>"]),
            ),
            id="skip_explicit_skip_push_true",
        ),
        pytest.param(
            MISSING_REGISTRIES_ADDRESS,
            True,
            CheckSkipResult.no_skip(),
            id="no_skip_missing_registries_and_default_repo_true",
        ),
    ],
)
def test_check_if_skip_push(
    rule_runner: RuleRunner,
    address: Address,
    default_registry: bool,
    expected: CheckSkipResult,
) -> None:
    _declare_targets(rule_runner)

    artifact_name = expected.skipped_packages[0].names[0] if expected.skipped_packages else None
    helm_subsystem = create_subsystem(
        HelmSubsystem,
        registries={"internal": {"address": "oci://www.example.com", "default": default_registry}},
    )
    tgt = rule_runner.get_target(address)
    publish_fs = HelmPublishFieldSet.create(tgt)
    package_fs = HelmPackageFieldSet.create(tgt)

    def mock_get_helm_chart(request: HelmChartRequest) -> HelmChart:
        assert request.field_set == publish_fs
        mock_helm_chart = Mock(spec=HelmChart)
        mock_info = Mock(spec=HelmChartMetadata)
        mock_info.artifact_name = artifact_name
        mock_helm_chart.info = mock_info
        return mock_helm_chart

    mock_calls = (
        {"pants.backend.helm.util_rules.chart.get_helm_chart": mock_get_helm_chart}
        if artifact_name
        else None
    )
    result = run_rule_with_mocks(
        check_if_skip_push,
        rule_args=[
            PublishHelmChartSkipRequest(publish_fs=publish_fs, package_fs=package_fs),
            helm_subsystem,
        ],
        mock_calls=mock_calls,
    )
    assert result == expected


def test_helm_push_use_default_registries(rule_runner: RuleRunner) -> None:
    _declare_targets(rule_runner)

    registries = {"internal": {"address": "oci://www.example.com", "default": True}}
    chart_metadata = HelmChartMetadata("missing-registries", "0.2.0")
    result, helm = _run_publish(
        rule_runner, Address("src/missing-registries"), chart_metadata, registries=registries
    )

    assert len(result) == 1
    assert_publish(
        result[0],
        ("oci://www.example.com/missing-registries-0.2.0",),
        None,
        process_assertion(
            argv=(helm.path, "push", "missing-registries-0.2.0.tgz", "oci://www.example.com")
        ),
    )


def test_helm_push_registries(rule_runner: RuleRunner) -> None:
    _declare_targets(rule_runner)

    registries = {"internal": {"address": "oci://www.example.com/internal"}}
    chart_metadata = HelmChartMetadata("registries", "0.1.0")
    result, helm = _run_publish(
        rule_runner,
        Address("src/registries"),
        chart_metadata,
        registries=registries,
        default_repo="charts",
    )

    assert len(result) == 2
    assert_publish(
        result[0],
        ("oci://www.example.com/internal/charts/registries-0.1.0",),
        None,
        process_assertion(
            argv=(
                helm.path,
                "push",
                "registries-0.1.0.tgz",
                "oci://www.example.com/internal/charts",
            )
        ),
    )
    assert_publish(
        result[1],
        ("oci://www.example.com/external/charts/registries-0.1.0",),
        None,
        process_assertion(
            argv=(
                helm.path,
                "push",
                "registries-0.1.0.tgz",
                "oci://www.example.com/external/charts",
            )
        ),
    )


def test_helm_push_registries_with_custom_repository(rule_runner: RuleRunner) -> None:
    _declare_targets(rule_runner)

    chart_metadata = HelmChartMetadata("repository", "0.1.0")
    result, helm = _run_publish(
        rule_runner, Address("src/repository"), chart_metadata, default_repo="default_charts"
    )
    assert len(result) == 1
    assert_publish(
        result[0],
        ("oci://www.example.com/external/mycharts/repository-0.1.0",),
        None,
        process_assertion(
            argv=(
                helm.path,
                "push",
                "repository-0.1.0.tgz",
                "oci://www.example.com/external/mycharts",
            )
        ),
    )

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	1✔
5
6	from collections.abc import Iterable	1✔
7	from textwrap import dedent	1✔
8	from typing import cast	1✔
9	from unittest.mock import Mock	1✔
10
11	import pytest	1✔
12
13	from pants.backend.helm.goals import publish	1✔
14	from pants.backend.helm.goals.package import BuiltHelmArtifact, HelmPackageFieldSet	1✔
15	from pants.backend.helm.goals.publish import (	1✔
16	HelmPublishFieldSet,
17	PublishHelmChartRequest,
18	PublishHelmChartSkipRequest,
19	check_if_skip_push,
20	)
21	from pants.backend.helm.subsystems.helm import HelmSubsystem	1✔
22	from pants.backend.helm.target_types import HelmChartTarget	1✔
23	from pants.backend.helm.util_rules import tool	1✔
24	from pants.backend.helm.util_rules.chart import HelmChart, HelmChartRequest	1✔
25	from pants.backend.helm.util_rules.chart_metadata import HelmChartMetadata	1✔
26	from pants.backend.helm.util_rules.tool import HelmBinary	1✔
27	from pants.core.goals.package import BuiltPackage	1✔
28	from pants.core.goals.publish import (	1✔
29	CheckSkipResult,
30	PublishOutputData,
31	PublishPackages,
32	PublishProcesses,
33	)
34	from pants.core.util_rules import external_tool	1✔
35	from pants.engine.addresses import Address	1✔
36	from pants.engine.fs import EMPTY_DIGEST	1✔
37	from pants.engine.process import InteractiveProcess	1✔
38	from pants.testutil.option_util import create_subsystem	1✔
39	from pants.testutil.process_util import process_assertion	1✔
40	from pants.testutil.rule_runner import QueryRule, RuleRunner, run_rule_with_mocks	1✔
41
42
43	@pytest.fixture	1✔
44	def rule_runner() -> RuleRunner:	1✔
45	return RuleRunner(	1✔
46	rules=[
47	*external_tool.rules(),
48	*publish.rules(),
49	*tool.rules(),
50	QueryRule(PublishProcesses, [PublishHelmChartRequest]),
51	QueryRule(HelmBinary, []),
52	],
53	target_types=[HelmChartTarget],
54	)
55
56
57	def _build(metadata: HelmChartMetadata) -> tuple[BuiltPackage, ...]:	1✔
58	return (	1✔
59	BuiltPackage(
60	EMPTY_DIGEST, (BuiltHelmArtifact.create(f"{metadata.artifact_name}.tgz", metadata),)
61	),
62	)
63
64
65	def _run_publish(	1✔
66	rule_runner: RuleRunner,
67	address: Address,
68	metadata: HelmChartMetadata,
69	*,
70	registries: dict \| None = None,
71	default_repo: str \| None = None,
72	) -> tuple[PublishProcesses, HelmBinary]:
73	opts: dict[str, str] = {}	1✔
74	opts.setdefault("--helm-registries", "{}")	1✔
75
76	if registries:	1✔
77	opts["--helm-registries"] = repr(registries)	1✔
78	if default_repo:	1✔
79	opts["--helm-default-registry-repository"] = default_repo	1✔
80
81	rule_runner.set_options([f"{key}={value}" for key, value in opts.items()])	1✔
82
83	target = cast(HelmChartTarget, rule_runner.get_target(address))	1✔
84	field_set = HelmPublishFieldSet.create(target)	1✔
85	packages = _build(metadata)	1✔
86
87	result = rule_runner.request(PublishProcesses, [field_set._request(packages)])	1✔
88	helm = rule_runner.request(HelmBinary, [])	1✔
89	return result, helm	1✔
90
91
92	def assert_publish(	1✔
93	publish: PublishPackages,
94	expect_names: tuple[str, ...],
95	expect_description: str \| None,
96	expect_process,
97	) -> None:
98	assert publish.names == expect_names	1✔
99	assert publish.description == expect_description	1✔
100	if expect_process:	1✔
101	assert publish.process	1✔
102	assert isinstance(publish.process, InteractiveProcess)	1✔
103	expect_process(publish.process.process)	1✔
104	else:
UNCOV 105	assert publish.process is None	×
106
107
108	def _declare_targets(rule_runner: RuleRunner) -> None:	1✔
109	rule_runner.write_files(	1✔
110	{
111	"src/missing-registries/BUILD": """helm_chart()""",
112	"src/skip-push/BUILD": """helm_chart(skip_push=True)""",
113	"src/registries/BUILD": dedent(
114	"""\
115	helm_chart(registries=["@internal", "oci://www.example.com/external"])
116	"""
117	),
118	"src/repository/BUILD": dedent(
119	"""\
120	helm_chart(registries=["oci://www.example.com/external"], repository="mycharts")
121	"""
122	),
123	}
124	)
125
126
127	def get_publish_output_data(target: Address, registries: Iterable[str]) -> PublishOutputData:	1✔
128	return PublishOutputData(	1✔
129	{
130	"publisher": "helm",
131	"target": target,
132	"registries": tuple(registries),
133	}
134	)
135
136
137	MISSING_REGISTRIES_ADDRESS = Address("src/missing-registries")	1✔
138	SKIP_PUSH_ADDRESS = Address("src/skip-push")	1✔
139	REGISTRIES_ADDRESS = Address("src/registries")	1✔
140	REPOSITORY_ADDRESS = Address("src/repository")	1✔
141
142
143	@pytest.mark.parametrize(	1✔
144	["address", "default_registry", "expected"],
145	[
146	pytest.param(
147	REGISTRIES_ADDRESS,
148	False,
149	CheckSkipResult.no_skip(),
150	id="no_skip_has_registries_and_skip_push_false",
151	),
152	pytest.param(
153	MISSING_REGISTRIES_ADDRESS,
154	False,
155	CheckSkipResult.skip(
156	names=["missing-registries-0.1.0"],
157	description="(by missing `registries` on src/missing-registries:missing-registries)",
158	data=get_publish_output_data(
159	MISSING_REGISTRIES_ADDRESS, ["<ALL DEFAULT HELM REGISTRIES>"]
160	),
161	),
162	id="skip_missing_registries",
163	),
164	pytest.param(
165	SKIP_PUSH_ADDRESS,
166	True,
167	CheckSkipResult.skip(
168	names=["skip-push-0.1.0"],
169	description="(by `skip_push` on src/skip-push:skip-push)",
170	data=get_publish_output_data(SKIP_PUSH_ADDRESS, ["<ALL DEFAULT HELM REGISTRIES>"]),
171	),
172	id="skip_explicit_skip_push_true",
173	),
174	pytest.param(
175	MISSING_REGISTRIES_ADDRESS,
176	True,
177	CheckSkipResult.no_skip(),
178	id="no_skip_missing_registries_and_default_repo_true",
179	),
180	],
181	)
182	def test_check_if_skip_push(	1✔
183	rule_runner: RuleRunner,
184	address: Address,
185	default_registry: bool,
186	expected: CheckSkipResult,
187	) -> None:
188	_declare_targets(rule_runner)	1✔
189
190	artifact_name = expected.skipped_packages[0].names[0] if expected.skipped_packages else None	1✔
191	helm_subsystem = create_subsystem(	1✔
192	HelmSubsystem,
193	registries={"internal": {"address": "oci://www.example.com", "default": default_registry}},
194	)
195	tgt = rule_runner.get_target(address)	1✔
196	publish_fs = HelmPublishFieldSet.create(tgt)	1✔
197	package_fs = HelmPackageFieldSet.create(tgt)	1✔
198
199	def mock_get_helm_chart(request: HelmChartRequest) -> HelmChart:	1✔
200	assert request.field_set == publish_fs	1✔
201	mock_helm_chart = Mock(spec=HelmChart)	1✔
202	mock_info = Mock(spec=HelmChartMetadata)	1✔
203	mock_info.artifact_name = artifact_name	1✔
204	mock_helm_chart.info = mock_info	1✔
205	return mock_helm_chart	1✔
206
207	mock_calls = (	1✔
208	{"pants.backend.helm.util_rules.chart.get_helm_chart": mock_get_helm_chart}
209	if artifact_name
210	else None
211	)
212	result = run_rule_with_mocks(	1✔
213	check_if_skip_push,
214	rule_args=[
215	PublishHelmChartSkipRequest(publish_fs=publish_fs, package_fs=package_fs),
216	helm_subsystem,
217	],
218	mock_calls=mock_calls,
219	)
220	assert result == expected	1✔
221
222
223	def test_helm_push_use_default_registries(rule_runner: RuleRunner) -> None:	1✔
224	_declare_targets(rule_runner)	1✔
225
226	registries = {"internal": {"address": "oci://www.example.com", "default": True}}	1✔
227	chart_metadata = HelmChartMetadata("missing-registries", "0.2.0")	1✔
228	result, helm = _run_publish(	1✔
229	rule_runner, Address("src/missing-registries"), chart_metadata, registries=registries
230	)
231
232	assert len(result) == 1	1✔
233	assert_publish(	1✔
234	result[0],
235	("oci://www.example.com/missing-registries-0.2.0",),
236	None,
237	process_assertion(
238	argv=(helm.path, "push", "missing-registries-0.2.0.tgz", "oci://www.example.com")
239	),
240	)
241
242
243	def test_helm_push_registries(rule_runner: RuleRunner) -> None:	1✔
244	_declare_targets(rule_runner)	1✔
245
246	registries = {"internal": {"address": "oci://www.example.com/internal"}}	1✔
247	chart_metadata = HelmChartMetadata("registries", "0.1.0")	1✔
248	result, helm = _run_publish(	1✔
249	rule_runner,
250	Address("src/registries"),
251	chart_metadata,
252	registries=registries,
253	default_repo="charts",
254	)
255
256	assert len(result) == 2	1✔
257	assert_publish(	1✔
258	result[0],
259	("oci://www.example.com/internal/charts/registries-0.1.0",),
260	None,
261	process_assertion(
262	argv=(
263	helm.path,
264	"push",
265	"registries-0.1.0.tgz",
266	"oci://www.example.com/internal/charts",
267	)
268	),
269	)
270	assert_publish(	1✔
271	result[1],
272	("oci://www.example.com/external/charts/registries-0.1.0",),
273	None,
274	process_assertion(
275	argv=(
276	helm.path,
277	"push",
278	"registries-0.1.0.tgz",
279	"oci://www.example.com/external/charts",
280	)
281	),
282	)
283
284
285	def test_helm_push_registries_with_custom_repository(rule_runner: RuleRunner) -> None:	1✔
286	_declare_targets(rule_runner)	1✔
287
288	chart_metadata = HelmChartMetadata("repository", "0.1.0")	1✔
289	result, helm = _run_publish(	1✔
290	rule_runner, Address("src/repository"), chart_metadata, default_repo="default_charts"
291	)
292	assert len(result) == 1	1✔
293	assert_publish(	1✔
294	result[0],
295	("oci://www.example.com/external/mycharts/repository-0.1.0",),
296	None,
297	process_assertion(
298	argv=(
299	helm.path,
300	"push",
301	"repository-0.1.0.tgz",
302	"oci://www.example.com/external/mycharts",
303	)
304	),
305	)

pantsbuild / pants / 21648319458

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