21457998286

Committed 28 Jan 2026 10:32PM UTC coverage: 80.281% (+0.01%) from 80.269%

Build # 21457998286

Build Type

Pull #23037

github

Committed by

web-flow

Commit Message

Merge 43b38d939 into 0fdb40370

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

Run Details

275 of 328 new or added lines in 13 files covered. (83.84%)

46 existing lines in 9 files now uncovered.

78960 of 98355 relevant lines covered (80.28%)

3.36 hits per line

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

63.16

/src/python/pants/backend/docker/goals/publish.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
from collections import defaultdict
from dataclasses import dataclass
from itertools import chain
from typing import DefaultDict, cast

from pants.backend.docker.goals.package_image import (
    BuiltDockerImage,
    DockerPackageFieldSet,
    GetImageRefsRequest,
    get_image_refs,
)
from pants.backend.docker.registries import DockerRegistryOptions
from pants.backend.docker.subsystems.docker_options import DockerOptions
from pants.backend.docker.target_types import DockerImageRegistriesField, DockerImageSkipPushField
from pants.backend.docker.util_rules.docker_binary import DockerBinary
from pants.core.goals.package import PackageFieldSet
from pants.core.goals.publish import (
    PreemptiveSkipRequest,
    PublishFieldSet,
    PublishOutputData,
    PublishPackages,
    PublishProcesses,
    PublishRequest,
    SkippedPublishPackages,
)
from pants.core.util_rules.env_vars import environment_vars_subset
from pants.engine.env_vars import EnvironmentVarsRequest
from pants.engine.process import InteractiveProcess, Process
from pants.engine.rules import collect_rules, implicitly, rule
from pants.engine.unions import UnionRule

logger = logging.getLogger(__name__)


class PublishDockerImageRequest(PublishRequest):
    pass


@dataclass(frozen=True)
class PublishDockerImageFieldSet(PublishFieldSet):
    publish_request_type = PublishDockerImageRequest
    required_fields = (DockerImageRegistriesField,)

    registries: DockerImageRegistriesField
    skip_push: DockerImageSkipPushField

    def make_skip_request(
        self, package_fs: PackageFieldSet
    ) -> PublishDockerImageSkipRequest | None:
        return (
            PublishDockerImageSkipRequest(publish_fs=self, package_fs=package_fs)
            if isinstance(package_fs, DockerPackageFieldSet)
            else None
        )

    def get_output_data(self) -> PublishOutputData:
        return PublishOutputData(
            {
                "publisher": "docker",
                "registries": self.registries.value or (),
                **super().get_output_data(),
            }
        )


class PublishDockerImageSkipRequest(PreemptiveSkipRequest[PublishDockerImageFieldSet]):
    pass


@rule
async def check_if_skip_push(
    request: PublishDockerImageSkipRequest, options: DockerOptions
) -> SkippedPublishPackages:
    skip_registries = {
        registry for registry in options.registries().registries.values() if registry.skip_push
    }
    if not (request.publish_fs.skip_push.value or skip_registries):
        return SkippedPublishPackages.no_skip()
    image_refs = await get_image_refs(
        GetImageRefsRequest(
            field_set=cast(DockerPackageFieldSet, request.package_fs), build_upstream_images=False
        ),
        **implicitly(),
    )
    if request.publish_fs.skip_push.value:
        return SkippedPublishPackages.skip(
            names=[tag.full_name for registry in image_refs for tag in registry.tags],
            description=f"(by `{request.publish_fs.skip_push.alias}` on {request.address})",
            data=request.publish_fs.get_output_data(),
        )
    if all(image_ref.registry in skip_registries for image_ref in image_refs):
        output_data = request.publish_fs.get_output_data()
        return SkippedPublishPackages(
            PublishPackages(
                names=tuple(tag.full_name for tag in image_ref.tags),
                description=f"(by skip_push on @{cast(DockerRegistryOptions, image_ref.registry).alias})",
                data=output_data,
            )
            for image_ref in image_refs
        )

    return SkippedPublishPackages.no_skip()


