21374897774

Committed 26 Jan 2026 09:37PM UTC coverage: 80.008% (-0.3%) from 80.269%

Build # 21374897774

Build Type

Pull #23037

github

Committed by

web-flow

Commit Message

Merge 4023b9eee into 09b8ecaa1

Pull Request Pull Request #23037: Enable publish without package 2

Run Details

105 of 178 new or added lines in 11 files covered. (58.99%)

238 existing lines in 14 files now uncovered.

78628 of 98275 relevant lines covered (80.01%)

3.35 hits per line

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

41.46

/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

import os
from collections.abc import Callable
from typing import cast

import pytest

from pants.backend.docker.goals.package_image import BuiltDockerImage, DockerPackageFieldSet
from pants.backend.docker.goals.publish import (
    PublishDockerImageFieldSet,
    PublishDockerImageRequest,
    rules,
)
from pants.backend.docker.subsystems.docker_options import DockerOptions
from pants.backend.docker.target_types import DockerImageTarget
from pants.backend.docker.util_rules import docker_binary
from pants.backend.docker.util_rules.docker_binary import DockerBinary
from pants.core.goals.package import BuiltPackage
from pants.core.goals.publish import PublishPackages, PublishProcesses
from pants.engine.addresses import Address
from pants.engine.fs import EMPTY_DIGEST
from pants.engine.process import InteractiveProcess, Process
from pants.testutil.option_util import create_subsystem
from pants.testutil.process_util import process_assertion
from pants.testutil.rule_runner import QueryRule, RuleRunner
from pants.util.frozendict import FrozenDict
from pants.util.value_interpolation import InterpolationContext


@pytest.fixture
def rule_runner() -> RuleRunner:
    rule_runner = RuleRunner(
        rules=[
            *rules(),
            *docker_binary.rules(),
            QueryRule(PublishProcesses, [PublishDockerImageRequest]),
            QueryRule(DockerBinary, []),
        ],
        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"])""",
        }
    )
    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
) -> tuple[PublishProcesses, DockerBinary]:
    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))
    fs = PublishDockerImageFieldSet.create(tgt)
    packages = build(tgt, docker_options)
    result = rule_runner.request(PublishProcesses, [fs._request(packages)])
    docker = rule_runner.request(DockerBinary, [])
    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


def test_docker_skip_push(rule_runner: RuleRunner) -> None:
    result, _ = run_publish(rule_runner, Address("src/skip-test"))
    assert len(result) == 1
    assert_publish(
        result[0],
        ("skip-test/skip-test:latest",),
        "(by `skip_push` on src/skip-test:skip-test)",
        None,
    )


def test_docker_push_images(rule_runner: RuleRunner) -> None:
    result, docker = run_publish(rule_runner, Address("src/default"))
    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,
        Address("src/registries"),
        {
            "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,
        Address("src/registries"),
        {
            "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:
    rule_runner.set_options(
        ["--docker-env-vars=DOCKER_CONFIG"],
        env_inherit={"PATH", "PYENV_ROOT", "HOME"},
        env={"DOCKER_CONFIG": "/etc/docker/custom-config"},
    )
    result, docker = run_publish(rule_runner, Address("src/default"))
    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"}),
        ),
    )


