• Home
  • Features
  • Pricing
  • Docs
  • Announcements
  • Sign In

pantsbuild / pants / 19015773527

02 Nov 2025 05:33PM UTC coverage: 17.872% (-62.4%) from 80.3%
19015773527

Pull #22816

github

web-flow
Merge a12d75757 into 6c024e162
Pull Request #22816: Update Pants internal Python to 3.14

4 of 5 new or added lines in 3 files covered. (80.0%)

28452 existing lines in 683 files now uncovered.

9831 of 55007 relevant lines covered (17.87%)

0.18 hits per line

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

0.0
/src/python/pants/backend/docker/registries.py
1
# Copyright 2021 Pants project contributors (see CONTRIBUTORS.md).
2
# Licensed under the Apache License, Version 2.0 (see LICENSE).
3

UNCOV
4
from __future__ import annotations
×
5

UNCOV
6
from collections.abc import Iterator
×
UNCOV
7
from dataclasses import dataclass
×
UNCOV
8
from typing import Any
×
9

UNCOV
10
from pants.util.frozendict import FrozenDict
×
UNCOV
11
from pants.util.strutil import softwrap
×
12

UNCOV
13
ALL_DEFAULT_REGISTRIES = "<all default registries>"
×
14

15

UNCOV
16
class DockerRegistryError(ValueError):
×
UNCOV
17
    pass
×
18

19

UNCOV
20
class DockerRegistryOptionsNotFoundError(DockerRegistryError):
×
UNCOV
21
    def __init__(self, message):
×
UNCOV
22
        super().__init__(
×
23
            f"{message}\n\n"
24
            "Use the [docker].registries configuration option to define custom registries."
25
        )
26

27

UNCOV
28
class DockerRegistryAddressCollisionError(DockerRegistryError):
×
UNCOV
29
    def __init__(self, first, second):
×
UNCOV
30
        message = softwrap(
×
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)
×
38

39

UNCOV
40
@dataclass(frozen=True)
×
UNCOV
41
class DockerRegistryOptions:
×
UNCOV
42
    address: str
×
UNCOV
43
    alias: str = ""
×
UNCOV
44
    default: bool = False
×
UNCOV
45
    skip_push: bool = False
×
UNCOV
46
    extra_image_tags: tuple[str, ...] = ()
×
UNCOV
47
    repository: str | None = None
×
UNCOV
48
    use_local_alias: bool = False
×
49

UNCOV
50
    @classmethod
×
UNCOV
51
    def from_dict(cls, alias: str, d: dict[str, Any]) -> DockerRegistryOptions:
×
UNCOV
52
        return cls(
×
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

UNCOV
64
    def register(self, registries: dict[str, DockerRegistryOptions]) -> None:
×
UNCOV
65
        if self.address in registries:
×
UNCOV
66
            collision = registries[self.address]
×
UNCOV
67
            raise DockerRegistryAddressCollisionError(collision, self)
×
UNCOV
68
        registries[self.address] = self
×
UNCOV
69
        if self.alias:
×
UNCOV
70
            registries[f"@{self.alias}"] = self
×
71

72

UNCOV
73
@dataclass(frozen=True)
×
UNCOV
74
class DockerRegistries:
×
UNCOV
75
    default: tuple[DockerRegistryOptions, ...]
×
UNCOV
76
    registries: FrozenDict[str, DockerRegistryOptions]
×
77

UNCOV
78
    @classmethod
×
UNCOV
79
    def from_dict(cls, d: dict[str, Any]) -> DockerRegistries:
×
UNCOV
80
        registries: dict[str, DockerRegistryOptions] = {}
×
UNCOV
81
        for alias, options in d.items():
×
UNCOV
82
            DockerRegistryOptions.from_dict(alias, options).register(registries)
×
UNCOV
83
        return cls(
×
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

UNCOV
90
    def get(self, *aliases_or_addresses: str) -> Iterator[DockerRegistryOptions]:
×
UNCOV
91
        for alias_or_address in aliases_or_addresses:
×
UNCOV
92
            if alias_or_address in self.registries:
×
93
                # Get configured registry by "@alias" or "address".
UNCOV
94
                yield self.registries[alias_or_address]
×
UNCOV
95
            elif alias_or_address.startswith("@"):
×
UNCOV
96
                raise DockerRegistryOptionsNotFoundError(
×
97
                    f"There is no Docker registry configured with alias: {alias_or_address[1:]}."
98
                )
UNCOV
99
            elif alias_or_address == ALL_DEFAULT_REGISTRIES:
×
UNCOV
100
                yield from self.default
×
101
            else:
102
                # Assume an explicit address from the BUILD file.
UNCOV
103
                yield DockerRegistryOptions(address=alias_or_address)
×
STATUS · Troubleshooting · Open an Issue · Sales · Support · CAREERS · ENTERPRISE · START FREE · SCHEDULE DEMO
ANNOUNCEMENTS · TWITTER · TOS & SLA · Supported CI Services · What's a CI service? · Automated Testing

© 2025 Coveralls, Inc