@rule
async def push_docker_images(
    request: PublishDockerImageRequest,
    docker: DockerBinary,
    options: DockerOptions,
    options_env_aware: DockerOptions.EnvironmentAware,
) -> PublishProcesses:
    tags = tuple(
        chain.from_iterable(
            cast(BuiltDockerImage, image).tags
            for pkg in request.packages
            for image in pkg.artifacts
        )
    )
    if request.field_set.skip_push.value:
        return PublishProcesses(
            [
                PublishPackages(
                    names=tags,
                    description=f"(by `{request.field_set.skip_push.alias}` on {request.field_set.address})",
                ),
            ]
        )

    env = await environment_vars_subset(
        EnvironmentVarsRequest(options_env_aware.env_vars), **implicitly()
    )
    skip_push_reasons: DefaultDict[str, DefaultDict[str, set[str]]] = defaultdict(
        lambda: defaultdict(set)
    )
    jobs: list[PublishPackages] = []
    refs: list[str] = []
    processes: list[Process | InteractiveProcess] = []

    for tag in tags:
        for registry in options.registries().registries.values():
            if registry.skip_push and tag.startswith(f"{registry.address}/"):
                skip_push_reasons["skip_push"][registry.alias].add(tag)
                break
            if registry.use_local_alias and tag.startswith(f"{registry.alias}/"):
                skip_push_reasons["use_local_alias"][registry.alias].add(tag)
                break
        else:
            refs.append(tag)
            push_process = docker.push_image(tag, env)
            if options.publish_noninteractively:
                processes.append(push_process)
            else:
                processes.append(InteractiveProcess.from_process(push_process))

    for ref, process in zip(refs, processes):
        jobs.append(
            PublishPackages(
                names=(ref,),
                process=process,
            )
        )

    for reason, skip_push in skip_push_reasons.items():
        for name, skip_tags in skip_push.items():
            jobs.append(
                PublishPackages(
                    names=tuple(skip_tags),
                    description=f"(by `{reason}` on registry @{name})",
                ),
            )

    return PublishProcesses(jobs)


