23505168284

Committed 24 Mar 2026 06:11PM UTC coverage: 92.618% (-0.3%) from 92.918%

Build # 23505168284

Build Type

Pull #23133

github

Committed by

web-flow

Commit Message

Merge e8047d851 into b84b29b9b

Pull Request Pull Request #23133: Add buildctl engine

Coverage Stats

215 of 296 new or added lines in 13 files covered. (72.64%)

233 existing lines in 12 files now uncovered.

91363 of 98645 relevant lines covered (92.62%)

4.05 hits per line

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

79.65

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

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

from __future__ import annotations

from collections.abc import Callable
from typing import cast
from unittest.mock import MagicMock

import pytest

from pants.backend.docker.goals.package_image import (
    DockerImageBuildProcess,
    DockerImageRefs,
    DockerPackageFieldSet,
    GetImageRefsRequest,
    ImageRefRegistry,
    ImageRefTag,
)
from pants.backend.docker.goals.publish import (
    PublishDockerImageFieldSet,
    PublishDockerImageSkipRequest,
    check_if_skip_push,
    push_docker_images,
)
from pants.backend.docker.package_types import BuiltDockerImage
from pants.backend.docker.registries import DockerRegistryOptions
from pants.backend.docker.subsystems.docker_options import DockerOptions
from pants.backend.docker.target_types import DockerImageTarget
from pants.backend.docker.util_rules.binaries import DockerBinary
from pants.core.goals.package import BuiltPackage
from pants.core.goals.publish import (
    CheckSkipResult,
    PublishOutputData,
    PublishPackages,
    PublishProcesses,
)
from pants.core.util_rules.env_vars import rules as env_vars_rules
from pants.engine.addresses import Address
from pants.engine.env_vars import EnvironmentVars, EnvironmentVarsRequest
from pants.engine.fs import EMPTY_DIGEST
from pants.engine.process import InteractiveProcess, Process
from pants.engine.rules import QueryRule
from pants.testutil.option_util import create_subsystem
from pants.testutil.process_util import process_assertion
from pants.testutil.rule_runner import RuleRunner, run_rule_with_mocks
from pants.util.frozendict import FrozenDict
from pants.util.value_interpolation import InterpolationContext


@pytest.fixture
def rule_runner() -> RuleRunner:
    rule_runner = RuleRunner(
        rules=[*env_vars_rules(), QueryRule(EnvironmentVars, [EnvironmentVarsRequest])],
        target_types=[DockerImageTarget],
    )
    rule_runner.set_options(
        [],
        env_inherit={"PATH", "PYENV_ROOT", "HOME"},
    )
    rule_runner.write_files(
        {
            "src/default/BUILD": """docker_image()""",
            "src/skip-test/BUILD": """docker_image(skip_push=True)""",
            "src/registries/BUILD": """docker_image(registries=["@inhouse1", "@inhouse2"])""",
            "src/push-on-package/BUILD": """docker_image(output={"type": "registry"})""",
        }
    )
    return rule_runner


def build(tgt: DockerImageTarget, options: DockerOptions):
    fs = DockerPackageFieldSet.create(tgt)
    image_refs = fs.image_refs(
        options.default_repository,
        options.registries(),
        InterpolationContext(),
    )
    return (
        BuiltPackage(
            EMPTY_DIGEST,
            (
                BuiltDockerImage.create(
                    "sha256:made-up",
                    tuple(t.full_name for r in image_refs for t in r.tags),
                    "made-up.json",
                ),
            ),
        ),
    )


