20632486505

Committed 01 Jan 2026 04:21AM UTC coverage: 43.231% (-37.1%) from 80.281%

Build # 20632486505

Build Type

Pull #22962

github

Committed by

web-flow

Commit Message

Merge 08d5c63b0 into f52ab6675

Pull Request Pull Request #22962: Bump the gha-deps group across 1 directory with 6 updates

Run Details

26122 of 60424 relevant lines covered (43.23%)

0.86 hits per line

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

53.97

/src/python/pants/backend/helm/resolve/remotes.py

# Copyright 2022 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, cast

from pants.util.frozendict import FrozenDict
from pants.util.memo import memoized_method

ALL_DEFAULT_HELM_REGISTRIES = "<ALL DEFAULT HELM REGISTRIES>"

OCI_REGISTRY_PROTOCOL = "oci://"


class InvalidHelmRegistryAddress(ValueError):
    def __init__(self, alias: str, address: str) -> None:
        super().__init__(
            f"The registry '{alias}' needs to have a valid OCI address URL "
            f"(using protocol '{OCI_REGISTRY_PROTOCOL}'). The given address was instead: {address}"
        )


class HelmRemoteAliasNotFoundError(ValueError):
    def __init__(self, alias: str) -> None:
        super().__init__(f"There is no Helm remote configured with alias: {alias}")


@dataclass(frozen=True)
class HelmRegistry:
    address: str
    alias: str = ""
    default: bool = False

    @classmethod
    def from_dict(cls, alias: str, d: dict[str, Any]) -> HelmRegistry:
        return cls(
            alias=alias,
            address=cast(str, d["address"]).rstrip("/"),
            default=d.get("default", alias == "default"),
        )

    def __post_init__(self) -> None:
        if not self.address.startswith(OCI_REGISTRY_PROTOCOL):
            raise InvalidHelmRegistryAddress(self.alias, self.address)

    def register(self, remotes: dict[str, HelmRegistry]) -> None:
        remotes[self.address] = self
        if self.alias:
            remotes[f"@{self.alias}"] = self

    def repository_ref(self, repository: str | None) -> str:
        return f"{self.address}/{repository or ''}".rstrip("/")

    def package_ref(self, artifact_name: str, *, repository: str | None) -> str:
        repo_ref = self.repository_ref(repository)
        return f"{repo_ref}/{artifact_name}"


@dataclass(frozen=True)
class HelmRemotes:
    default: tuple[HelmRegistry, ...]
    all: FrozenDict[str, HelmRegistry]

    @classmethod
    def from_dict(cls, d_regs: dict[str, Any]) -> HelmRemotes:
        remotes: dict[str, HelmRegistry] = {}
        for alias, opts in d_regs.items():
            HelmRegistry.from_dict(alias, opts).register(remotes)
        return cls(
            default=tuple(
                sorted(
                    {r for r in remotes.values() if isinstance(r, HelmRegistry) and r.default},
                    key=lambda r: r.address,
                )
            ),
            all=FrozenDict(remotes),
        )

    def get(self, *aliases_or_addresses: str) -> Iterator[HelmRegistry]:
        for alias_or_address in aliases_or_addresses:
            if alias_or_address in self.all:
                yield self.all[alias_or_address]
            elif alias_or_address.startswith("@"):
                raise HelmRemoteAliasNotFoundError(alias_or_address)
            elif alias_or_address == ALL_DEFAULT_HELM_REGISTRIES:
                yield from self.default
            elif alias_or_address.startswith(OCI_REGISTRY_PROTOCOL):
                yield HelmRegistry(address=alias_or_address)

    @memoized_method
    def registries(self) -> tuple[HelmRegistry, ...]:
        return tuple(set(self.all.values()))

    @property
    def default_registry(self) -> HelmRegistry | None:
        remote = self.all.get("default")
        if not remote and self.default:
            remote = self.default[0]
        return remote

