20332790708

Committed 18 Dec 2025 09:48AM UTC coverage: 64.992% (-15.3%) from 80.295%

Build # 20332790708

Build Type

Pull #22949

github

Committed by

web-flow

Commit Message

Merge f730a56cd into 407284c67

Pull Request Pull Request #22949: Add experimental uv resolver for Python lockfiles

Run Details

54 of 97 new or added lines in 5 files covered. (55.67%)

8270 existing lines in 295 files now uncovered.

48990 of 75379 relevant lines covered (64.99%)

1.81 hits per line

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

52.33

/src/python/pants/backend/helm/util_rules/chart_metadata.py

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

from __future__ import annotations

import dataclasses
from dataclasses import dataclass
from enum import Enum
from typing import Any

import yaml

from pants.backend.helm.target_types import HelmChartMetaSourceField
from pants.backend.helm.util_rules.sources import HelmChartRootRequest, find_chart_source_root
from pants.backend.helm.utils.yaml import snake_case_attr_dict
from pants.base.glob_match_error_behavior import GlobMatchErrorBehavior
from pants.engine.engine_aware import EngineAwareParameter
from pants.engine.fs import (
    CreateDigest,
    Digest,
    DigestSubset,
    FileContent,
    GlobExpansionConjunction,
    PathGlobs,
)
from pants.engine.internals.graph import hydrate_sources
from pants.engine.internals.native_engine import RemovePrefix
from pants.engine.internals.selectors import concurrently
from pants.engine.intrinsics import (
    create_digest,
    digest_subset_to_digest,
    get_digest_contents,
    remove_prefix,
)
from pants.engine.rules import collect_rules, implicitly, rule
from pants.engine.target import HydrateSourcesRequest
from pants.util.frozendict import FrozenDict
from pants.util.strutil import bullet_list


class ChartType(Enum):
    """Type of Helm Chart."""

    APPLICATION = "application"
    LIBRARY = "library"


class InvalidChartTypeValueError(ValueError):
    def __init__(self, value: str) -> None:
        super().__init__(
            f"Invalid value '{value}' for Helm Chart `type`. Valid values are: {[t.value for t in list(ChartType)]}"
        )


class MissingChartMetadataException(Exception):
    pass


class AmbiguousChartMetadataException(Exception):
    pass


@dataclass(frozen=True)
class HelmChartDependency:
    name: str
    repository: str | None = None
    version: str | None = None
    alias: str | None = None
    condition: str | None = None
    tags: tuple[str, ...] = dataclasses.field(default_factory=tuple)
    import_values: tuple[str, ...] = dataclasses.field(default_factory=tuple)

    @classmethod
    def from_dict(cls, d: dict[str, Any]) -> HelmChartDependency:
        attrs = snake_case_attr_dict(d)

        tags = attrs.pop("tags", [])
        import_values = attrs.pop("import_values", [])

        return cls(tags=tuple(tags), import_values=tuple(import_values), **attrs)

    def to_json_dict(self) -> dict[str, Any]:
        d: dict[str, Any] = {"name": self.name}
        if self.repository:
            d["repository"] = self.repository.rstrip("/")
        if self.version:
            d["version"] = self.version
        if self.alias:
            d["alias"] = self.alias
        if self.condition:
            d["condition"] = self.condition
        if self.tags:
            d["tags"] = list(self.tags)
        if self.import_values:
            d["import-values"] = list(self.import_values)
        return d


@dataclass(frozen=True)
class HelmChartMaintainer:
    name: str
    email: str | None = None
    url: str | None = None

    @classmethod
    def from_dict(cls, d: dict[str, Any]) -> HelmChartMaintainer:
        return cls(**d)

    def to_json_dict(self) -> dict[str, Any]:
        d = {"name": self.name}
        if self.email:
            d["email"] = self.email
        if self.url:
            d["url"] = self.url
        return d


DEFAULT_API_VERSION = "v2"


