21803785359

Committed 08 Feb 2026 07:13PM UTC coverage: 43.3% (-37.0%) from 80.277%

Build # 21803785359

Build Type

Pull #23085

github

Committed by

web-flow

Commit Message

Merge 7c1cd926d into 40389cc58

Pull Request Pull Request #23085: A helper method for indexing paths by source root

Run Details

2 of 6 new or added lines in 1 file covered. (33.33%)

17114 existing lines in 539 files now uncovered.

26075 of 60219 relevant lines covered (43.3%)

0.43 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	1✔
5
6	from collections.abc import Iterator	1✔
7	from dataclasses import dataclass	1✔
8	from typing import Any, cast	1✔
9
10	from pants.util.frozendict import FrozenDict	1✔
11	from pants.util.memo import memoized_method	1✔
12
13	ALL_DEFAULT_HELM_REGISTRIES = "<ALL DEFAULT HELM REGISTRIES>"	1✔
14
15	OCI_REGISTRY_PROTOCOL = "oci://"	1✔
16
17
18	class InvalidHelmRegistryAddress(ValueError):	1✔
19	def __init__(self, alias: str, address: str) -> None:	1✔
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):	1✔
27	def __init__(self, alias: str) -> None:	1✔
UNCOV 28	super().__init__(f"There is no Helm remote configured with alias: {alias}")	×
29
30
31	@dataclass(frozen=True)	1✔
32	class HelmRegistry:	1✔
33	address: str	1✔
34	alias: str = ""	1✔
35	default: bool = False	1✔
36
37	@classmethod	1✔
38	def from_dict(cls, alias: str, d: dict[str, Any]) -> HelmRegistry:	1✔
UNCOV 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:	1✔
UNCOV 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:	1✔
UNCOV 50	remotes[self.address] = self	×
UNCOV 51	if self.alias:	×
UNCOV 52	remotes[f"@{self.alias}"] = self	×
53
54	def repository_ref(self, repository: str \| None) -> str:	1✔
55	return f"{self.address}/{repository or ''}".rstrip("/")	×
56
57	def package_ref(self, artifact_name: str, *, repository: str \| None) -> str:	1✔
58	repo_ref = self.repository_ref(repository)	×
59	return f"{repo_ref}/{artifact_name}"	×
60
61
62	@dataclass(frozen=True)	1✔
63	class HelmRemotes:	1✔
64	default: tuple[HelmRegistry, ...]	1✔
65	all: FrozenDict[str, HelmRegistry]	1✔
66
67	@classmethod	1✔
68	def from_dict(cls, d_regs: dict[str, Any]) -> HelmRemotes:	1✔
UNCOV 69	remotes: dict[str, HelmRegistry] = {}	×
UNCOV 70	for alias, opts in d_regs.items():	×
UNCOV 71	HelmRegistry.from_dict(alias, opts).register(remotes)	×
UNCOV 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]:	1✔
UNCOV 83	for alias_or_address in aliases_or_addresses:	×
UNCOV 84	if alias_or_address in self.all:	×
UNCOV 85	yield self.all[alias_or_address]	×
UNCOV 86	elif alias_or_address.startswith("@"):	×
UNCOV 87	raise HelmRemoteAliasNotFoundError(alias_or_address)	×
UNCOV 88	elif alias_or_address == ALL_DEFAULT_HELM_REGISTRIES:	×
UNCOV 89	yield from self.default	×
UNCOV 90	elif alias_or_address.startswith(OCI_REGISTRY_PROTOCOL):	×
UNCOV 91	yield HelmRegistry(address=alias_or_address)	×
92
93	@memoized_method	1✔
94	def registries(self) -> tuple[HelmRegistry, ...]:	1✔
95	return tuple(set(self.all.values()))	×
96
97	@property	1✔
98	def default_registry(self) -> HelmRegistry \| None:	1✔
99	remote = self.all.get("default")	×
100	if not remote and self.default:	×
101	remote = self.default[0]	×
102	return remote	×

pantsbuild / pants / 21803785359

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