21552830208

Committed 31 Jan 2026 11:40PM UTC coverage: 80.277% (-0.05%) from 80.324%

Build # 21552830208

Build Type

Pull #23062

github

Committed by

web-flow

Commit Message

Merge 808a9786c into 2c4dcf9cf

Pull Request Pull Request #23062: Remove support for Get

Run Details

18 of 25 new or added lines in 4 files covered. (72.0%)

17119 existing lines in 541 files now uncovered.

78278 of 97510 relevant lines covered (80.28%)

3.36 hits per line

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

100.0

/src/python/pants/backend/docker/registries.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 Iterator
from dataclasses import dataclass
from typing import Any

from pants.util.frozendict import FrozenDict
from pants.util.strutil import softwrap

ALL_DEFAULT_REGISTRIES = "<all default registries>"


class DockerRegistryError(ValueError):
    pass


class DockerRegistryOptionsNotFoundError(DockerRegistryError):
    def __init__(self, message):
        super().__init__(
            f"{message}\n\n"
            "Use the [docker].registries configuration option to define custom registries."
        )


class DockerRegistryAddressCollisionError(DockerRegistryError):
    def __init__(self, first, second):
        message = softwrap(
            f"""
            Duplicated docker registry address for aliases: {first.alias}, {second.alias}.
            Each registry `address` in `[docker].registries` must be unique.
            """
        )

        super().__init__(message)


@dataclass(frozen=True)
class DockerRegistryOptions:
    address: str
    alias: str = ""
    default: bool = False
    skip_push: bool = False
    extra_image_tags: tuple[str, ...] = ()
    repository: str | None = None
    use_local_alias: bool = False

    @classmethod
    def from_dict(cls, alias: str, d: dict[str, Any]) -> DockerRegistryOptions:
        return cls(
            alias=alias,
            address=d["address"],
            default=d.get("default", alias == "default"),
            skip_push=d.get("skip_push", DockerRegistryOptions.skip_push),
            extra_image_tags=tuple(
                d.get("extra_image_tags", DockerRegistryOptions.extra_image_tags)
            ),
            repository=d.get("repository"),
            use_local_alias=d.get("use_local_alias", False),
        )

    def register(self, registries: dict[str, DockerRegistryOptions]) -> None:
        if self.address in registries:
            collision = registries[self.address]
            raise DockerRegistryAddressCollisionError(collision, self)
        registries[self.address] = self
        if self.alias:
            registries[f"@{self.alias}"] = self


