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

95.24

/src/python/pants/backend/docker/util_rules/docker_binary.py

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

from __future__ import annotations

import logging
import os
from collections.abc import Mapping, Sequence
from dataclasses import dataclass
from typing import cast

from pants.backend.docker.subsystems.docker_options import DockerOptions
from pants.backend.docker.util_rules.docker_build_args import DockerBuildArgs
from pants.core.util_rules.system_binaries import (
    BinaryPath,
    BinaryPathRequest,
    BinaryPathTest,
    BinaryShims,
    BinaryShimsRequest,
    create_binary_shims,
    find_binary,
)
from pants.engine.fs import Digest
from pants.engine.internals.selectors import concurrently
from pants.engine.process import Process, ProcessCacheScope
from pants.engine.rules import collect_rules, implicitly, rule
from pants.util.logging import LogLevel
from pants.util.strutil import pluralize

logger = logging.getLogger(__name__)


@dataclass(frozen=True)
class DockerBinary(BinaryPath):
    """The `docker` binary."""

    extra_env: Mapping[str, str]
    extra_input_digests: Mapping[str, Digest] | None

    is_podman: bool

    def __init__(
        self,
        path: str,
        fingerprint: str | None = None,
        extra_env: Mapping[str, str] | None = None,
        extra_input_digests: Mapping[str, Digest] | None = None,
        is_podman: bool = False,
    ) -> None:
        object.__setattr__(self, "extra_env", {} if extra_env is None else extra_env)
        object.__setattr__(self, "extra_input_digests", extra_input_digests)
        object.__setattr__(self, "is_podman", is_podman)
        super().__init__(path, fingerprint)

    def _get_process_environment(self, env: Mapping[str, str]) -> Mapping[str, str]:
        if not self.extra_env:
            return env

        res = {**self.extra_env, **env}

        # Merge the PATH entries, in case they are present in both `env` and `self.extra_env`.
        res["PATH"] = os.pathsep.join(
            p for p in (m.get("PATH") for m in (self.extra_env, env)) if p
        )
        return res

    def build_image(
        self,
        tags: tuple[str, ...],
        digest: Digest,
        dockerfile: str,
        build_args: DockerBuildArgs,
        context_root: str,
        env: Mapping[str, str],
        use_buildx: bool,
        extra_args: tuple[str, ...] = (),
    ) -> Process:
        if use_buildx:
            build_commands = ["buildx", "build"]
        else:
            build_commands = ["build"]

        args = [self.path, *build_commands, *extra_args]

        for tag in tags:
            args.extend(["--tag", tag])

        for build_arg in build_args:
            args.extend(["--build-arg", build_arg])

        args.extend(["--file", dockerfile])

        # Docker context root.
        args.append(context_root)

        return Process(
            argv=tuple(args),
            description=(
                f"Building docker image {tags[0]}"
                + (f" +{pluralize(len(tags) - 1, 'additional tag')}." if len(tags) > 1 else "")
            ),
            env=self._get_process_environment(env),
            input_digest=digest,
            immutable_input_digests=self.extra_input_digests,
            # We must run the docker build commands every time, even if nothing has changed,
            # in case the user ran `docker image rm` outside of Pants.
            cache_scope=ProcessCacheScope.PER_SESSION,
        )

    def push_image(self, tag: str, env: Mapping[str, str] | None = None) -> Process:
        return Process(
            argv=(self.path, "push", tag),
            cache_scope=ProcessCacheScope.PER_SESSION,
            description=f"Pushing docker image {tag}",
            env=self._get_process_environment(env or {}),
            immutable_input_digests=self.extra_input_digests,
        )

    def run_image(
        self,
        tag: str,
        *,
        docker_run_args: tuple[str, ...] | None = None,
        image_args: tuple[str, ...] | None = None,
        env: Mapping[str, str] | None = None,
    ) -> Process:
        return Process(
            argv=(self.path, "run", *(docker_run_args or []), tag, *(image_args or [])),
            cache_scope=ProcessCacheScope.PER_SESSION,
            description=f"Running docker image {tag}",
            env=self._get_process_environment(env or {}),
            immutable_input_digests=self.extra_input_digests,
        )


