26342152999

Committed 23 May 2026 07:59PM UTC coverage: 91.165% (-1.6%) from 92.792%

Build # 26342152999

Build Type

push

github

Committed by

web-flow

Commit Message

Run Linux ARM CI on Depot runners (#23363)

RunsOn is deprecating their v2 stack, and rather than migrate
to v3 we should use the resources kindly donated by Depot.

GitHub also now has Linux ARM runners, should we need them.

Coverage Stats

87305 of 95766 relevant lines covered (91.16%)

3.87 hits per line

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

48.68

/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 (
    DockerPackageFieldSet,
    GetImageRefsRequest,
    get_docker_image_build_process,
    get_image_refs,
)
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 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 (
    CheckSkipRequest,
    CheckSkipResult,
    PublishFieldSet,
    PublishOutputData,
    PublishPackages,
    PublishProcesses,
    PublishRequest,
)
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, DockerPackageFieldSet):
    publish_request_type = PublishDockerImageRequest
    required_fields = (  # type: ignore[assignment]
        *DockerPackageFieldSet.required_fields,
        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(CheckSkipRequest[PublishDockerImageFieldSet]):
    package_fs: DockerPackageFieldSet


@rule
async def check_if_skip_push(
    request: PublishDockerImageSkipRequest, options: DockerOptions
) -> CheckSkipResult:
    skip_registries = {
        registry for registry in options.registries().registries.values() if registry.skip_push
    }
    if skip_registries or request.publish_fs.skip_push.value:
        image_refs = await get_image_refs(
            GetImageRefsRequest(field_set=request.package_fs, build_upstream_images=False),
            **implicitly(),
        )
        if request.publish_fs.skip_push.value:
            return CheckSkipResult.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 CheckSkipResult(
                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 (
        CheckSkipResult.skip(skip_packaging_only=True)
        if request.package_fs.pushes_on_package()
        else CheckSkipResult.no_skip()
    )


@rule
async def push_docker_images(
    request: PublishDockerImageRequest,
    docker: DockerBinary,
    options: DockerOptions,
    options_env_aware: DockerOptions.EnvironmentAware,
) -> PublishProcesses:
    if cast(DockerPackageFieldSet, request.field_set).pushes_on_package():
        build_process = await get_docker_image_build_process(request.field_set, **implicitly())
        return PublishProcesses(
            [
                PublishPackages(
                    names=build_process.tags,
                    process=build_process.process
                    if options.publish_noninteractively
                    else InteractiveProcess.from_process(build_process.process),
                )
            ]
        )

    tags = tuple(
        chain.from_iterable(
            cast(BuiltDockerImage, image).tags
            for pkg in request.packages
            for image in pkg.artifacts
        )
    )

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

pantsbuild / pants / 26342152999

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