@dataclass(frozen=True)
class DockerRegistries:
    default: tuple[DockerRegistryOptions, ...]
    registries: FrozenDict[str, DockerRegistryOptions]

    @classmethod
    def from_dict(cls, d: dict[str, Any]) -> DockerRegistries:
        registries: dict[str, DockerRegistryOptions] = {}
        for alias, options in d.items():
            DockerRegistryOptions.from_dict(alias, options).register(registries)
        return cls(
            default=tuple(
                sorted({r for r in registries.values() if r.default}, key=lambda r: r.address)
            ),
            registries=FrozenDict(registries),
        )

    def get(self, *aliases_or_addresses: str) -> Iterator[DockerRegistryOptions]:
        for alias_or_address in aliases_or_addresses:
            if alias_or_address in self.registries:
                # Get configured registry by "@alias" or "address".
                yield self.registries[alias_or_address]
            elif alias_or_address.startswith("@"):
                raise DockerRegistryOptionsNotFoundError(
                    f"There is no Docker registry configured with alias: {alias_or_address[1:]}."
                )
            elif alias_or_address == ALL_DEFAULT_REGISTRIES:
                yield from self.default
            else:
                # Assume an explicit address from the BUILD file.
                yield DockerRegistryOptions(address=alias_or_address)

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	11✔
5
6	from collections.abc import Iterator	11✔
7	from dataclasses import dataclass	11✔
8	from typing import Any	11✔
9
10	from pants.util.frozendict import FrozenDict	11✔
11	from pants.util.strutil import softwrap	11✔
12
13	ALL_DEFAULT_REGISTRIES = "<all default registries>"	11✔
14
15
16	class DockerRegistryError(ValueError):	11✔
17	pass	11✔
18
19
20	class DockerRegistryOptionsNotFoundError(DockerRegistryError):	11✔
21	def __init__(self, message):	11✔
UNCOV 22	super().__init__(	1✔
23	f"{message}\n\n"
24	"Use the [docker].registries configuration option to define custom registries."
25	)
26
27
28	class DockerRegistryAddressCollisionError(DockerRegistryError):	11✔
29	def __init__(self, first, second):	11✔
UNCOV 30	message = softwrap(	1✔
31	f"""
32	Duplicated docker registry address for aliases: {first.alias}, {second.alias}.
33	Each registry `address` in `[docker].registries` must be unique.
34	"""
35	)
36
UNCOV 37	super().__init__(message)	1✔
38
39
40	@dataclass(frozen=True)	11✔
41	class DockerRegistryOptions:	11✔
42	address: str	11✔
43	alias: str = ""	11✔
44	default: bool = False	11✔
45	skip_push: bool = False	11✔
46	extra_image_tags: tuple[str, ...] = ()	11✔
47	repository: str \| None = None	11✔
48	use_local_alias: bool = False	11✔
49
50	@classmethod	11✔
51	def from_dict(cls, alias: str, d: dict[str, Any]) -> DockerRegistryOptions:	11✔
UNCOV 52	return cls(	3✔
53	alias=alias,
54	address=d["address"],
55	default=d.get("default", alias == "default"),
56	skip_push=d.get("skip_push", DockerRegistryOptions.skip_push),
57	extra_image_tags=tuple(
58	d.get("extra_image_tags", DockerRegistryOptions.extra_image_tags)
59	),
60	repository=d.get("repository"),
61	use_local_alias=d.get("use_local_alias", False),
62	)
63
64	def register(self, registries: dict[str, DockerRegistryOptions]) -> None:	11✔
UNCOV 65	if self.address in registries:	3✔
UNCOV 66	collision = registries[self.address]	1✔
UNCOV 67	raise DockerRegistryAddressCollisionError(collision, self)	1✔
UNCOV 68	registries[self.address] = self	3✔
UNCOV 69	if self.alias:	3✔
UNCOV 70	registries[f"@{self.alias}"] = self	3✔
71
72
73	@dataclass(frozen=True)	11✔
74	class DockerRegistries:	11✔
75	default: tuple[DockerRegistryOptions, ...]	11✔
76	registries: FrozenDict[str, DockerRegistryOptions]	11✔
77
78	@classmethod	11✔
79	def from_dict(cls, d: dict[str, Any]) -> DockerRegistries:	11✔
UNCOV 80	registries: dict[str, DockerRegistryOptions] = {}	3✔
UNCOV 81	for alias, options in d.items():	3✔
UNCOV 82	DockerRegistryOptions.from_dict(alias, options).register(registries)	3✔
UNCOV 83	return cls(	3✔
84	default=tuple(
85	sorted({r for r in registries.values() if r.default}, key=lambda r: r.address)
86	),
87	registries=FrozenDict(registries),
88	)
89
90	def get(self, *aliases_or_addresses: str) -> Iterator[DockerRegistryOptions]:	11✔
UNCOV 91	for alias_or_address in aliases_or_addresses:	3✔
UNCOV 92	if alias_or_address in self.registries:	3✔
93	# Get configured registry by "@alias" or "address".
UNCOV 94	yield self.registries[alias_or_address]	3✔
UNCOV 95	elif alias_or_address.startswith("@"):	3✔
UNCOV 96	raise DockerRegistryOptionsNotFoundError(	1✔
97	f"There is no Docker registry configured with alias: {alias_or_address[1:]}."
98	)
UNCOV 99	elif alias_or_address == ALL_DEFAULT_REGISTRIES:	3✔
UNCOV 100	yield from self.default	2✔
101	else:
102	# Assume an explicit address from the BUILD file.
UNCOV 103	yield DockerRegistryOptions(address=alias_or_address)	2✔

pantsbuild / pants / 21552830208

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