19529437518

Committed 20 Nov 2025 07:44AM UTC coverage: 78.884% (-1.4%) from 80.302%

Build # 19529437518

Build Type

push

github

Committed by

web-flow

Commit Message

nfpm.native_libs: Add RPM package depends from packaged pex_binaries (#22899)

## PR Series Overview

This is the second in a series of PRs that introduces a new backend:
`pants.backend.npm.native_libs`
Initially, the backend will be available as:
`pants.backend.experimental.nfpm.native_libs`

I proposed this new backend (originally named `bindeps`) in discussion
#22396.

This backend will inspect ELF bin/lib files (like `lib*.so`) in packaged
contents (for this PR series, only in `pex_binary` targets) to identify
package dependency metadata and inject that metadata on the relevant
`nfpm_deb_package` or `nfpm_rpm_package` targets. Effectively, it will
provide an approximation of these native packager features:
- `rpm`: `rpmdeps` + `elfdeps`
- `deb`: `dh_shlibdeps` + `dpkg-shlibdeps` (These substitute
`${shlibs:Depends}` in debian control files have)

### Goal: Host-agnostic package builds

This pants backend is designed to be host-agnostic, like
[nFPM](https://nfpm.goreleaser.com/).

Native packaging tools are often restricted to a single release of a
single distro. Unlike native package builders, this new pants backend
does not use any of those distro-specific or distro-release-specific
utilities or local package databases. This new backend should be able to
run (help with building deb and rpm packages) anywhere that pants can
run (MacOS, rpm linux distros, deb linux distros, other linux distros,
docker, ...).

### Previous PRs in series

- #22873

## PR Overview

This PR adds rules in `nfpm.native_libs` to add package dependency
metadata to `nfpm_rpm_package`. The 2 new rules are:

- `inject_native_libs_dependencies_in_package_fields`:

    - An implementation of the polymorphic rule `inject_nfpm_package_fields`.
      This rule is low priority (`priority = 2`) so that in-repo plugins can
      override/augment what it injects. (See #22864)

    - Rule logic overview:
        - find any pex_binaries that will be packaged in an `nfpm_rpm_package`
   ... (continued)

Run Details

96 of 118 new or added lines in 3 files covered. (81.36%)

910 existing lines in 53 files now uncovered.

73897 of 93678 relevant lines covered (78.88%)

3.21 hits per line

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

89.53

/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	9✔
5
6	import dataclasses	9✔
7	from dataclasses import dataclass	9✔
8	from enum import Enum	9✔
9	from typing import Any	9✔
10
11	import yaml	9✔
12
13	from pants.backend.helm.target_types import HelmChartMetaSourceField	9✔
14	from pants.backend.helm.util_rules.sources import HelmChartRootRequest, find_chart_source_root	9✔
15	from pants.backend.helm.utils.yaml import snake_case_attr_dict	9✔
16	from pants.base.glob_match_error_behavior import GlobMatchErrorBehavior	9✔
17	from pants.engine.engine_aware import EngineAwareParameter	9✔
18	from pants.engine.fs import (	9✔
19	CreateDigest,
20	Digest,
21	DigestSubset,
22	FileContent,
23	GlobExpansionConjunction,
24	PathGlobs,
25	)
26	from pants.engine.internals.graph import hydrate_sources	9✔
27	from pants.engine.internals.native_engine import RemovePrefix	9✔
28	from pants.engine.internals.selectors import concurrently	9✔
29	from pants.engine.intrinsics import (	9✔
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	9✔
36	from pants.engine.target import HydrateSourcesRequest	9✔
37	from pants.util.frozendict import FrozenDict	9✔
38	from pants.util.strutil import bullet_list	9✔
39
40
41	class ChartType(Enum):	9✔
42	"""Type of Helm Chart."""
43
44	APPLICATION = "application"	9✔
45	LIBRARY = "library"	9✔
46
47
48	class InvalidChartTypeValueError(ValueError):	9✔
49	def __init__(self, value: str) -> None:	9✔
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):	9✔
56	pass	9✔
57
58
59	class AmbiguousChartMetadataException(Exception):	9✔
60	pass	9✔
61
62
63	@dataclass(frozen=True)	9✔
64	class HelmChartDependency:	9✔
65	name: str	9✔
66	repository: str \| None = None	9✔
67	version: str \| None = None	9✔
68	alias: str \| None = None	9✔
69	condition: str \| None = None	9✔
70	tags: tuple[str, ...] = dataclasses.field(default_factory=tuple)	9✔
71	import_values: tuple[str, ...] = dataclasses.field(default_factory=tuple)	9✔
72
73	@classmethod	9✔
74	def from_dict(cls, d: dict[str, Any]) -> HelmChartDependency:	9✔
75	attrs = snake_case_attr_dict(d)	1✔
76
77	tags = attrs.pop("tags", [])	1✔
78	import_values = attrs.pop("import_values", [])	1✔
79
80	return cls(tags=tuple(tags), import_values=tuple(import_values), **attrs)	1✔
81
82	def to_json_dict(self) -> dict[str, Any]:	9✔
83	d: dict[str, Any] = {"name": self.name}	1✔
84	if self.repository:	1✔
85	d["repository"] = self.repository.rstrip("/")	1✔
86	if self.version:	1✔
87	d["version"] = self.version	1✔
88	if self.alias:	1✔
89	d["alias"] = self.alias	1✔
90	if self.condition:	1✔
91	d["condition"] = self.condition	1✔
92	if self.tags:	1✔
93	d["tags"] = list(self.tags)	1✔
94	if self.import_values:	1✔
95	d["import-values"] = list(self.import_values)	1✔
96	return d	1✔
97
98
99	@dataclass(frozen=True)	9✔
100	class HelmChartMaintainer:	9✔
101	name: str	9✔
102	email: str \| None = None	9✔
103	url: str \| None = None	9✔
104
105	@classmethod	9✔
106	def from_dict(cls, d: dict[str, Any]) -> HelmChartMaintainer:	9✔
107	return cls(**d)	1✔
108
109	def to_json_dict(self) -> dict[str, Any]:	9✔
110	d = {"name": self.name}	1✔
111	if self.email:	1✔
112	d["email"] = self.email	1✔
113	if self.url:	1✔
114	d["url"] = self.url	1✔
115	return d	1✔
116
117
118	DEFAULT_API_VERSION = "v2"	9✔
119
120
121	@dataclass(frozen=True)	9✔
122	class HelmChartMetadata:	9✔
123	name: str	9✔
124	version: str	9✔
125	api_version: str = DEFAULT_API_VERSION	9✔
126	type: ChartType = ChartType.APPLICATION	9✔
127	kube_version: str \| None = None	9✔
128	app_version: str \| None = None	9✔
129	icon: str \| None = None	9✔
130	description: str \| None = None	9✔
131	dependencies: tuple[HelmChartDependency, ...] = dataclasses.field(default_factory=tuple)	9✔
132	keywords: tuple[str, ...] = dataclasses.field(default_factory=tuple)	9✔
133	sources: tuple[str, ...] = dataclasses.field(default_factory=tuple)	9✔
134	home: str \| None = None	9✔
135	maintainers: tuple[HelmChartMaintainer, ...] = dataclasses.field(default_factory=tuple)	9✔
136	deprecated: bool \| None = None	9✔
137	annotations: FrozenDict[str, str] = dataclasses.field(default_factory=FrozenDict)	9✔
138
139	@classmethod	9✔
140	def from_dict(cls, d: dict[str, Any]) -> HelmChartMetadata:	9✔
141	chart_type: ChartType \| None = None	1✔
142	type_str = d.pop("type", None)	1✔
143	if type_str:	1✔
144	try:	1✔
145	chart_type = ChartType(type_str)	1✔
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.
150	api_version = d.pop("apiVersion", "v1")	1✔
151	dependencies = [HelmChartDependency.from_dict(d) for d in d.pop("dependencies", [])]	1✔
152	maintainers = [HelmChartMaintainer.from_dict(d) for d in d.pop("maintainers", [])]	1✔
153	keywords = d.pop("keywords", [])	1✔
154	sources = d.pop("sources", [])	1✔
155	annotations = d.pop("annotations", {})	1✔
156
157	attrs = snake_case_attr_dict(d)	1✔
158
159	return cls(	1✔
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	9✔
171	def from_bytes(cls, content: bytes) -> HelmChartMetadata:	9✔
172	return cls.from_dict(yaml.safe_load(content))	1✔
173
174	@property	9✔
175	def artifact_name(self) -> str:	9✔
UNCOV 176	return f"{self.name}-{self.version}"	×
177
178	def to_json_dict(self) -> dict[str, Any]:	9✔
179	d: dict[str, Any] = {	1✔
180	"apiVersion": self.api_version,
181	"name": self.name,
182	"version": self.version,
183	}
184	if self.api_version != "v1":	1✔
185	d["type"] = self.type.value	1✔
186	if self.icon:	1✔
187	d["icon"] = self.icon	1✔
188	if self.description:	1✔
189	d["description"] = self.description	1✔
190	if self.app_version:	1✔
191	d["appVersion"] = self.app_version	1✔
192	if self.kube_version:	1✔
193	d["kubeVersion"] = self.kube_version	1✔
194	if self.dependencies:	1✔
195	d["dependencies"] = [item.to_json_dict() for item in self.dependencies]	1✔
196	if self.maintainers:	1✔
197	d["maintainers"] = [item.to_json_dict() for item in self.maintainers]	1✔
198	if self.annotations:	1✔
199	d["annotations"] = dict(self.annotations.items())	1✔
200	if self.keywords:	1✔
201	d["keywords"] = list(self.keywords)	1✔
202	if self.sources:	1✔
203	d["sources"] = list(self.sources)	1✔
204	if self.home:	1✔
205	d["home"] = self.home	1✔
206	if self.deprecated:	1✔
207	d["deprecated"] = self.deprecated	1✔
208	return d	1✔
209
210	def to_yaml(self) -> str:	9✔
211	return yaml.dump(self.to_json_dict())	×
212
213
214	HELM_CHART_METADATA_FILENAMES = ["Chart.yaml", "Chart.yml"]	9✔
215
216
217	@dataclass(frozen=True)	9✔
218	class ParseHelmChartMetadataDigest(EngineAwareParameter):	9✔
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	9✔
225	description_of_origin: str	9✔
226
227	def debug_hint(self) -> str \| None:	9✔
228	return self.description_of_origin	×
229
230
231	@rule	9✔
232	async def parse_chart_metadata_from_digest(	9✔
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	9✔
262	async def parse_chart_metadata_from_field(field: HelmChartMetaSourceField) -> HelmChartMetadata:	9✔
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	9✔
286	async def render_chart_metadata(metadata: HelmChartMetadata) -> Digest:	9✔
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():	9✔
294	return collect_rules()	8✔

pantsbuild / pants / 19529437518

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