@pytest.mark.parametrize(("target_dir", "expected"), [("default", True), ("skip-test", False)])
def test_publish_field_set_package_before_publish(
    target_dir: str, expected: bool, rule_runner: RuleRunner
) -> None:
    tgt = rule_runner.get_target(Address(os.path.join("src", target_dir)))
    fs = PublishDockerImageFieldSet.create(tgt)
    assert fs.package_before_publish(DockerPackageFieldSet.create(tgt)) is expected

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	import os	1✔
7	from collections.abc import Callable	1✔
8	from typing import cast	1✔
9
10	import pytest	1✔
11
12	from pants.backend.docker.goals.package_image import BuiltDockerImage, DockerPackageFieldSet	1✔
13	from pants.backend.docker.goals.publish import (	1✔
14	PublishDockerImageFieldSet,
15	PublishDockerImageRequest,
16	rules,
17	)
18	from pants.backend.docker.subsystems.docker_options import DockerOptions	1✔
19	from pants.backend.docker.target_types import DockerImageTarget	1✔
20	from pants.backend.docker.util_rules import docker_binary	1✔
21	from pants.backend.docker.util_rules.docker_binary import DockerBinary	1✔
22	from pants.core.goals.package import BuiltPackage	1✔
23	from pants.core.goals.publish import PublishPackages, PublishProcesses	1✔
24	from pants.engine.addresses import Address	1✔
25	from pants.engine.fs import EMPTY_DIGEST	1✔
26	from pants.engine.process import InteractiveProcess, Process	1✔
27	from pants.testutil.option_util import create_subsystem	1✔
28	from pants.testutil.process_util import process_assertion	1✔
29	from pants.testutil.rule_runner import QueryRule, RuleRunner	1✔
30	from pants.util.frozendict import FrozenDict	1✔
31	from pants.util.value_interpolation import InterpolationContext	1✔
32
33
34	@pytest.fixture	1✔
35	def rule_runner() -> RuleRunner:	1✔
36	rule_runner = RuleRunner(	1✔
37	rules=[
38	*rules(),
39	*docker_binary.rules(),
40	QueryRule(PublishProcesses, [PublishDockerImageRequest]),
41	QueryRule(DockerBinary, []),
42	],
43	target_types=[DockerImageTarget],
44	)
UNCOV 45	rule_runner.set_options(	×
46	[],
47	env_inherit={"PATH", "PYENV_ROOT", "HOME"},
48	)
UNCOV 49	rule_runner.write_files(	×
50	{
51	"src/default/BUILD": """docker_image()""",
52	"src/skip-test/BUILD": """docker_image(skip_push=True)""",
53	"src/registries/BUILD": """docker_image(registries=["@inhouse1", "@inhouse2"])""",
54	}
55	)
UNCOV 56	return rule_runner	×
57
58
59	def build(tgt: DockerImageTarget, options: DockerOptions):	1✔
UNCOV 60	fs = DockerPackageFieldSet.create(tgt)	×
UNCOV 61	image_refs = fs.image_refs(	×
62	options.default_repository,
63	options.registries(),
64	InterpolationContext(),
65	)
UNCOV 66	return (	×
67	BuiltPackage(
68	EMPTY_DIGEST,
69	(
70	BuiltDockerImage.create(
71	"sha256:made-up",
72	tuple(t.full_name for r in image_refs for t in r.tags),
73	"made-up.json",
74	),
75	),
76	),
77	)
78
79
80	def run_publish(	1✔
81	rule_runner: RuleRunner, address: Address, options: dict \| None = None
82	) -> tuple[PublishProcesses, DockerBinary]:
UNCOV 83	opts = options or {}	×
UNCOV 84	opts.setdefault("registries", {})	×
UNCOV 85	opts.setdefault("default_repository", "{directory}/{name}")	×
UNCOV 86	docker_options = create_subsystem(DockerOptions, **opts)	×
UNCOV 87	tgt = cast(DockerImageTarget, rule_runner.get_target(address))	×
UNCOV 88	fs = PublishDockerImageFieldSet.create(tgt)	×
UNCOV 89	packages = build(tgt, docker_options)	×
UNCOV 90	result = rule_runner.request(PublishProcesses, [fs._request(packages)])	×
UNCOV 91	docker = rule_runner.request(DockerBinary, [])	×
UNCOV 92	return result, docker	×
93
94
95	def assert_publish(	1✔
96	publish: PublishPackages,
97	expect_names: tuple[str, ...],
98	expect_description: str \| None,
99	expect_process: Callable[[Process], None] \| None,
100	) -> None:
UNCOV 101	assert publish.names == expect_names	×
UNCOV 102	assert publish.description == expect_description	×
UNCOV 103	if expect_process:	×
UNCOV 104	assert publish.process	×
UNCOV 105	assert isinstance(publish.process, InteractiveProcess)	×
UNCOV 106	expect_process(publish.process.process)	×
107	else:
UNCOV 108	assert publish.process is None	×
109
110
111	def test_docker_skip_push(rule_runner: RuleRunner) -> None:	1✔
UNCOV 112	result, _ = run_publish(rule_runner, Address("src/skip-test"))	×
UNCOV 113	assert len(result) == 1	×
UNCOV 114	assert_publish(	×
115	result[0],
116	("skip-test/skip-test:latest",),
117	"(by `skip_push` on src/skip-test:skip-test)",
118	None,
119	)
120
121
122	def test_docker_push_images(rule_runner: RuleRunner) -> None:	1✔
UNCOV 123	result, docker = run_publish(rule_runner, Address("src/default"))	×
UNCOV 124	assert len(result) == 1	×
UNCOV 125	assert_publish(	×
126	result[0],
127	("default/default:latest",),
128	None,
129	process_assertion(argv=(docker.path, "push", "default/default:latest")),
130	)
131
132
133	def test_docker_push_registries(rule_runner: RuleRunner) -> None:	1✔
UNCOV 134	registries = {	×
135	"inhouse1": {"address": "inhouse1.registry"},
136	"inhouse2": {"address": "inhouse2.registry"},
137	}
UNCOV 138	rule_runner.set_options(	×
139	[f"--docker-registries={registries}"],
140	env_inherit={"PATH", "PYENV_ROOT", "HOME"},
141	)
UNCOV 142	result, docker = run_publish(	×
143	rule_runner,
144	Address("src/registries"),
145	{
146	"registries": registries,
147	},
148	)
UNCOV 149	assert len(result) == 2	×
UNCOV 150	assert_publish(	×
151	result[0],
152	("inhouse1.registry/registries/registries:latest",),
153	None,
154	process_assertion(
155	argv=(
156	docker.path,
157	"push",
158	"inhouse1.registry/registries/registries:latest",
159	)
160	),
161	)
UNCOV 162	assert_publish(	×
163	result[1],
164	("inhouse2.registry/registries/registries:latest",),
165	None,
166	process_assertion(
167	argv=(
168	docker.path,
169	"push",
170	"inhouse2.registry/registries/registries:latest",
171	)
172	),
173	)
174
175
176	def test_docker_skip_push_registries(rule_runner: RuleRunner) -> None:	1✔
UNCOV 177	registries = {	×
178	"inhouse1": {"address": "inhouse1.registry"},
179	"inhouse2": {"address": "inhouse2.registry", "skip_push": True},
180	}
UNCOV 181	rule_runner.set_options(	×
182	[f"--docker-registries={registries}"],
183	env_inherit={"PATH", "PYENV_ROOT", "HOME"},
184	)
UNCOV 185	result, docker = run_publish(	×
186	rule_runner,
187	Address("src/registries"),
188	{
189	"registries": registries,
190	},
191	)
UNCOV 192	assert len(result) == 2	×
UNCOV 193	assert_publish(	×
194	result[0],
195	("inhouse1.registry/registries/registries:latest",),
196	None,
197	process_assertion(
198	argv=(
199	docker.path,
200	"push",
201	"inhouse1.registry/registries/registries:latest",
202	)
203	),
204	)
UNCOV 205	assert_publish(	×
206	result[1],
207	("inhouse2.registry/registries/registries:latest",),
208	"(by `skip_push` on registry @inhouse2)",
209	None,
210	)
211
212
213	def test_docker_push_env(rule_runner: RuleRunner) -> None:	1✔
UNCOV 214	rule_runner.set_options(	×
215	["--docker-env-vars=DOCKER_CONFIG"],
216	env_inherit={"PATH", "PYENV_ROOT", "HOME"},
217	env={"DOCKER_CONFIG": "/etc/docker/custom-config"},
218	)
UNCOV 219	result, docker = run_publish(rule_runner, Address("src/default"))	×
UNCOV 220	assert len(result) == 1	×
UNCOV 221	assert_publish(	×
222	result[0],
223	("default/default:latest",),
224	None,
225	process_assertion(
226	argv=(
227	docker.path,
228	"push",
229	"default/default:latest",
230	),
231	env=FrozenDict({"DOCKER_CONFIG": "/etc/docker/custom-config"}),
232	),
233	)
234
235
236	@pytest.mark.parametrize(("target_dir", "expected"), [("default", True), ("skip-test", False)])	1✔
237	def test_publish_field_set_package_before_publish(	1✔
238	target_dir: str, expected: bool, rule_runner: RuleRunner
239	) -> None:
NEW 240	tgt = rule_runner.get_target(Address(os.path.join("src", target_dir)))	×
NEW 241	fs = PublishDockerImageFieldSet.create(tgt)	×
NEW 242	assert fs.package_before_publish(DockerPackageFieldSet.create(tgt)) is expected	×

pantsbuild / pants / 21374897774

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