22285099215

Committed 22 Feb 2026 08:52PM UTC coverage: 75.854% (-17.1%) from 92.936%

Build # 22285099215

Build Type

Pull #23121

github

Committed by

web-flow

Commit Message

Merge c7299df9c into ba8359840

Pull Request Pull Request #23121: fix issue with optional fields in dependency validator

Run Details

28 of 29 new or added lines in 2 files covered. (96.55%)

11174 existing lines in 400 files now uncovered.

53694 of 70786 relevant lines covered (75.85%)

1.88 hits per line

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

69.7

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

# Copyright 2022 Pants project contributors (see CONTRIBUTORS.md).
# Licensed under the Apache License, Version 2.0 (see LICENSE).

from __future__ import annotations

from abc import ABC, abstractmethod
from collections.abc import Iterable
from dataclasses import dataclass
from typing import Any, cast

from pants.backend.helm.subsystems.helm import HelmSubsystem
from pants.backend.helm.target_types import (
    AllHelmArtifactTargets,
    HelmArtifactFieldSet,
    HelmArtifactRegistryField,
    HelmArtifactRepositoryField,
    HelmChartTarget,
)
from pants.backend.helm.util_rules.chart_metadata import rules as metadata_rules
from pants.engine.addresses import Address
from pants.engine.engine_aware import EngineAwareReturnType
from pants.engine.rules import collect_rules, concurrently, implicitly, rule
from pants.engine.target import Target
from pants.util.frozendict import FrozenDict
from pants.util.strutil import bullet_list


class MissingHelmArtifactLocation(ValueError):
    def __init__(self, address: Address) -> None:
        super().__init__(
            f"Target at address '{address}' needs to specify either `{HelmArtifactRegistryField.alias}`, "
            f"`{HelmArtifactRepositoryField.alias}` or both."
        )


class DuplicateHelmChartNamesFound(Exception):
    def __init__(self, duplicates: Iterable[tuple[str, Address]]) -> None:
        super().__init__(
            f"Found more than one `{HelmChartTarget.alias}` target using the same chart name:\n\n"
            f"{bullet_list([f'{addr} -> {name}' for name, addr in duplicates])}"
        )


class HelmArtifactLocationSpec(ABC):
    @property
    @abstractmethod
    def spec(self) -> str: ...

    @property
    def is_url(self) -> bool:
        return len(self.spec.split("://")) == 2

    @property
    def is_alias(self) -> bool:
        return self.spec.startswith("@")


@dataclass(frozen=True)
class HelmArtifactRegistryLocationSpec(HelmArtifactLocationSpec):
    registry: str
    repository: str | None

    @property
    def spec(self) -> str:
        return self.registry


@dataclass(frozen=True)
class HelmArtifactClassicRepositoryLocationSpec(HelmArtifactLocationSpec):
    repository: str

    @property
    def spec(self) -> str:
        return self.repository


@dataclass(frozen=True)
class HelmArtifactRequirement:
    name: str
    version: str
    location: HelmArtifactLocationSpec


@dataclass(frozen=True)
class HelmArtifact:
    requirement: HelmArtifactRequirement
    address: Address

    @classmethod
    def from_target(cls, target: Target) -> HelmArtifact:
        return cls.from_field_set(HelmArtifactFieldSet.create(target))

    @classmethod
    def from_field_set(cls, field_set: HelmArtifactFieldSet) -> HelmArtifact:
        registry = field_set.registry.value
        repository = field_set.repository.value
        if not registry and not repository:
            raise MissingHelmArtifactLocation(field_set.address)

        registry_location: HelmArtifactRegistryLocationSpec | None = None
        if registry:
            registry_location = HelmArtifactRegistryLocationSpec(registry.rstrip("/"), repository)

        location = registry_location or HelmArtifactClassicRepositoryLocationSpec(
            cast(str, repository).rstrip("/")
        )
        req = HelmArtifactRequirement(
            name=cast(str, field_set.artifact.value),
            version=cast(str, field_set.version.value),
            location=location,
        )

        return cls(requirement=req, address=field_set.address)

    @property
    def name(self) -> str:
        return self.requirement.name

    @property
    def version(self) -> str:
        return self.requirement.version


@dataclass(frozen=True)
class ResolvedHelmArtifact(HelmArtifact, EngineAwareReturnType):
    location_url: str

    @classmethod
    def from_unresolved(cls, artifact: HelmArtifact, *, location_url: str) -> ResolvedHelmArtifact:
        return cls(
            requirement=artifact.requirement,
            address=artifact.address,
            location_url=location_url,
        )

    @property
    def chart_url(self) -> str:
        return f"{self.location_url}/{self.name}"

    def metadata(self) -> dict[str, Any] | None:
        return {
            "name": self.requirement.name,
            "version": self.requirement.version,
            "location": self.requirement.location.spec,
            "address": self.address.spec,
            "url": self.chart_url,
        }