async def _get_docker_tools_shims(
    *,
    tools: Sequence[str],
    optional_tools: Sequence[str],
    search_path: Sequence[str],
    rationale: str,
) -> BinaryShims:
    all_binary_first_paths: list[BinaryPath] = []

    if tools:
        tools_requests = [
            BinaryPathRequest(binary_name=binary_name, search_path=search_path)
            for binary_name in tools
        ]

        tools_paths = await concurrently(
            find_binary(tools_request, **implicitly()) for tools_request in tools_requests
        )

        all_binary_first_paths.extend(
            [
                path.first_path_or_raise(request, rationale=rationale)
                for request, path in zip(tools_requests, tools_paths)
            ]
        )

    if optional_tools:
        optional_tools_requests = [
            BinaryPathRequest(binary_name=binary_name, search_path=search_path)
            for binary_name in optional_tools
        ]

        optional_tools_paths = await concurrently(
            find_binary(optional_tools_request, **implicitly())
            for optional_tools_request in optional_tools_requests
        )

        all_binary_first_paths.extend(
            [
                cast(BinaryPath, path.first_path)  # safe since we check for non-empty paths below
                for path in optional_tools_paths
                if path.paths
            ]
        )

    tools_shims = await create_binary_shims(
        BinaryShimsRequest.for_paths(
            *all_binary_first_paths,
            rationale=rationale,
        ),
        **implicitly(),
    )

    return tools_shims


@rule(desc="Finding the `docker` binary and related tooling", level=LogLevel.DEBUG)
async def get_docker(
    docker_options: DockerOptions, docker_options_env_aware: DockerOptions.EnvironmentAware
) -> DockerBinary:
    search_path = docker_options_env_aware.executable_search_path

    first_path: BinaryPath | None = None
    is_podman = False

    if getattr(docker_options.options, "experimental_enable_podman", False):
        # Enable podman support with `pants.backend.experimental.docker.podman`
        request = BinaryPathRequest(
            binary_name="podman",
            search_path=search_path,
            test=BinaryPathTest(args=["-v"]),
        )
        paths = await find_binary(request, **implicitly())
        first_path = paths.first_path
        if first_path:
            is_podman = True
            logger.warning("podman found. Podman support is experimental.")

    if not first_path:
        request = BinaryPathRequest(
            binary_name="docker",
            search_path=search_path,
            test=BinaryPathTest(args=["-v"]),
        )
        paths = await find_binary(request, **implicitly())
        first_path = paths.first_path_or_raise(request, rationale="interact with the docker daemon")

    if not docker_options.tools and not docker_options.optional_tools:
        return DockerBinary(first_path.path, first_path.fingerprint, is_podman=is_podman)

    tools_shims = await _get_docker_tools_shims(
        tools=docker_options.tools,
        optional_tools=docker_options.optional_tools,
        search_path=search_path,
        rationale="use docker",
    )

    extra_env = {"PATH": tools_shims.path_component}
    extra_input_digests = tools_shims.immutable_input_digests

    return DockerBinary(
        first_path.path,
        first_path.fingerprint,
        extra_env=extra_env,
        extra_input_digests=extra_input_digests,
        is_podman=is_podman,
    )


