19381742489

Committed 15 Nov 2025 12:52AM UTC coverage: 49.706% (-30.6%) from 80.29%

Build # 19381742489

Build Type

Pull #22890

github

Committed by

web-flow

Commit Message

Merge d961abf79 into 42e1ebd41

Pull Request Pull Request #22890: Updated all python subsystem constraints to 3.14

Run Details

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

14659 existing lines in 485 files now uncovered.

31583 of 63540 relevant lines covered (49.71%)

0.79 hits per line

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

67.68

/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	2✔
5
6	from abc import ABC, abstractmethod	2✔
7	from collections.abc import Iterable	2✔
8	from dataclasses import dataclass	2✔
9	from typing import Any, cast	2✔
10
11	from pants.backend.helm.subsystems.helm import HelmSubsystem	2✔
12	from pants.backend.helm.target_types import (	2✔
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	2✔
20	from pants.engine.addresses import Address	2✔
21	from pants.engine.engine_aware import EngineAwareReturnType	2✔
22	from pants.engine.rules import collect_rules, concurrently, implicitly, rule	2✔
23	from pants.engine.target import Target	2✔
24	from pants.util.frozendict import FrozenDict	2✔
25	from pants.util.strutil import bullet_list	2✔
26
27
28	class MissingHelmArtifactLocation(ValueError):	2✔
29	def __init__(self, address: Address) -> None:	2✔
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):	2✔
37	def __init__(self, duplicates: Iterable[tuple[str, Address]]) -> None:	2✔
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):	2✔
45	@property
46	@abstractmethod
47	def spec(self) -> str: ...
48
49	@property	2✔
50	def is_url(self) -> bool:	2✔
51	return len(self.spec.split("://")) == 2	×
52
53	@property	2✔
54	def is_alias(self) -> bool:	2✔
55	return self.spec.startswith("@")	×
56
57
58	@dataclass(frozen=True)	2✔
59	class HelmArtifactRegistryLocationSpec(HelmArtifactLocationSpec):	2✔
60	registry: str	2✔
61	repository: str \| None	2✔
62
63	@property	2✔
64	def spec(self) -> str:	2✔
65	return self.registry	×
66
67
68	@dataclass(frozen=True)	2✔
69	class HelmArtifactClassicRepositoryLocationSpec(HelmArtifactLocationSpec):	2✔
70	repository: str	2✔
71
72	@property	2✔
73	def spec(self) -> str:	2✔
74	return self.repository	×
75
76
77	@dataclass(frozen=True)	2✔
78	class HelmArtifactRequirement:	2✔
79	name: str	2✔
80	version: str	2✔
81	location: HelmArtifactLocationSpec	2✔
82
83
84	@dataclass(frozen=True)	2✔
85	class HelmArtifact:	2✔
86	requirement: HelmArtifactRequirement	2✔
87	address: Address	2✔
88
89	@classmethod	2✔
90	def from_target(cls, target: Target) -> HelmArtifact:	2✔
UNCOV 91	return cls.from_field_set(HelmArtifactFieldSet.create(target))	×
92
93	@classmethod	2✔
94	def from_field_set(cls, field_set: HelmArtifactFieldSet) -> HelmArtifact:	2✔
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	2✔
116	def name(self) -> str:	2✔
117	return self.requirement.name	×
118
119	@property	2✔
120	def version(self) -> str:	2✔
121	return self.requirement.version	×
122
123
124	@dataclass(frozen=True)	2✔
125	class ResolvedHelmArtifact(HelmArtifact, EngineAwareReturnType):	2✔
126	location_url: str	2✔
127
128	@classmethod	2✔
129	def from_unresolved(cls, artifact: HelmArtifact, *, location_url: str) -> ResolvedHelmArtifact:	2✔
130	return cls(	×
131	requirement=artifact.requirement,
132	address=artifact.address,
133	location_url=location_url,
134	)
135
136	@property	2✔
137	def chart_url(self) -> str:	2✔
138	return f"{self.location_url}/{self.name}"	×
139
140	def metadata(self) -> dict[str, Any] \| None:	2✔
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	2✔
151	async def resolved_helm_artifact(	2✔
152	artifact: HelmArtifact, subsystem: HelmSubsystem
153	) -> ResolvedHelmArtifact:
154	remotes = subsystem.remotes()	×
155
156	candidate_remotes = list(remotes.get(artifact.requirement.location.spec))	×
157	if candidate_remotes:	×
158	loc_url = candidate_remotes[0].address	×
159	if isinstance(artifact.requirement.location, HelmArtifactRegistryLocationSpec):	×
160	loc_url = f"{loc_url}/{artifact.requirement.location.repository or ''}".rstrip("/")	×
161	else:
162	loc_url = artifact.requirement.location.spec	×
163
164	return ResolvedHelmArtifact.from_unresolved(artifact, location_url=loc_url)	×
165
166
167	class ThirdPartyHelmArtifactMapping(FrozenDict[str, Address]):	2✔
168	pass	2✔
169
170
171	@rule	2✔
172	async def third_party_helm_artifact_mapping(	2✔
173	all_helm_artifact_tgts: AllHelmArtifactTargets,
174	) -> ThirdPartyHelmArtifactMapping:
175	artifacts = await concurrently(	×
176	resolved_helm_artifact(**implicitly({HelmArtifact.from_target(tgt): HelmArtifact}))
177	for tgt in all_helm_artifact_tgts
178	)
179	return ThirdPartyHelmArtifactMapping(	×
180	{artifact.chart_url: artifact.address for artifact in artifacts}
181	)
182
183
184	def rules():	2✔
185	return [collect_rules(), metadata_rules()]	2✔

pantsbuild / pants / 19381742489

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