@rule
async def resolved_helm_artifact(
    artifact: HelmArtifact, subsystem: HelmSubsystem
) -> ResolvedHelmArtifact:
    remotes = subsystem.remotes()

    candidate_remotes = list(remotes.get(artifact.requirement.location.spec))
    if candidate_remotes:
        loc_url = candidate_remotes[0].address
        if isinstance(artifact.requirement.location, HelmArtifactRegistryLocationSpec):
            loc_url = f"{loc_url}/{artifact.requirement.location.repository or ''}".rstrip("/")
    else:
        loc_url = artifact.requirement.location.spec

    return ResolvedHelmArtifact.from_unresolved(artifact, location_url=loc_url)


class ThirdPartyHelmArtifactMapping(FrozenDict[str, Address]):
    pass


@rule
async def third_party_helm_artifact_mapping(
    all_helm_artifact_tgts: AllHelmArtifactTargets,
) -> ThirdPartyHelmArtifactMapping:
    artifacts = await concurrently(
        resolved_helm_artifact(**implicitly({HelmArtifact.from_target(tgt): HelmArtifact}))
        for tgt in all_helm_artifact_tgts
    )
    return ThirdPartyHelmArtifactMapping(
        {artifact.chart_url: artifact.address for artifact in artifacts}
    )