def rules():
    return (
        *collect_rules(),
        *PublishDockerImageFieldSet.rules(),
        UnionRule(PreemptiveSkipRequest, PublishDockerImageSkipRequest),
    )

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	6✔
5
6	import logging	6✔
7	from collections import defaultdict	6✔
8	from dataclasses import dataclass	6✔
9	from itertools import chain	6✔
10	from typing import DefaultDict, cast	6✔
11
12	from pants.backend.docker.goals.package_image import (	6✔
13	BuiltDockerImage,
14	DockerPackageFieldSet,
15	GetImageRefsRequest,
16	get_image_refs,
17	)
18	from pants.backend.docker.registries import DockerRegistryOptions	6✔
19	from pants.backend.docker.subsystems.docker_options import DockerOptions	6✔
20	from pants.backend.docker.target_types import DockerImageRegistriesField, DockerImageSkipPushField	6✔
21	from pants.backend.docker.util_rules.docker_binary import DockerBinary	6✔
22	from pants.core.goals.package import PackageFieldSet	6✔
23	from pants.core.goals.publish import (	6✔
24	PreemptiveSkipRequest,
25	PublishFieldSet,
26	PublishOutputData,
27	PublishPackages,
28	PublishProcesses,
29	PublishRequest,
30	SkippedPublishPackages,
31	)
32	from pants.core.util_rules.env_vars import environment_vars_subset	6✔
33	from pants.engine.env_vars import EnvironmentVarsRequest	6✔
34	from pants.engine.process import InteractiveProcess, Process	6✔
35	from pants.engine.rules import collect_rules, implicitly, rule	6✔
36	from pants.engine.unions import UnionRule	6✔
37
38	logger = logging.getLogger(__name__)	6✔
39
40
41	class PublishDockerImageRequest(PublishRequest):	6✔
42	pass	6✔
43
44
45	@dataclass(frozen=True)	6✔
46	class PublishDockerImageFieldSet(PublishFieldSet):	6✔
47	publish_request_type = PublishDockerImageRequest	6✔
48	required_fields = (DockerImageRegistriesField,)	6✔
49
50	registries: DockerImageRegistriesField	6✔
51	skip_push: DockerImageSkipPushField	6✔
52
53	def make_skip_request(	6✔
54	self, package_fs: PackageFieldSet
55	) -> PublishDockerImageSkipRequest \| None:
NEW 56	return (	×
57	PublishDockerImageSkipRequest(publish_fs=self, package_fs=package_fs)
58	if isinstance(package_fs, DockerPackageFieldSet)
59	else None
60	)
61
62	def get_output_data(self) -> PublishOutputData:	6✔
63	return PublishOutputData(	1✔
64	{
65	"publisher": "docker",
66	"registries": self.registries.value or (),
67	**super().get_output_data(),
68	}
69	)
70
71
72	class PublishDockerImageSkipRequest(PreemptiveSkipRequest[PublishDockerImageFieldSet]):	6✔
73	pass	6✔
74
75
76	@rule	6✔
77	async def check_if_skip_push(	6✔
78	request: PublishDockerImageSkipRequest, options: DockerOptions
79	) -> SkippedPublishPackages:
80	skip_registries = {	1✔
81	registry for registry in options.registries().registries.values() if registry.skip_push
82	}
83	if not (request.publish_fs.skip_push.value or skip_registries):	1✔
84	return SkippedPublishPackages.no_skip()	1✔
85	image_refs = await get_image_refs(	1✔
86	GetImageRefsRequest(
87	field_set=cast(DockerPackageFieldSet, request.package_fs), build_upstream_images=False
88	),
89	**implicitly(),
90	)
91	if request.publish_fs.skip_push.value:	1✔
92	return SkippedPublishPackages.skip(	1✔
93	names=[tag.full_name for registry in image_refs for tag in registry.tags],
94	description=f"(by `{request.publish_fs.skip_push.alias}` on {request.address})",
95	data=request.publish_fs.get_output_data(),
96	)
97	if all(image_ref.registry in skip_registries for image_ref in image_refs):	1✔
98	output_data = request.publish_fs.get_output_data()	1✔
99	return SkippedPublishPackages(	1✔
100	PublishPackages(
101	names=tuple(tag.full_name for tag in image_ref.tags),
102	description=f"(by skip_push on @{cast(DockerRegistryOptions, image_ref.registry).alias})",
103	data=output_data,
104	)
105	for image_ref in image_refs
106	)
107
108	return SkippedPublishPackages.no_skip()	1✔
109
110
111	@rule	6✔
112	async def push_docker_images(	6✔
113	request: PublishDockerImageRequest,
114	docker: DockerBinary,
115	options: DockerOptions,
116	options_env_aware: DockerOptions.EnvironmentAware,
117	) -> PublishProcesses:
118	tags = tuple(	×
119	chain.from_iterable(
120	cast(BuiltDockerImage, image).tags
121	for pkg in request.packages
122	for image in pkg.artifacts
123	)
124	)
UNCOV 125	if request.field_set.skip_push.value:	×
126	return PublishProcesses(	×
127	[
128	PublishPackages(
129	names=tags,
130	description=f"(by `{request.field_set.skip_push.alias}` on {request.field_set.address})",
131	),
132	]
133	)
134
135	env = await environment_vars_subset(	×
136	EnvironmentVarsRequest(options_env_aware.env_vars), **implicitly()
137	)
138	skip_push_reasons: DefaultDict[str, DefaultDict[str, set[str]]] = defaultdict(	×
139	lambda: defaultdict(set)
140	)
141	jobs: list[PublishPackages] = []	×
142	refs: list[str] = []	×
143	processes: list[Process \| InteractiveProcess] = []	×
144
145	for tag in tags:	×
146	for registry in options.registries().registries.values():	×
147	if registry.skip_push and tag.startswith(f"{registry.address}/"):	×
148	skip_push_reasons["skip_push"][registry.alias].add(tag)	×
149	break	×
150	if registry.use_local_alias and tag.startswith(f"{registry.alias}/"):	×
151	skip_push_reasons["use_local_alias"][registry.alias].add(tag)	×
152	break	×
153	else:
154	refs.append(tag)	×
155	push_process = docker.push_image(tag, env)	×
156	if options.publish_noninteractively:	×
157	processes.append(push_process)	×
158	else:
159	processes.append(InteractiveProcess.from_process(push_process))	×
160
161	for ref, process in zip(refs, processes):	×
162	jobs.append(	×
163	PublishPackages(
164	names=(ref,),
165	process=process,
166	)
167	)
168
169	for reason, skip_push in skip_push_reasons.items():	×
170	for name, skip_tags in skip_push.items():	×
171	jobs.append(	×
172	PublishPackages(
173	names=tuple(skip_tags),
174	description=f"(by `{reason}` on registry @{name})",
175	),
176	)
177
178	return PublishProcesses(jobs)	×
179
180
181	def rules():	6✔
182	return (	5✔
183	*collect_rules(),
184	*PublishDockerImageFieldSet.rules(),
185	UnionRule(PreemptiveSkipRequest, PublishDockerImageSkipRequest),
186	)

pantsbuild / pants / 21457998286

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