def run_publish(
    rule_runner: RuleRunner,
    address: Address,
    options: dict | None = None,
    env_vars: list[str] | None = None,
    mock_calls: dict | None = None,
) -> tuple[PublishProcesses, DockerBinary]:
    opts = options or {}
    opts.setdefault("registries", {})
    opts.setdefault("default_repository", "{directory}/{name}")
    opts.setdefault("publish_noninteractively", False)
    docker_options = create_subsystem(DockerOptions, **opts)
    tgt = cast(DockerImageTarget, rule_runner.get_target(address))
    fs = PublishDockerImageFieldSet.create(tgt)
    packages = build(tgt, docker_options)
    docker = DockerBinary("/dummy/docker")
    mock_env_aware = MagicMock(spec=DockerOptions.EnvironmentAware)
    if env_vars:
        mock_env_aware.env_vars = env_vars

    mock_calls = mock_calls or {
        "pants.core.util_rules.env_vars.environment_vars_subset": lambda *args: rule_runner.request(
            EnvironmentVars, args
        )
    }
    result = run_rule_with_mocks(
        push_docker_images,
        rule_args=[fs._request(packages), docker, docker_options, mock_env_aware],
        mock_calls=mock_calls,
    )
    return result, docker


def assert_publish(
    publish: PublishPackages,
    expect_names: tuple[str, ...],
    expect_description: str | None,
    expect_process: Callable[[Process], None] | None,
) -> 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


SKIP_TEST_ADDRESS = Address("src/skip-test")
REGISTRIES_ADDRESS = Address("src/registries")
DEFAULT_ADDRESS = Address("src/default")
PUSH_ON_PACKAGE_ADDRESS = Address("src/push-on-package")


@pytest.mark.parametrize(
    ["address", "options", "image_refs", "expected"],
    [
        pytest.param(
            DEFAULT_ADDRESS,
            {},
            None,
            CheckSkipResult.no_skip(),
            id="no_skip_conditions_early_exit",
        ),
        pytest.param(
            SKIP_TEST_ADDRESS,
            {},
            DockerImageRefs(
                [
                    ImageRefRegistry(
                        registry=None,
                        repository="skip-test/skip-test",
                        tags=(
                            ImageRefTag(
                                template="latest",
                                formatted="latest",
                                full_name="skip-test/skip-test:latest",
                                uses_local_alias=False,
                            ),
                        ),
                    ),
                ]
            ),
            CheckSkipResult.skip(
                names=["skip-test/skip-test:latest"],
                description="(by `skip_push` on src/skip-test:skip-test)",
                data={
                    "publisher": "docker",
                    "target": SKIP_TEST_ADDRESS,
                    "registries": ["<all default registries>"],
                },
            ),
            id="target_skip_push_true",
        ),
        pytest.param(
            REGISTRIES_ADDRESS,
            {
                "registries": {
                    "inhouse1": {"address": "inhouse1.registry", "skip_push": True},
                    "inhouse2": {"address": "inhouse2.registry", "skip_push": True},
                }
            },
            DockerImageRefs(
                [
                    ImageRefRegistry(
                        registry=DockerRegistryOptions(
                            address="inhouse1.registry",
                            alias="inhouse1",
                            skip_push=True,
                        ),
                        repository="registries/registries",
                        tags=(
                            ImageRefTag(
                                template="latest",
                                formatted="latest",
                                full_name="inhouse1.registry/registries/registries:latest",
                                uses_local_alias=False,
                            ),
                        ),
                    ),
                    ImageRefRegistry(
                        registry=DockerRegistryOptions(
                            address="inhouse2.registry",
                            alias="inhouse2",
                            skip_push=True,
                        ),
                        repository="registries/registries",
                        tags=(
                            ImageRefTag(
                                template="latest",
                                formatted="latest",
                                full_name="inhouse2.registry/registries/registries:latest",
                                uses_local_alias=False,
                            ),
                        ),
                    ),
                ]
            ),
            CheckSkipResult(
                [
                    PublishPackages(
                        names=("inhouse1.registry/registries/registries:latest",),
                        description="(by skip_push on @inhouse1)",
                        data=PublishOutputData.deep_freeze(
                            {
                                "publisher": "docker",
                                "target": REGISTRIES_ADDRESS,
                                "registries": ["@inhouse1", "@inhouse2"],
                            }
                        ),
                    ),
                    PublishPackages(
                        names=("inhouse2.registry/registries/registries:latest",),
                        description="(by skip_push on @inhouse2)",
                        data=PublishOutputData.deep_freeze(
                            {
                                "publisher": "docker",
                                "target": REGISTRIES_ADDRESS,
                                "registries": ["@inhouse1", "@inhouse2"],
                            }
                        ),
                    ),
                ]
            ),
            id="all_registries_skip_push_true",
        ),
        pytest.param(
            REGISTRIES_ADDRESS,
            {
                "registries": {
                    "inhouse1": {"address": "inhouse1.registry", "skip_push": True},
                    "inhouse2": {"address": "inhouse2.registry"},
                }
            },
            DockerImageRefs(
                [
                    ImageRefRegistry(
                        registry=DockerRegistryOptions(
                            address="inhouse1.registry",
                            alias="inhouse1",
                            skip_push=True,
                        ),
                        repository="registries/registries",
                        tags=(
                            ImageRefTag(
                                template="latest",
                                formatted="latest",
                                full_name="inhouse1.registry/registries/registries:latest",
                                uses_local_alias=False,
                            ),
                        ),
                    ),
                    ImageRefRegistry(
                        registry=DockerRegistryOptions(
                            address="inhouse2.registry",
                            alias="inhouse2",
                            skip_push=False,
                        ),
                        repository="registries/registries",
                        tags=(
                            ImageRefTag(
                                template="latest",
                                formatted="latest",
                                full_name="inhouse2.registry/registries/registries:latest",
                                uses_local_alias=False,
                            ),
                        ),
                    ),
                ]
            ),
            CheckSkipResult.no_skip(),
            id="mixed_registries_should_not_skip",
        ),
    ],
)
def test_check_if_skip_push(
    rule_runner: RuleRunner,
    address: Address,
    options: dict,
    image_refs: DockerImageRefs | None,
    expected: CheckSkipResult,
) -> None:
    opts = options or {}
    opts.setdefault("registries", {})
    opts.setdefault("default_repository", "{directory}/{name}")
    docker_options = create_subsystem(DockerOptions, **opts)
    tgt = cast(DockerImageTarget, rule_runner.get_target(address))
    package_fs = DockerPackageFieldSet.create(tgt)
    publish_fs = PublishDockerImageFieldSet.create(tgt)

    def mock_get_image_refs(request: GetImageRefsRequest) -> DockerImageRefs:
        assert request.field_set == package_fs
        assert request.build_upstream_images is False
        return cast(DockerImageRefs, image_refs)

    mock_calls = (
        {"pants.backend.docker.goals.package_image.get_image_refs": mock_get_image_refs}
        if image_refs
        else None
    )
    result = run_rule_with_mocks(
        check_if_skip_push,
        rule_args=[
            PublishDockerImageSkipRequest(publish_fs=publish_fs, package_fs=package_fs),
            docker_options,
        ],
        mock_calls=mock_calls,
    )
    assert result == expected