def rules():
    return [*collect_rules(), *metadata_rules()]

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	4✔
5
6	from abc import ABC, abstractmethod	4✔
7	from collections.abc import Iterable	4✔
8	from dataclasses import dataclass	4✔
9	from typing import Any, cast	4✔
10
11	from pants.backend.helm.subsystems.helm import HelmSubsystem	4✔
12	from pants.backend.helm.target_types import (	4✔
13	AllHelmArtifactTargets,
14	HelmArtifactFieldSet,
15	HelmArtifactRegistryField,
16	HelmArtifactRepositoryField,
17	HelmChartTarget,
18	)
19	from pants.backend.helm.util_rules.chart_metadata import rules as metadata_rules	4✔
20	from pants.engine.addresses import Address	4✔
21	from pants.engine.engine_aware import EngineAwareReturnType	4✔
22	from pants.engine.rules import collect_rules, concurrently, implicitly, rule	4✔
23	from pants.engine.target import Target	4✔
24	from pants.util.frozendict import FrozenDict	4✔
25	from pants.util.strutil import bullet_list	4✔
26
27
28	class MissingHelmArtifactLocation(ValueError):	4✔
29	def __init__(self, address: Address) -> None:	4✔
30	super().__init__(	×
31	f"Target at address '{address}' needs to specify either `{HelmArtifactRegistryField.alias}`, "
32	f"`{HelmArtifactRepositoryField.alias}` or both."
33	)
34
35
36	class DuplicateHelmChartNamesFound(Exception):	4✔
37	def __init__(self, duplicates: Iterable[tuple[str, Address]]) -> None:	4✔
38	super().__init__(	×
39	f"Found more than one `{HelmChartTarget.alias}` target using the same chart name:\n\n"
40	f"{bullet_list([f'{addr} -> {name}' for name, addr in duplicates])}"
41	)
42
43
44	class HelmArtifactLocationSpec(ABC):	4✔
45	@property
46	@abstractmethod
47	def spec(self) -> str: ...
48
49	@property	4✔
50	def is_url(self) -> bool:	4✔
51	return len(self.spec.split("://")) == 2	×
52
53	@property	4✔
54	def is_alias(self) -> bool:	4✔
55	return self.spec.startswith("@")	×
56
57
58	@dataclass(frozen=True)	4✔
59	class HelmArtifactRegistryLocationSpec(HelmArtifactLocationSpec):	4✔
60	registry: str	4✔
61	repository: str \| None	4✔
62
63	@property	4✔
64	def spec(self) -> str:	4✔
UNCOV 65	return self.registry	×
66
67
68	@dataclass(frozen=True)	4✔
69	class HelmArtifactClassicRepositoryLocationSpec(HelmArtifactLocationSpec):	4✔
70	repository: str	4✔
71
72	@property	4✔
73	def spec(self) -> str:	4✔
UNCOV 74	return self.repository	×
75
76
77	@dataclass(frozen=True)	4✔
78	class HelmArtifactRequirement:	4✔
79	name: str	4✔
80	version: str	4✔
81	location: HelmArtifactLocationSpec	4✔
82
83
84	@dataclass(frozen=True)	4✔
85	class HelmArtifact:	4✔
86	requirement: HelmArtifactRequirement	4✔
87	address: Address	4✔
88
89	@classmethod	4✔
90	def from_target(cls, target: Target) -> HelmArtifact:	4✔
UNCOV 91	return cls.from_field_set(HelmArtifactFieldSet.create(target))	×
92
93	@classmethod	4✔
94	def from_field_set(cls, field_set: HelmArtifactFieldSet) -> HelmArtifact:	4✔
UNCOV 95	registry = field_set.registry.value	×
UNCOV 96	repository = field_set.repository.value	×
UNCOV 97	if not registry and not repository:	×
98	raise MissingHelmArtifactLocation(field_set.address)	×
99
UNCOV 100	registry_location: HelmArtifactRegistryLocationSpec \| None = None	×
UNCOV 101	if registry:	×
UNCOV 102	registry_location = HelmArtifactRegistryLocationSpec(registry.rstrip("/"), repository)	×
103
UNCOV 104	location = registry_location or HelmArtifactClassicRepositoryLocationSpec(	×
105	cast(str, repository).rstrip("/")
106	)
UNCOV 107	req = HelmArtifactRequirement(	×
108	name=cast(str, field_set.artifact.value),
109	version=cast(str, field_set.version.value),
110	location=location,
111	)
112
UNCOV 113	return cls(requirement=req, address=field_set.address)	×
114
115	@property	4✔
116	def name(self) -> str:	4✔
UNCOV 117	return self.requirement.name	×
118
119	@property	4✔
120	def version(self) -> str:	4✔
UNCOV 121	return self.requirement.version	×
122
123
124	@dataclass(frozen=True)	4✔
125	class ResolvedHelmArtifact(HelmArtifact, EngineAwareReturnType):	4✔
126	location_url: str	4✔
127
128	@classmethod	4✔
129	def from_unresolved(cls, artifact: HelmArtifact, *, location_url: str) -> ResolvedHelmArtifact:	4✔
UNCOV 130	return cls(	×
131	requirement=artifact.requirement,
132	address=artifact.address,
133	location_url=location_url,
134	)
135
136	@property	4✔
137	def chart_url(self) -> str:	4✔
UNCOV 138	return f"{self.location_url}/{self.name}"	×
139
140	def metadata(self) -> dict[str, Any] \| None:	4✔
141	return {	×
142	"name": self.requirement.name,
143	"version": self.requirement.version,
144	"location": self.requirement.location.spec,
145	"address": self.address.spec,
146	"url": self.chart_url,
147	}
148
149
150	@rule	4✔
151	async def resolved_helm_artifact(	4✔
152	artifact: HelmArtifact, subsystem: HelmSubsystem
153	) -> ResolvedHelmArtifact:
UNCOV 154	remotes = subsystem.remotes()	×
155
UNCOV 156	candidate_remotes = list(remotes.get(artifact.requirement.location.spec))	×
UNCOV 157	if candidate_remotes:	×
UNCOV 158	loc_url = candidate_remotes[0].address	×
UNCOV 159	if isinstance(artifact.requirement.location, HelmArtifactRegistryLocationSpec):	×
UNCOV 160	loc_url = f"{loc_url}/{artifact.requirement.location.repository or ''}".rstrip("/")	×
161	else:
UNCOV 162	loc_url = artifact.requirement.location.spec	×
163
UNCOV 164	return ResolvedHelmArtifact.from_unresolved(artifact, location_url=loc_url)	×
165
166
167	class ThirdPartyHelmArtifactMapping(FrozenDict[str, Address]):	4✔
168	pass	4✔
169
170
171	@rule	4✔
172	async def third_party_helm_artifact_mapping(	4✔
173	all_helm_artifact_tgts: AllHelmArtifactTargets,
174	) -> ThirdPartyHelmArtifactMapping:
175	artifacts = await concurrently(	4✔
176	resolved_helm_artifact(**implicitly({HelmArtifact.from_target(tgt): HelmArtifact}))
177	for tgt in all_helm_artifact_tgts
178	)
179	return ThirdPartyHelmArtifactMapping(	4✔
180	{artifact.chart_url: artifact.address for artifact in artifacts}
181	)
182
183
184	def rules():	4✔
185	return [collect_rules(), metadata_rules()]	4✔

pantsbuild / pants / 22285099215

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