18252174847

Committed 05 Oct 2025 01:36AM UTC coverage: 43.382% (-36.9%) from 80.261%

Build # 18252174847

Build Type

push

github

Committed by

web-flow

Commit Message

run tests on mac arm (#22717)

Just doing the minimal to pull forward the x86_64 pattern.

ref #20993

Run Details

25776 of 59416 relevant lines covered (43.38%)

1.3 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	3✔
5
6	from abc import ABC, abstractmethod	3✔
7	from collections.abc import Iterable	3✔
8	from dataclasses import dataclass	3✔
9	from typing import Any, cast	3✔
10
11	from pants.backend.helm.subsystems.helm import HelmSubsystem	3✔
12	from pants.backend.helm.target_types import (	3✔
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	3✔
20	from pants.engine.addresses import Address	3✔
21	from pants.engine.engine_aware import EngineAwareReturnType	3✔
22	from pants.engine.rules import collect_rules, concurrently, implicitly, rule	3✔
23	from pants.engine.target import Target	3✔
24	from pants.util.frozendict import FrozenDict	3✔
25	from pants.util.strutil import bullet_list	3✔
26
27
28	class MissingHelmArtifactLocation(ValueError):	3✔
29	def __init__(self, address: Address) -> None:	3✔
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):	3✔
37	def __init__(self, duplicates: Iterable[tuple[str, Address]]) -> None:	3✔
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):	3✔
45	@property
46	@abstractmethod
47	def spec(self) -> str: ...
48
49	@property	3✔
50	def is_url(self) -> bool:	3✔
51	return len(self.spec.split("://")) == 2	×
52
53	@property	3✔
54	def is_alias(self) -> bool:	3✔
55	return self.spec.startswith("@")	×
56
57
58	@dataclass(frozen=True)	3✔
59	class HelmArtifactRegistryLocationSpec(HelmArtifactLocationSpec):	3✔
60	registry: str	3✔
61	repository: str \| None	3✔
62
63	@property	3✔
64	def spec(self) -> str:	3✔
65	return self.registry	×
66
67
68	@dataclass(frozen=True)	3✔
69	class HelmArtifactClassicRepositoryLocationSpec(HelmArtifactLocationSpec):	3✔
70	repository: str	3✔
71
72	@property	3✔
73	def spec(self) -> str:	3✔
74	return self.repository	×
75
76
77	@dataclass(frozen=True)	3✔
78	class HelmArtifactRequirement:	3✔
79	name: str	3✔
80	version: str	3✔
81	location: HelmArtifactLocationSpec	3✔
82
83
84	@dataclass(frozen=True)	3✔
85	class HelmArtifact:	3✔
86	requirement: HelmArtifactRequirement	3✔
87	address: Address	3✔
88
89	@classmethod	3✔
90	def from_target(cls, target: Target) -> HelmArtifact:	3✔
91	return cls.from_field_set(HelmArtifactFieldSet.create(target))	×
92
93	@classmethod	3✔
94	def from_field_set(cls, field_set: HelmArtifactFieldSet) -> HelmArtifact:	3✔
95	registry = field_set.registry.value	×
96	repository = field_set.repository.value	×
97	if not registry and not repository:	×
98	raise MissingHelmArtifactLocation(field_set.address)	×
99
100	registry_location: HelmArtifactRegistryLocationSpec \| None = None	×
101	if registry:	×
102	registry_location = HelmArtifactRegistryLocationSpec(registry.rstrip("/"), repository)	×
103
104	location = registry_location or HelmArtifactClassicRepositoryLocationSpec(	×
105	cast(str, repository).rstrip("/")
106	)
107	req = HelmArtifactRequirement(	×
108	name=cast(str, field_set.artifact.value),
109	version=cast(str, field_set.version.value),
110	location=location,
111	)
112
113	return cls(requirement=req, address=field_set.address)	×
114
115	@property	3✔
116	def name(self) -> str:	3✔
117	return self.requirement.name	×
118
119	@property	3✔
120	def version(self) -> str:	3✔
121	return self.requirement.version	×
122
123
124	@dataclass(frozen=True)	3✔
125	class ResolvedHelmArtifact(HelmArtifact, EngineAwareReturnType):	3✔
126	location_url: str	3✔
127
128	@classmethod	3✔
129	def from_unresolved(cls, artifact: HelmArtifact, *, location_url: str) -> ResolvedHelmArtifact:	3✔
130	return cls(	×
131	requirement=artifact.requirement,
132	address=artifact.address,
133	location_url=location_url,
134	)
135
136	@property	3✔
137	def chart_url(self) -> str:	3✔
138	return f"{self.location_url}/{self.name}"	×
139
140	def metadata(self) -> dict[str, Any] \| None:	3✔
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	3✔
151	async def resolved_helm_artifact(	3✔
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]):	3✔
168	pass	3✔
169
170
171	@rule	3✔
172	async def third_party_helm_artifact_mapping(	3✔
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():	3✔
185	return [collect_rules(), metadata_rules()]	3✔

pantsbuild / pants / 18252174847

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