def test_docker_push_images(rule_runner: RuleRunner) -> None:
    result, docker = run_publish(rule_runner, DEFAULT_ADDRESS)
    assert len(result) == 1
    assert_publish(
        result[0],
        ("default/default:latest",),
        None,
        process_assertion(argv=(docker.path, "push", "default/default:latest")),
    )


def test_docker_push_registries(rule_runner: RuleRunner) -> None:
    registries = {
        "inhouse1": {"address": "inhouse1.registry"},
        "inhouse2": {"address": "inhouse2.registry"},
    }
    rule_runner.set_options(
        [f"--docker-registries={registries}"],
        env_inherit={"PATH", "PYENV_ROOT", "HOME"},
    )
    result, docker = run_publish(
        rule_runner,
        REGISTRIES_ADDRESS,
        {
            "registries": registries,
        },
    )
    assert len(result) == 2
    assert_publish(
        result[0],
        ("inhouse1.registry/registries/registries:latest",),
        None,
        process_assertion(
            argv=(
                docker.path,
                "push",
                "inhouse1.registry/registries/registries:latest",
            )
        ),
    )
    assert_publish(
        result[1],
        ("inhouse2.registry/registries/registries:latest",),
        None,
        process_assertion(
            argv=(
                docker.path,
                "push",
                "inhouse2.registry/registries/registries:latest",
            )
        ),
    )


