19015773527

Committed 02 Nov 2025 05:33PM UTC coverage: 17.872% (-62.4%) from 80.3%

Build # 19015773527

Build Type

Pull #22816

github

Committed by

web-flow

Commit Message

Merge a12d75757 into 6c024e162

Pull Request Pull Request #22816: Update Pants internal Python to 3.14

Run Details

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

28452 existing lines in 683 files now uncovered.

9831 of 55007 relevant lines covered (17.87%)

0.18 hits per line

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

0.0

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

1	# Copyright 2022 Pants project contributors (see CONTRIBUTORS.md).
2	# Licensed under the Apache License, Version 2.0 (see LICENSE).
3
UNCOV 4	from __future__ import annotations	×
5
UNCOV 6	from abc import ABC, abstractmethod	×
UNCOV 7	from collections.abc import Iterable	×
UNCOV 8	from dataclasses import dataclass	×
UNCOV 9	from typing import Any, cast	×
10
UNCOV 11	from pants.backend.helm.subsystems.helm import HelmSubsystem	×
UNCOV 12	from pants.backend.helm.target_types import (	×
13	AllHelmArtifactTargets,
14	HelmArtifactFieldSet,
15	HelmArtifactRegistryField,
16	HelmArtifactRepositoryField,
17	HelmChartTarget,
18	)
UNCOV 19	from pants.backend.helm.util_rules.chart_metadata import rules as metadata_rules	×
UNCOV 20	from pants.engine.addresses import Address	×
UNCOV 21	from pants.engine.engine_aware import EngineAwareReturnType	×
UNCOV 22	from pants.engine.rules import collect_rules, concurrently, implicitly, rule	×
UNCOV 23	from pants.engine.target import Target	×
UNCOV 24	from pants.util.frozendict import FrozenDict	×
UNCOV 25	from pants.util.strutil import bullet_list	×
26
27
UNCOV 28	class MissingHelmArtifactLocation(ValueError):	×
UNCOV 29	def __init__(self, address: Address) -> None:	×
30	super().__init__(	×
31	f"Target at address '{address}' needs to specify either `{HelmArtifactRegistryField.alias}`, "
32	f"`{HelmArtifactRepositoryField.alias}` or both."
33	)
34
35
UNCOV 36	class DuplicateHelmChartNamesFound(Exception):	×
UNCOV 37	def __init__(self, duplicates: Iterable[tuple[str, Address]]) -> None:	×
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
UNCOV 44	class HelmArtifactLocationSpec(ABC):	×
45	@property
46	@abstractmethod
47	def spec(self) -> str: ...
48
UNCOV 49	@property	×
UNCOV 50	def is_url(self) -> bool:	×
51	return len(self.spec.split("://")) == 2	×
52
UNCOV 53	@property	×
UNCOV 54	def is_alias(self) -> bool:	×
55	return self.spec.startswith("@")	×
56
57
UNCOV 58	@dataclass(frozen=True)	×
UNCOV 59	class HelmArtifactRegistryLocationSpec(HelmArtifactLocationSpec):	×
UNCOV 60	registry: str	×
UNCOV 61	repository: str \| None	×
62
UNCOV 63	@property	×
UNCOV 64	def spec(self) -> str:	×
65	return self.registry	×
66
67
UNCOV 68	@dataclass(frozen=True)	×
UNCOV 69	class HelmArtifactClassicRepositoryLocationSpec(HelmArtifactLocationSpec):	×
UNCOV 70	repository: str	×
71
UNCOV 72	@property	×
UNCOV 73	def spec(self) -> str:	×
74	return self.repository	×
75
76
UNCOV 77	@dataclass(frozen=True)	×
UNCOV 78	class HelmArtifactRequirement:	×
UNCOV 79	name: str	×
UNCOV 80	version: str	×
UNCOV 81	location: HelmArtifactLocationSpec	×
82
83
UNCOV 84	@dataclass(frozen=True)	×
UNCOV 85	class HelmArtifact:	×
UNCOV 86	requirement: HelmArtifactRequirement	×
UNCOV 87	address: Address	×
88
UNCOV 89	@classmethod	×
UNCOV 90	def from_target(cls, target: Target) -> HelmArtifact:	×
UNCOV 91	return cls.from_field_set(HelmArtifactFieldSet.create(target))	×
92
UNCOV 93	@classmethod	×
UNCOV 94	def from_field_set(cls, field_set: HelmArtifactFieldSet) -> HelmArtifact:	×
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
UNCOV 115	@property	×
UNCOV 116	def name(self) -> str:	×
117	return self.requirement.name	×
118
UNCOV 119	@property	×
UNCOV 120	def version(self) -> str:	×
121	return self.requirement.version	×
122
123
UNCOV 124	@dataclass(frozen=True)	×
UNCOV 125	class ResolvedHelmArtifact(HelmArtifact, EngineAwareReturnType):	×
UNCOV 126	location_url: str	×
127
UNCOV 128	@classmethod	×
UNCOV 129	def from_unresolved(cls, artifact: HelmArtifact, *, location_url: str) -> ResolvedHelmArtifact:	×
130	return cls(	×
131	requirement=artifact.requirement,
132	address=artifact.address,
133	location_url=location_url,
134	)
135
UNCOV 136	@property	×
UNCOV 137	def chart_url(self) -> str:	×
138	return f"{self.location_url}/{self.name}"	×
139
UNCOV 140	def metadata(self) -> dict[str, Any] \| None:	×
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
UNCOV 150	@rule	×
UNCOV 151	async def resolved_helm_artifact(	×
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
UNCOV 167	class ThirdPartyHelmArtifactMapping(FrozenDict[str, Address]):	×
UNCOV 168	pass	×
169
170
UNCOV 171	@rule	×
UNCOV 172	async def third_party_helm_artifact_mapping(	×
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
UNCOV 184	def rules():	×
UNCOV 185	return [collect_rules(), metadata_rules()]	×

pantsbuild / pants / 19015773527

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