@dataclass(frozen=True)
class HelmChartMetadata:
    name: str
    version: str
    api_version: str = DEFAULT_API_VERSION
    type: ChartType = ChartType.APPLICATION
    kube_version: str | None = None
    app_version: str | None = None
    icon: str | None = None
    description: str | None = None
    dependencies: tuple[HelmChartDependency, ...] = dataclasses.field(default_factory=tuple)
    keywords: tuple[str, ...] = dataclasses.field(default_factory=tuple)
    sources: tuple[str, ...] = dataclasses.field(default_factory=tuple)
    home: str | None = None
    maintainers: tuple[HelmChartMaintainer, ...] = dataclasses.field(default_factory=tuple)
    deprecated: bool | None = None
    annotations: FrozenDict[str, str] = dataclasses.field(default_factory=FrozenDict)

    @classmethod
    def from_dict(cls, d: dict[str, Any]) -> HelmChartMetadata:
        chart_type: ChartType | None = None
        type_str = d.pop("type", None)
        if type_str:
            try:
                chart_type = ChartType(type_str)
            except KeyError:
                raise InvalidChartTypeValueError(type_str)

        # If the `apiVersion` is missing in the original `dict`, then we assume we are dealing with `v1` charts.
        api_version = d.pop("apiVersion", "v1")
        dependencies = [HelmChartDependency.from_dict(d) for d in d.pop("dependencies", [])]
        maintainers = [HelmChartMaintainer.from_dict(d) for d in d.pop("maintainers", [])]
        keywords = d.pop("keywords", [])
        sources = d.pop("sources", [])
        annotations = d.pop("annotations", {})

        attrs = snake_case_attr_dict(d)

        return cls(
            api_version=api_version,
            dependencies=tuple(dependencies),
            maintainers=tuple(maintainers),
            keywords=tuple(keywords),
            type=chart_type or ChartType.APPLICATION,
            annotations=FrozenDict(annotations),
            sources=tuple(sources),
            **attrs,
        )

    @classmethod
    def from_bytes(cls, content: bytes) -> HelmChartMetadata:
        return cls.from_dict(yaml.safe_load(content))

    @property
    def artifact_name(self) -> str:
        return f"{self.name}-{self.version}"

    def to_json_dict(self) -> dict[str, Any]:
        d: dict[str, Any] = {
            "apiVersion": self.api_version,
            "name": self.name,
            "version": self.version,
        }
        if self.api_version != "v1":
            d["type"] = self.type.value
        if self.icon:
            d["icon"] = self.icon
        if self.description:
            d["description"] = self.description
        if self.app_version:
            d["appVersion"] = self.app_version
        if self.kube_version:
            d["kubeVersion"] = self.kube_version
        if self.dependencies:
            d["dependencies"] = [item.to_json_dict() for item in self.dependencies]
        if self.maintainers:
            d["maintainers"] = [item.to_json_dict() for item in self.maintainers]
        if self.annotations:
            d["annotations"] = dict(self.annotations.items())
        if self.keywords:
            d["keywords"] = list(self.keywords)
        if self.sources:
            d["sources"] = list(self.sources)
        if self.home:
            d["home"] = self.home
        if self.deprecated:
            d["deprecated"] = self.deprecated
        return d

    def to_yaml(self) -> str:
        return yaml.dump(self.to_json_dict())


HELM_CHART_METADATA_FILENAMES = ["Chart.yaml", "Chart.yml"]


@dataclass(frozen=True)
class ParseHelmChartMetadataDigest(EngineAwareParameter):
    """Request to parse the Helm chart definition file (i.e. `Chart.yaml`) from the given digest.

    The definition file is expected to be at the root of the digest.
    """

    digest: Digest
    description_of_origin: str

    def debug_hint(self) -> str | None:
        return self.description_of_origin


@rule
async def parse_chart_metadata_from_digest(
    request: ParseHelmChartMetadataDigest,
) -> HelmChartMetadata:
    subset = await digest_subset_to_digest(
        DigestSubset(
            request.digest,
            PathGlobs(
                HELM_CHART_METADATA_FILENAMES,
                glob_match_error_behavior=GlobMatchErrorBehavior.error,
                conjunction=GlobExpansionConjunction.any_match,
                description_of_origin=request.description_of_origin,
            ),
        )
    )

    file_contents = await get_digest_contents(subset)

    if len(file_contents) == 0:
        raise MissingChartMetadataException(
            f"Could not find any file that matched with either {HELM_CHART_METADATA_FILENAMES} in target at {request.description_of_origin}."
        )
    if len(file_contents) > 1:
        raise AmbiguousChartMetadataException(
            f"Found more than one Helm chart metadata file at '{request.description_of_origin}':\n{bullet_list([f.path for f in file_contents])}"
        )

    return HelmChartMetadata.from_bytes(file_contents[0].content)