def test_docker_skip_push_registries(rule_runner: RuleRunner) -> None:
    registries = {
        "inhouse1": {"address": "inhouse1.registry"},
        "inhouse2": {"address": "inhouse2.registry", "skip_push": True},
    }
    rule_runner.set_options(
        [f"--docker-registries={registries}"],
        env_inherit={"PATH", "PYENV_ROOT", "HOME"},
    )
    result, docker = run_publish(
        rule_runner,
        REGISTRIES_ADDRESS,
        {
            "registries": registries,
        },
    )
    assert len(result) == 2
    assert_publish(
        result[0],
        ("inhouse1.registry/registries/registries:latest",),
        None,
        process_assertion(
            argv=(
                docker.path,
                "push",
                "inhouse1.registry/registries/registries:latest",
            )
        ),
    )
    assert_publish(
        result[1],
        ("inhouse2.registry/registries/registries:latest",),
        "(by `skip_push` on registry @inhouse2)",
        None,
    )


def test_docker_push_env(rule_runner: RuleRunner) -> None:
    result, docker = run_publish(
        rule_runner, DEFAULT_ADDRESS, env_vars=["DOCKER_CONFIG=/etc/docker/custom-config"]
    )
    assert len(result) == 1
    assert_publish(
        result[0],
        ("default/default:latest",),
        None,
        process_assertion(
            argv=(
                docker.path,
                "push",
                "default/default:latest",
            ),
            env=FrozenDict({"DOCKER_CONFIG": "/etc/docker/custom-config"}),
        ),
    )


def test_docker_push_on_package(rule_runner: RuleRunner) -> None:
    """Test push_docker_images when pushes_on_package() returns True."""
    docker = DockerBinary("/dummy/docker")

    # Create mock build process that will be returned by get_docker_image_build_process
    mock_tags = ("push-on-package/push-on-package:latest",)
    expected_process = Process(
        argv=(docker.path, "build", "--output", "type=registry", "--tag", mock_tags[0], "."),
        description="Build and push docker image",
    )

    def expect_process(process: Process) -> None:
        assert process == expected_process

    def mock_get_build_process(field_set: DockerPackageFieldSet) -> DockerImageBuildProcess:
        assert field_set.address == PUSH_ON_PACKAGE_ADDRESS
        return DockerImageBuildProcess(
            process=expected_process,
            context=MagicMock(),  # Mock DockerBuildContext
            context_root=".",
            image_refs=MagicMock(),  # Mock DockerImageRefs
            tags=mock_tags,
        )

    result, docker = run_publish(
        rule_runner,
        PUSH_ON_PACKAGE_ADDRESS,
        mock_calls={
            "pants.backend.docker.goals.package_image.get_docker_image_build_process": mock_get_build_process,
        },
    )

    assert len(result) == 1
    assert_publish(
        result[0],
        mock_tags,
        None,
        expect_process,
    )