def rules():
    return collect_rules()

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	9✔
5
6	import logging	9✔
7	import os	9✔
8	from collections.abc import Mapping, Sequence	9✔
9	from dataclasses import dataclass	9✔
10	from typing import cast	9✔
11
12	from pants.backend.docker.subsystems.docker_options import DockerOptions	9✔
13	from pants.backend.docker.util_rules.docker_build_args import DockerBuildArgs	9✔
14	from pants.core.util_rules.system_binaries import (	9✔
15	BinaryPath,
16	BinaryPathRequest,
17	BinaryPathTest,
18	BinaryShims,
19	BinaryShimsRequest,
20	create_binary_shims,
21	find_binary,
22	)
23	from pants.engine.fs import Digest	9✔
24	from pants.engine.internals.selectors import concurrently	9✔
25	from pants.engine.process import Process, ProcessCacheScope	9✔
26	from pants.engine.rules import collect_rules, implicitly, rule	9✔
27	from pants.util.logging import LogLevel	9✔
28	from pants.util.strutil import pluralize	9✔
29
30	logger = logging.getLogger(__name__)	9✔
31
32
33	@dataclass(frozen=True)	9✔
34	class DockerBinary(BinaryPath):	9✔
35	"""The `docker` binary."""
36
37	extra_env: Mapping[str, str]	9✔
38	extra_input_digests: Mapping[str, Digest] \| None	9✔
39
40	is_podman: bool	9✔
41
42	def __init__(	9✔
43	self,
44	path: str,
45	fingerprint: str \| None = None,
46	extra_env: Mapping[str, str] \| None = None,
47	extra_input_digests: Mapping[str, Digest] \| None = None,
48	is_podman: bool = False,
49	) -> None:
50	object.__setattr__(self, "extra_env", {} if extra_env is None else extra_env)	1✔
51	object.__setattr__(self, "extra_input_digests", extra_input_digests)	1✔
52	object.__setattr__(self, "is_podman", is_podman)	1✔
53	super().__init__(path, fingerprint)	1✔
54
55	def _get_process_environment(self, env: Mapping[str, str]) -> Mapping[str, str]:	9✔
56	if not self.extra_env:	1✔
57	return env	1✔
58
59	res = {self.extra_env, env}	×
60
61	# Merge the PATH entries, in case they are present in both `env` and `self.extra_env`.
62	res["PATH"] = os.pathsep.join(	×
63	p for p in (m.get("PATH") for m in (self.extra_env, env)) if p
64	)
65	return res	×
66
67	def build_image(	9✔
68	self,
69	tags: tuple[str, ...],
70	digest: Digest,
71	dockerfile: str,
72	build_args: DockerBuildArgs,
73	context_root: str,
74	env: Mapping[str, str],
75	use_buildx: bool,
76	extra_args: tuple[str, ...] = (),
77	) -> Process:
78	if use_buildx:	1✔
UNCOV 79	build_commands = ["buildx", "build"]	×
80	else:
81	build_commands = ["build"]	1✔
82
83	args = [self.path, build_commands, extra_args]	1✔
84
85	for tag in tags:	1✔
86	args.extend(["--tag", tag])	1✔
87
88	for build_arg in build_args:	1✔
89	args.extend(["--build-arg", build_arg])	1✔
90
91	args.extend(["--file", dockerfile])	1✔
92
93	# Docker context root.
94	args.append(context_root)	1✔
95
96	return Process(	1✔
97	argv=tuple(args),
98	description=(
99	f"Building docker image {tags[0]}"
100	+ (f" +{pluralize(len(tags) - 1, 'additional tag')}." if len(tags) > 1 else "")
101	),
102	env=self._get_process_environment(env),
103	input_digest=digest,
104	immutable_input_digests=self.extra_input_digests,
105	# We must run the docker build commands every time, even if nothing has changed,
106	# in case the user ran `docker image rm` outside of Pants.
107	cache_scope=ProcessCacheScope.PER_SESSION,
108	)
109
110	def push_image(self, tag: str, env: Mapping[str, str] \| None = None) -> Process:	9✔
111	return Process(	1✔
112	argv=(self.path, "push", tag),
113	cache_scope=ProcessCacheScope.PER_SESSION,
114	description=f"Pushing docker image {tag}",
115	env=self._get_process_environment(env or {}),
116	immutable_input_digests=self.extra_input_digests,
117	)
118
119	def run_image(	9✔
120	self,
121	tag: str,
122	*,
123	docker_run_args: tuple[str, ...] \| None = None,
124	image_args: tuple[str, ...] \| None = None,
125	env: Mapping[str, str] \| None = None,
126	) -> Process:
127	return Process(	1✔
128	argv=(self.path, "run", (docker_run_args or []), tag, (image_args or [])),
129	cache_scope=ProcessCacheScope.PER_SESSION,
130	description=f"Running docker image {tag}",
131	env=self._get_process_environment(env or {}),
132	immutable_input_digests=self.extra_input_digests,
133	)
134
135
136	async def _get_docker_tools_shims(	9✔
137	*,
138	tools: Sequence[str],
139	optional_tools: Sequence[str],
140	search_path: Sequence[str],
141	rationale: str,
142	) -> BinaryShims:
143	all_binary_first_paths: list[BinaryPath] = []	1✔
144
145	if tools:	1✔
146	tools_requests = [	1✔
147	BinaryPathRequest(binary_name=binary_name, search_path=search_path)
148	for binary_name in tools
149	]
150
151	tools_paths = await concurrently(	1✔
152	find_binary(tools_request, **implicitly()) for tools_request in tools_requests
153	)
154
155	all_binary_first_paths.extend(	1✔
156	[
157	path.first_path_or_raise(request, rationale=rationale)
158	for request, path in zip(tools_requests, tools_paths)
159	]
160	)
161
162	if optional_tools:	1✔
163	optional_tools_requests = [	1✔
164	BinaryPathRequest(binary_name=binary_name, search_path=search_path)
165	for binary_name in optional_tools
166	]
167
168	optional_tools_paths = await concurrently(	1✔
169	find_binary(optional_tools_request, **implicitly())
170	for optional_tools_request in optional_tools_requests
171	)
172
173	all_binary_first_paths.extend(	1✔
174	[
175	cast(BinaryPath, path.first_path) # safe since we check for non-empty paths below
176	for path in optional_tools_paths
177	if path.paths
178	]
179	)
180
181	tools_shims = await create_binary_shims(	1✔
182	BinaryShimsRequest.for_paths(
183	*all_binary_first_paths,
184	rationale=rationale,
185	),
186	**implicitly(),
187	)
188
189	return tools_shims	1✔
190
191
192	@rule(desc="Finding the `docker` binary and related tooling", level=LogLevel.DEBUG)	9✔
193	async def get_docker(	9✔
194	docker_options: DockerOptions, docker_options_env_aware: DockerOptions.EnvironmentAware
195	) -> DockerBinary:
196	search_path = docker_options_env_aware.executable_search_path	1✔
197
198	first_path: BinaryPath \| None = None	1✔
199	is_podman = False	1✔
200
201	if getattr(docker_options.options, "experimental_enable_podman", False):	1✔
202	# Enable podman support with `pants.backend.experimental.docker.podman`
203	request = BinaryPathRequest(	1✔
204	binary_name="podman",
205	search_path=search_path,
206	test=BinaryPathTest(args=["-v"]),
207	)
208	paths = await find_binary(request, **implicitly())	1✔
209	first_path = paths.first_path	1✔
210	if first_path:	1✔
211	is_podman = True	1✔
212	logger.warning("podman found. Podman support is experimental.")	1✔
213
214	if not first_path:	1✔
215	request = BinaryPathRequest(	1✔
216	binary_name="docker",
217	search_path=search_path,
218	test=BinaryPathTest(args=["-v"]),
219	)
220	paths = await find_binary(request, **implicitly())	1✔
221	first_path = paths.first_path_or_raise(request, rationale="interact with the docker daemon")	1✔
222
223	if not docker_options.tools and not docker_options.optional_tools:	1✔
224	return DockerBinary(first_path.path, first_path.fingerprint, is_podman=is_podman)	1✔
225
226	tools_shims = await _get_docker_tools_shims(	1✔
227	tools=docker_options.tools,
228	optional_tools=docker_options.optional_tools,
229	search_path=search_path,
230	rationale="use docker",
231	)
232
233	extra_env = {"PATH": tools_shims.path_component}	1✔
234	extra_input_digests = tools_shims.immutable_input_digests	1✔
235
236	return DockerBinary(	1✔
237	first_path.path,
238	first_path.fingerprint,
239	extra_env=extra_env,
240	extra_input_digests=extra_input_digests,
241	is_podman=is_podman,
242	)
243
244
245	def rules():	9✔
246	return collect_rules()	9✔

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