@rule
async def parse_chart_metadata_from_field(field: HelmChartMetaSourceField) -> HelmChartMetadata:
    chart_root, source_files = await concurrently(
        find_chart_source_root(HelmChartRootRequest(field)),
        hydrate_sources(
            HydrateSourcesRequest(
                field, for_sources_types=(HelmChartMetaSourceField,), enable_codegen=True
            ),
            **implicitly(),
        ),
    )

    metadata_digest = await remove_prefix(
        RemovePrefix(source_files.snapshot.digest, chart_root.path)
    )

    return await parse_chart_metadata_from_digest(
        ParseHelmChartMetadataDigest(
            metadata_digest,
            description_of_origin=f"the `helm_chart` {field.address.spec}",
        )
    )


@rule
async def render_chart_metadata(metadata: HelmChartMetadata) -> Digest:
    yaml_contents = bytes(metadata.to_yaml(), "utf-8")
    return await create_digest(
        CreateDigest([FileContent(HELM_CHART_METADATA_FILENAMES[0], yaml_contents)])
    )


def rules():
    return collect_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	5✔
5
6	import dataclasses	5✔
7	from dataclasses import dataclass	5✔
8	from enum import Enum	5✔
9	from typing import Any	5✔
10
11	import yaml	5✔
12
13	from pants.backend.helm.target_types import HelmChartMetaSourceField	5✔
14	from pants.backend.helm.util_rules.sources import HelmChartRootRequest, find_chart_source_root	5✔
15	from pants.backend.helm.utils.yaml import snake_case_attr_dict	5✔
16	from pants.base.glob_match_error_behavior import GlobMatchErrorBehavior	5✔
17	from pants.engine.engine_aware import EngineAwareParameter	5✔
18	from pants.engine.fs import (	5✔
19	CreateDigest,
20	Digest,
21	DigestSubset,
22	FileContent,
23	GlobExpansionConjunction,
24	PathGlobs,
25	)
26	from pants.engine.internals.graph import hydrate_sources	5✔
27	from pants.engine.internals.native_engine import RemovePrefix	5✔
28	from pants.engine.internals.selectors import concurrently	5✔
29	from pants.engine.intrinsics import (	5✔
30	create_digest,
31	digest_subset_to_digest,
32	get_digest_contents,
33	remove_prefix,
34	)
35	from pants.engine.rules import collect_rules, implicitly, rule	5✔
36	from pants.engine.target import HydrateSourcesRequest	5✔
37	from pants.util.frozendict import FrozenDict	5✔
38	from pants.util.strutil import bullet_list	5✔
39
40
41	class ChartType(Enum):	5✔
42	"""Type of Helm Chart."""
43
44	APPLICATION = "application"	5✔
45	LIBRARY = "library"	5✔
46
47
48	class InvalidChartTypeValueError(ValueError):	5✔
49	def __init__(self, value: str) -> None:	5✔
50	super().__init__(	×
51	f"Invalid value '{value}' for Helm Chart `type`. Valid values are: {[t.value for t in list(ChartType)]}"
52	)
53
54
55	class MissingChartMetadataException(Exception):	5✔
56	pass	5✔
57
58
59	class AmbiguousChartMetadataException(Exception):	5✔
60	pass	5✔
61
62
63	@dataclass(frozen=True)	5✔
64	class HelmChartDependency:	5✔
65	name: str	5✔
66	repository: str \| None = None	5✔
67	version: str \| None = None	5✔
68	alias: str \| None = None	5✔
69	condition: str \| None = None	5✔
70	tags: tuple[str, ...] = dataclasses.field(default_factory=tuple)	5✔
71	import_values: tuple[str, ...] = dataclasses.field(default_factory=tuple)	5✔
72
73	@classmethod	5✔
74	def from_dict(cls, d: dict[str, Any]) -> HelmChartDependency:	5✔
UNCOV 75	attrs = snake_case_attr_dict(d)	×
76
UNCOV 77	tags = attrs.pop("tags", [])	×
UNCOV 78	import_values = attrs.pop("import_values", [])	×
79
UNCOV 80	return cls(tags=tuple(tags), import_values=tuple(import_values), **attrs)	×
81
82	def to_json_dict(self) -> dict[str, Any]:	5✔
UNCOV 83	d: dict[str, Any] = {"name": self.name}	×
UNCOV 84	if self.repository:	×
UNCOV 85	d["repository"] = self.repository.rstrip("/")	×
UNCOV 86	if self.version:	×
UNCOV 87	d["version"] = self.version	×
UNCOV 88	if self.alias:	×
UNCOV 89	d["alias"] = self.alias	×
UNCOV 90	if self.condition:	×
UNCOV 91	d["condition"] = self.condition	×
UNCOV 92	if self.tags:	×
UNCOV 93	d["tags"] = list(self.tags)	×
UNCOV 94	if self.import_values:	×
UNCOV 95	d["import-values"] = list(self.import_values)	×
UNCOV 96	return d	×
97
98
99	@dataclass(frozen=True)	5✔
100	class HelmChartMaintainer:	5✔
101	name: str	5✔
102	email: str \| None = None	5✔
103	url: str \| None = None	5✔
104
105	@classmethod	5✔
106	def from_dict(cls, d: dict[str, Any]) -> HelmChartMaintainer:	5✔
UNCOV 107	return cls(**d)	×
108
109	def to_json_dict(self) -> dict[str, Any]:	5✔
UNCOV 110	d = {"name": self.name}	×
UNCOV 111	if self.email:	×
UNCOV 112	d["email"] = self.email	×
UNCOV 113	if self.url:	×
UNCOV 114	d["url"] = self.url	×
UNCOV 115	return d	×
116
117
118	DEFAULT_API_VERSION = "v2"	5✔
119
120
121	@dataclass(frozen=True)	5✔
122	class HelmChartMetadata:	5✔
123	name: str	5✔
124	version: str	5✔
125	api_version: str = DEFAULT_API_VERSION	5✔
126	type: ChartType = ChartType.APPLICATION	5✔
127	kube_version: str \| None = None	5✔
128	app_version: str \| None = None	5✔
129	icon: str \| None = None	5✔
130	description: str \| None = None	5✔
131	dependencies: tuple[HelmChartDependency, ...] = dataclasses.field(default_factory=tuple)	5✔
132	keywords: tuple[str, ...] = dataclasses.field(default_factory=tuple)	5✔
133	sources: tuple[str, ...] = dataclasses.field(default_factory=tuple)	5✔
134	home: str \| None = None	5✔
135	maintainers: tuple[HelmChartMaintainer, ...] = dataclasses.field(default_factory=tuple)	5✔
136	deprecated: bool \| None = None	5✔
137	annotations: FrozenDict[str, str] = dataclasses.field(default_factory=FrozenDict)	5✔
138
139	@classmethod	5✔
140	def from_dict(cls, d: dict[str, Any]) -> HelmChartMetadata:	5✔
UNCOV 141	chart_type: ChartType \| None = None	×
UNCOV 142	type_str = d.pop("type", None)	×
UNCOV 143	if type_str:	×
UNCOV 144	try:	×
UNCOV 145	chart_type = ChartType(type_str)	×
146	except KeyError:	×
147	raise InvalidChartTypeValueError(type_str)	×
148
149	# If the `apiVersion` is missing in the original `dict`, then we assume we are dealing with `v1` charts.
UNCOV 150	api_version = d.pop("apiVersion", "v1")	×
UNCOV 151	dependencies = [HelmChartDependency.from_dict(d) for d in d.pop("dependencies", [])]	×
UNCOV 152	maintainers = [HelmChartMaintainer.from_dict(d) for d in d.pop("maintainers", [])]	×
UNCOV 153	keywords = d.pop("keywords", [])	×
UNCOV 154	sources = d.pop("sources", [])	×
UNCOV 155	annotations = d.pop("annotations", {})	×
156
UNCOV 157	attrs = snake_case_attr_dict(d)	×
158
UNCOV 159	return cls(	×
160	api_version=api_version,
161	dependencies=tuple(dependencies),
162	maintainers=tuple(maintainers),
163	keywords=tuple(keywords),
164	type=chart_type or ChartType.APPLICATION,
165	annotations=FrozenDict(annotations),
166	sources=tuple(sources),
167	**attrs,
168	)
169
170	@classmethod	5✔
171	def from_bytes(cls, content: bytes) -> HelmChartMetadata:	5✔
UNCOV 172	return cls.from_dict(yaml.safe_load(content))	×
173
174	@property	5✔
175	def artifact_name(self) -> str:	5✔
176	return f"{self.name}-{self.version}"	1✔
177
178	def to_json_dict(self) -> dict[str, Any]:	5✔
UNCOV 179	d: dict[str, Any] = {	×
180	"apiVersion": self.api_version,
181	"name": self.name,
182	"version": self.version,
183	}
UNCOV 184	if self.api_version != "v1":	×
UNCOV 185	d["type"] = self.type.value	×
UNCOV 186	if self.icon:	×
UNCOV 187	d["icon"] = self.icon	×
UNCOV 188	if self.description:	×
UNCOV 189	d["description"] = self.description	×
UNCOV 190	if self.app_version:	×
UNCOV 191	d["appVersion"] = self.app_version	×
UNCOV 192	if self.kube_version:	×
UNCOV 193	d["kubeVersion"] = self.kube_version	×
UNCOV 194	if self.dependencies:	×
UNCOV 195	d["dependencies"] = [item.to_json_dict() for item in self.dependencies]	×
UNCOV 196	if self.maintainers:	×
UNCOV 197	d["maintainers"] = [item.to_json_dict() for item in self.maintainers]	×
UNCOV 198	if self.annotations:	×
UNCOV 199	d["annotations"] = dict(self.annotations.items())	×
UNCOV 200	if self.keywords:	×
UNCOV 201	d["keywords"] = list(self.keywords)	×
UNCOV 202	if self.sources:	×
UNCOV 203	d["sources"] = list(self.sources)	×
UNCOV 204	if self.home:	×
UNCOV 205	d["home"] = self.home	×
UNCOV 206	if self.deprecated:	×
UNCOV 207	d["deprecated"] = self.deprecated	×
UNCOV 208	return d	×
209
210	def to_yaml(self) -> str:	5✔
211	return yaml.dump(self.to_json_dict())	×
212
213
214	HELM_CHART_METADATA_FILENAMES = ["Chart.yaml", "Chart.yml"]	5✔
215
216
217	@dataclass(frozen=True)	5✔
218	class ParseHelmChartMetadataDigest(EngineAwareParameter):	5✔
219	"""Request to parse the Helm chart definition file (i.e. `Chart.yaml`) from the given digest.
220
221	The definition file is expected to be at the root of the digest.
222	"""
223
224	digest: Digest	5✔
225	description_of_origin: str	5✔
226
227	def debug_hint(self) -> str \| None:	5✔
228	return self.description_of_origin	×
229
230
231	@rule	5✔
232	async def parse_chart_metadata_from_digest(	5✔
233	request: ParseHelmChartMetadataDigest,
234	) -> HelmChartMetadata:
235	subset = await digest_subset_to_digest(	×
236	DigestSubset(
237	request.digest,
238	PathGlobs(
239	HELM_CHART_METADATA_FILENAMES,
240	glob_match_error_behavior=GlobMatchErrorBehavior.error,
241	conjunction=GlobExpansionConjunction.any_match,
242	description_of_origin=request.description_of_origin,
243	),
244	)
245	)
246
247	file_contents = await get_digest_contents(subset)	×
248
249	if len(file_contents) == 0:	×
250	raise MissingChartMetadataException(	×
251	f"Could not find any file that matched with either {HELM_CHART_METADATA_FILENAMES} in target at {request.description_of_origin}."
252	)
253	if len(file_contents) > 1:	×
254	raise AmbiguousChartMetadataException(	×
255	f"Found more than one Helm chart metadata file at '{request.description_of_origin}':\n{bullet_list([f.path for f in file_contents])}"
256	)
257
258	return HelmChartMetadata.from_bytes(file_contents[0].content)	×
259
260
261	@rule	5✔
262	async def parse_chart_metadata_from_field(field: HelmChartMetaSourceField) -> HelmChartMetadata:	5✔
263	chart_root, source_files = await concurrently(	×
264	find_chart_source_root(HelmChartRootRequest(field)),
265	hydrate_sources(
266	HydrateSourcesRequest(
267	field, for_sources_types=(HelmChartMetaSourceField,), enable_codegen=True
268	),
269	**implicitly(),
270	),
271	)
272
273	metadata_digest = await remove_prefix(	×
274	RemovePrefix(source_files.snapshot.digest, chart_root.path)
275	)
276
277	return await parse_chart_metadata_from_digest(	×
278	ParseHelmChartMetadataDigest(
279	metadata_digest,
280	description_of_origin=f"the `helm_chart` {field.address.spec}",
281	)
282	)
283
284
285	@rule	5✔
286	async def render_chart_metadata(metadata: HelmChartMetadata) -> Digest:	5✔
287	yaml_contents = bytes(metadata.to_yaml(), "utf-8")	×
288	return await create_digest(	×
289	CreateDigest([FileContent(HELM_CHART_METADATA_FILENAMES[0], yaml_contents)])
290	)
291
292
293	def rules():	5✔
294	return collect_rules()	4✔

pantsbuild / pants / 20332790708

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