1	# Copyright 2021 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 Callable	1✔
7	from typing import cast	1✔
8	from unittest.mock import MagicMock	1✔
9
10	import pytest	1✔
11
12	from pants.backend.docker.goals.package_image import (	1✔
13	DockerImageBuildProcess,
14	DockerImageRefs,
15	DockerPackageFieldSet,
16	GetImageRefsRequest,
17	ImageRefRegistry,
18	ImageRefTag,
19	)
20	from pants.backend.docker.goals.publish import (	1✔
21	PublishDockerImageFieldSet,
22	PublishDockerImageSkipRequest,
23	check_if_skip_push,
24	push_docker_images,
25	)
26	from pants.backend.docker.package_types import BuiltDockerImage	1✔
27	from pants.backend.docker.registries import DockerRegistryOptions	1✔
28	from pants.backend.docker.subsystems.docker_options import DockerOptions	1✔
29	from pants.backend.docker.target_types import DockerImageTarget	1✔
30	from pants.backend.docker.util_rules.binaries import DockerBinary	1✔
31	from pants.core.goals.package import BuiltPackage	1✔
32	from pants.core.goals.publish import (	1✔
33	CheckSkipResult,
34	PublishOutputData,
35	PublishPackages,
36	PublishProcesses,
37	)
38	from pants.core.util_rules.env_vars import rules as env_vars_rules	1✔
39	from pants.engine.addresses import Address	1✔
40	from pants.engine.env_vars import EnvironmentVars, EnvironmentVarsRequest	1✔
41	from pants.engine.fs import EMPTY_DIGEST	1✔
42	from pants.engine.process import InteractiveProcess, Process	1✔
43	from pants.engine.rules import QueryRule	1✔
44	from pants.testutil.option_util import create_subsystem	1✔
45	from pants.testutil.process_util import process_assertion	1✔
46	from pants.testutil.rule_runner import RuleRunner, run_rule_with_mocks	1✔
47	from pants.util.frozendict import FrozenDict	1✔
48	from pants.util.value_interpolation import InterpolationContext	1✔
49
50
51	@pytest.fixture	1✔
52	def rule_runner() -> RuleRunner:	1✔
53	rule_runner = RuleRunner(	1✔
54	rules=[*env_vars_rules(), QueryRule(EnvironmentVars, [EnvironmentVarsRequest])],
55	target_types=[DockerImageTarget],
56	)
57	rule_runner.set_options(	1✔
58	[],
59	env_inherit={"PATH", "PYENV_ROOT", "HOME"},
60	)
61	rule_runner.write_files(	1✔
62	{
63	"src/default/BUILD": """docker_image()""",
64	"src/skip-test/BUILD": """docker_image(skip_push=True)""",
65	"src/registries/BUILD": """docker_image(registries=["@inhouse1", "@inhouse2"])""",
66	"src/push-on-package/BUILD": """docker_image(output={"type": "registry"})""",
67	}
68	)
69	return rule_runner	1✔
70
71
72	def build(tgt: DockerImageTarget, options: DockerOptions):	1✔
73	fs = DockerPackageFieldSet.create(tgt)	1✔
74	image_refs = fs.image_refs(	1✔
75	options.default_repository,
76	options.registries(),
77	InterpolationContext(),
78	)
79	return (	1✔
80	BuiltPackage(
81	EMPTY_DIGEST,
82	(
83	BuiltDockerImage.create(
84	"sha256:made-up",
85	tuple(t.full_name for r in image_refs for t in r.tags),
86	"made-up.json",
87	),
88	),
89	),
90	)
91
92
93	def run_publish(	1✔
94	rule_runner: RuleRunner,
95	address: Address,
96	options: dict \| None = None,
97	env_vars: list[str] \| None = None,
98	mock_calls: dict \| None = None,
99	) -> tuple[PublishProcesses, DockerBinary]:
100	opts = options or {}	1✔
101	opts.setdefault("registries", {})	1✔
102	opts.setdefault("default_repository", "{directory}/{name}")	1✔
103	opts.setdefault("publish_noninteractively", False)	1✔
104	docker_options = create_subsystem(DockerOptions, **opts)	1✔
105	tgt = cast(DockerImageTarget, rule_runner.get_target(address))	1✔
106	fs = PublishDockerImageFieldSet.create(tgt)	1✔
107	packages = build(tgt, docker_options)	1✔
108	docker = DockerBinary("/dummy/docker")	1✔
109	mock_env_aware = MagicMock(spec=DockerOptions.EnvironmentAware)	1✔
110	if env_vars:	1✔
111	mock_env_aware.env_vars = env_vars	1✔
112
113	mock_calls = mock_calls or {	1✔
114	"pants.core.util_rules.env_vars.environment_vars_subset": lambda *args: rule_runner.request(
115	EnvironmentVars, args
116	)
117	}
118	result = run_rule_with_mocks(	1✔
119	push_docker_images,
120	rule_args=[fs._request(packages), docker, docker_options, mock_env_aware],
121	mock_calls=mock_calls,
122	)
UNCOV 123	return result, docker	×
124
125
126	def assert_publish(	1✔
127	publish: PublishPackages,
128	expect_names: tuple[str, ...],
129	expect_description: str \| None,
130	expect_process: Callable[[Process], None] \| None,
131	) -> None:
UNCOV 132	assert publish.names == expect_names	×
UNCOV 133	assert publish.description == expect_description	×
UNCOV 134	if expect_process:	×
UNCOV 135	assert publish.process	×
UNCOV 136	assert isinstance(publish.process, InteractiveProcess)	×
UNCOV 137	expect_process(publish.process.process)	×
138	else:
UNCOV 139	assert publish.process is None	×
140
141
142	SKIP_TEST_ADDRESS = Address("src/skip-test")	1✔
143	REGISTRIES_ADDRESS = Address("src/registries")	1✔
144	DEFAULT_ADDRESS = Address("src/default")	1✔
145	PUSH_ON_PACKAGE_ADDRESS = Address("src/push-on-package")	1✔
146
147
148	@pytest.mark.parametrize(	1✔
149	["address", "options", "image_refs", "expected"],
150	[
151	pytest.param(
152	DEFAULT_ADDRESS,
153	{},
154	None,
155	CheckSkipResult.no_skip(),
156	id="no_skip_conditions_early_exit",
157	),
158	pytest.param(
159	SKIP_TEST_ADDRESS,
160	{},
161	DockerImageRefs(
162	[
163	ImageRefRegistry(
164	registry=None,
165	repository="skip-test/skip-test",
166	tags=(
167	ImageRefTag(
168	template="latest",
169	formatted="latest",
170	full_name="skip-test/skip-test:latest",
171	uses_local_alias=False,
172	),
173	),
174	),
175	]
176	),
177	CheckSkipResult.skip(
178	names=["skip-test/skip-test:latest"],
179	description="(by `skip_push` on src/skip-test:skip-test)",
180	data={
181	"publisher": "docker",
182	"target": SKIP_TEST_ADDRESS,
183	"registries": ["<all default registries>"],
184	},
185	),
186	id="target_skip_push_true",
187	),
188	pytest.param(
189	REGISTRIES_ADDRESS,
190	{
191	"registries": {
192	"inhouse1": {"address": "inhouse1.registry", "skip_push": True},
193	"inhouse2": {"address": "inhouse2.registry", "skip_push": True},
194	}
195	},
196	DockerImageRefs(
197	[
198	ImageRefRegistry(
199	registry=DockerRegistryOptions(
200	address="inhouse1.registry",
201	alias="inhouse1",
202	skip_push=True,
203	),
204	repository="registries/registries",
205	tags=(
206	ImageRefTag(
207	template="latest",
208	formatted="latest",
209	full_name="inhouse1.registry/registries/registries:latest",
210	uses_local_alias=False,
211	),
212	),
213	),
214	ImageRefRegistry(
215	registry=DockerRegistryOptions(
216	address="inhouse2.registry",
217	alias="inhouse2",
218	skip_push=True,
219	),
220	repository="registries/registries",
221	tags=(
222	ImageRefTag(
223	template="latest",
224	formatted="latest",
225	full_name="inhouse2.registry/registries/registries:latest",
226	uses_local_alias=False,
227	),
228	),
229	),
230	]
231	),
232	CheckSkipResult(
233	[
234	PublishPackages(
235	names=("inhouse1.registry/registries/registries:latest",),
236	description="(by skip_push on @inhouse1)",
237	data=PublishOutputData.deep_freeze(
238	{
239	"publisher": "docker",
240	"target": REGISTRIES_ADDRESS,
241	"registries": ["@inhouse1", "@inhouse2"],
242	}
243	),
244	),
245	PublishPackages(
246	names=("inhouse2.registry/registries/registries:latest",),
247	description="(by skip_push on @inhouse2)",
248	data=PublishOutputData.deep_freeze(
249	{
250	"publisher": "docker",
251	"target": REGISTRIES_ADDRESS,
252	"registries": ["@inhouse1", "@inhouse2"],
253	}
254	),
255	),
256	]
257	),
258	id="all_registries_skip_push_true",
259	),
260	pytest.param(
261	REGISTRIES_ADDRESS,
262	{
263	"registries": {
264	"inhouse1": {"address": "inhouse1.registry", "skip_push": True},
265	"inhouse2": {"address": "inhouse2.registry"},
266	}
267	},
268	DockerImageRefs(
269	[
270	ImageRefRegistry(
271	registry=DockerRegistryOptions(
272	address="inhouse1.registry",
273	alias="inhouse1",
274	skip_push=True,
275	),
276	repository="registries/registries",
277	tags=(
278	ImageRefTag(
279	template="latest",
280	formatted="latest",
281	full_name="inhouse1.registry/registries/registries:latest",
282	uses_local_alias=False,
283	),
284	),
285	),
286	ImageRefRegistry(
287	registry=DockerRegistryOptions(
288	address="inhouse2.registry",
289	alias="inhouse2",
290	skip_push=False,
291	),
292	repository="registries/registries",
293	tags=(
294	ImageRefTag(
295	template="latest",
296	formatted="latest",
297	full_name="inhouse2.registry/registries/registries:latest",
298	uses_local_alias=False,
299	),
300	),
301	),
302	]
303	),
304	CheckSkipResult.no_skip(),
305	id="mixed_registries_should_not_skip",
306	),
307	],
308	)
309	def test_check_if_skip_push(	1✔
310	rule_runner: RuleRunner,
311	address: Address,
312	options: dict,
313	image_refs: DockerImageRefs \| None,
314	expected: CheckSkipResult,
315	) -> None:
316	opts = options or {}	1✔
317	opts.setdefault("registries", {})	1✔
318	opts.setdefault("default_repository", "{directory}/{name}")	1✔
319	docker_options = create_subsystem(DockerOptions, **opts)	1✔
320	tgt = cast(DockerImageTarget, rule_runner.get_target(address))	1✔
321	package_fs = DockerPackageFieldSet.create(tgt)	1✔
322	publish_fs = PublishDockerImageFieldSet.create(tgt)	1✔
323
324	def mock_get_image_refs(request: GetImageRefsRequest) -> DockerImageRefs:	1✔
325	assert request.field_set == package_fs	1✔
326	assert request.build_upstream_images is False	1✔
327	return cast(DockerImageRefs, image_refs)	1✔
328
329	mock_calls = (	1✔
330	{"pants.backend.docker.goals.package_image.get_image_refs": mock_get_image_refs}
331	if image_refs
332	else None
333	)
334	result = run_rule_with_mocks(	1✔
335	check_if_skip_push,
336	rule_args=[
337	PublishDockerImageSkipRequest(publish_fs=publish_fs, package_fs=package_fs),
338	docker_options,
339	],
340	mock_calls=mock_calls,
341	)
342	assert result == expected	1✔
343
344
345	def test_docker_push_images(rule_runner: RuleRunner) -> None:	1✔
346	result, docker = run_publish(rule_runner, DEFAULT_ADDRESS)	1✔
UNCOV 347	assert len(result) == 1	×
UNCOV 348	assert_publish(	×
349	result[0],
350	("default/default:latest",),
351	None,
352	process_assertion(argv=(docker.path, "push", "default/default:latest")),
353	)
354
355
356	def test_docker_push_registries(rule_runner: RuleRunner) -> None:	1✔
357	registries = {	1✔
358	"inhouse1": {"address": "inhouse1.registry"},
359	"inhouse2": {"address": "inhouse2.registry"},
360	}
361	rule_runner.set_options(	1✔
362	[f"--docker-registries={registries}"],
363	env_inherit={"PATH", "PYENV_ROOT", "HOME"},
364	)
365	result, docker = run_publish(	1✔
366	rule_runner,
367	REGISTRIES_ADDRESS,
368	{
369	"registries": registries,
370	},
371	)
UNCOV 372	assert len(result) == 2	×
UNCOV 373	assert_publish(	×
374	result[0],
375	("inhouse1.registry/registries/registries:latest",),
376	None,
377	process_assertion(
378	argv=(
379	docker.path,
380	"push",
381	"inhouse1.registry/registries/registries:latest",
382	)
383	),
384	)
UNCOV 385	assert_publish(	×
386	result[1],
387	("inhouse2.registry/registries/registries:latest",),
388	None,
389	process_assertion(
390	argv=(
391	docker.path,
392	"push",
393	"inhouse2.registry/registries/registries:latest",
394	)
395	),
396	)
397
398
399	def test_docker_skip_push_registries(rule_runner: RuleRunner) -> None:	1✔
400	registries = {	1✔
401	"inhouse1": {"address": "inhouse1.registry"},
402	"inhouse2": {"address": "inhouse2.registry", "skip_push": True},
403	}
404	rule_runner.set_options(	1✔
405	[f"--docker-registries={registries}"],
406	env_inherit={"PATH", "PYENV_ROOT", "HOME"},
407	)
408	result, docker = run_publish(	1✔
409	rule_runner,
410	REGISTRIES_ADDRESS,
411	{
412	"registries": registries,
413	},
414	)
UNCOV 415	assert len(result) == 2	×
UNCOV 416	assert_publish(	×
417	result[0],
418	("inhouse1.registry/registries/registries:latest",),
419	None,
420	process_assertion(
421	argv=(
422	docker.path,
423	"push",
424	"inhouse1.registry/registries/registries:latest",
425	)
426	),
427	)
UNCOV 428	assert_publish(	×
429	result[1],
430	("inhouse2.registry/registries/registries:latest",),
431	"(by `skip_push` on registry @inhouse2)",
432	None,
433	)
434
435
436	def test_docker_push_env(rule_runner: RuleRunner) -> None:	1✔
437	result, docker = run_publish(	1✔
438	rule_runner, DEFAULT_ADDRESS, env_vars=["DOCKER_CONFIG=/etc/docker/custom-config"]
439	)
UNCOV 440	assert len(result) == 1	×
UNCOV 441	assert_publish(	×
442	result[0],
443	("default/default:latest",),
444	None,
445	process_assertion(
446	argv=(
447	docker.path,
448	"push",
449	"default/default:latest",
450	),
451	env=FrozenDict({"DOCKER_CONFIG": "/etc/docker/custom-config"}),
452	),
453	)
454
455
456	def test_docker_push_on_package(rule_runner: RuleRunner) -> None:	1✔
457	"""Test push_docker_images when pushes_on_package() returns True."""
458	docker = DockerBinary("/dummy/docker")	1✔
459
460	# Create mock build process that will be returned by get_docker_image_build_process
461	mock_tags = ("push-on-package/push-on-package:latest",)	1✔
462	expected_process = Process(	1✔
463	argv=(docker.path, "build", "--output", "type=registry", "--tag", mock_tags[0], "."),
464	description="Build and push docker image",
465	)
466
467	def expect_process(process: Process) -> None:	1✔
UNCOV 468	assert process == expected_process	×
469
470	def mock_get_build_process(field_set: DockerPackageFieldSet) -> DockerImageBuildProcess:	1✔
UNCOV 471	assert field_set.address == PUSH_ON_PACKAGE_ADDRESS	×
UNCOV 472	return DockerImageBuildProcess(	×
473	process=expected_process,
474	context=MagicMock(), # Mock DockerBuildContext
475	context_root=".",
476	image_refs=MagicMock(), # Mock DockerImageRefs
477	tags=mock_tags,
478	)
479
480	result, docker = run_publish(	1✔
481	rule_runner,
482	PUSH_ON_PACKAGE_ADDRESS,
483	mock_calls={
484	"pants.backend.docker.goals.package_image.get_docker_image_build_process": mock_get_build_process,
485	},
486	)
487
UNCOV 488	assert len(result) == 1	×
UNCOV 489	assert_publish(	×
490	result[0],
491	mock_tags,
492	None,
493	expect_process,
494	)

pantsbuild / pants / 23505168284

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