1	# Copyright 2022 Pants project contributors (see CONTRIBUTORS.md).
2	# Licensed under the Apache License, Version 2.0 (see LICENSE).
3
4	from __future__ import annotations	2✔
5
6	from collections.abc import Iterator	2✔
7	from dataclasses import dataclass	2✔
8	from typing import Any, cast	2✔
9
10	from pants.util.frozendict import FrozenDict	2✔
11	from pants.util.memo import memoized_method	2✔
12
13	ALL_DEFAULT_HELM_REGISTRIES = "<ALL DEFAULT HELM REGISTRIES>"	2✔
14
15	OCI_REGISTRY_PROTOCOL = "oci://"	2✔
16
17
18	class InvalidHelmRegistryAddress(ValueError):	2✔
19	def __init__(self, alias: str, address: str) -> None:	2✔
20	super().__init__(	×
21	f"The registry '{alias}' needs to have a valid OCI address URL "
22	f"(using protocol '{OCI_REGISTRY_PROTOCOL}'). The given address was instead: {address}"
23	)
24
25
26	class HelmRemoteAliasNotFoundError(ValueError):	2✔
27	def __init__(self, alias: str) -> None:	2✔
28	super().__init__(f"There is no Helm remote configured with alias: {alias}")	×
29
30
31	@dataclass(frozen=True)	2✔
32	class HelmRegistry:	2✔
33	address: str	2✔
34	alias: str = ""	2✔
35	default: bool = False	2✔
36
37	@classmethod	2✔
38	def from_dict(cls, alias: str, d: dict[str, Any]) -> HelmRegistry:	2✔
39	return cls(	×
40	alias=alias,
41	address=cast(str, d["address"]).rstrip("/"),
42	default=d.get("default", alias == "default"),
43	)
44
45	def __post_init__(self) -> None:	2✔
46	if not self.address.startswith(OCI_REGISTRY_PROTOCOL):	×
47	raise InvalidHelmRegistryAddress(self.alias, self.address)	×
48
49	def register(self, remotes: dict[str, HelmRegistry]) -> None:	2✔
50	remotes[self.address] = self	×
51	if self.alias:	×
52	remotes[f"@{self.alias}"] = self	×
53
54	def repository_ref(self, repository: str \| None) -> str:	2✔
55	return f"{self.address}/{repository or ''}".rstrip("/")	×
56
57	def package_ref(self, artifact_name: str, *, repository: str \| None) -> str:	2✔
58	repo_ref = self.repository_ref(repository)	×
59	return f"{repo_ref}/{artifact_name}"	×
60
61
62	@dataclass(frozen=True)	2✔
63	class HelmRemotes:	2✔
64	default: tuple[HelmRegistry, ...]	2✔
65	all: FrozenDict[str, HelmRegistry]	2✔
66
67	@classmethod	2✔
68	def from_dict(cls, d_regs: dict[str, Any]) -> HelmRemotes:	2✔
69	remotes: dict[str, HelmRegistry] = {}	×
70	for alias, opts in d_regs.items():	×
71	HelmRegistry.from_dict(alias, opts).register(remotes)	×
72	return cls(	×
73	default=tuple(
74	sorted(
75	{r for r in remotes.values() if isinstance(r, HelmRegistry) and r.default},
76	key=lambda r: r.address,
77	)
78	),
79	all=FrozenDict(remotes),
80	)
81
82	def get(self, *aliases_or_addresses: str) -> Iterator[HelmRegistry]:	2✔
83	for alias_or_address in aliases_or_addresses:	×
84	if alias_or_address in self.all:	×
85	yield self.all[alias_or_address]	×
86	elif alias_or_address.startswith("@"):	×
87	raise HelmRemoteAliasNotFoundError(alias_or_address)	×
88	elif alias_or_address == ALL_DEFAULT_HELM_REGISTRIES:	×
89	yield from self.default	×
90	elif alias_or_address.startswith(OCI_REGISTRY_PROTOCOL):	×
91	yield HelmRegistry(address=alias_or_address)	×
92
93	@memoized_method	2✔
94	def registries(self) -> tuple[HelmRegistry, ...]:	2✔
95	return tuple(set(self.all.values()))	×
96
97	@property	2✔
98	def default_registry(self) -> HelmRegistry \| None:	2✔
99	remote = self.all.get("default")	×
100	if not remote and self.default:	×
101	remote = self.default[0]	×
102	return remote	×

pantsbuild / pants / 20632486505

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