19123999670

Committed 06 Nov 2025 03:38AM UTC coverage: 80.307% (+0.007%) from 80.3%

Build # 19123999670

Build Type

Pull #22861

github

Committed by

web-flow

Commit Message

Merge d8b6795f2 into 89462b7ef

Pull Request Pull Request #22861: nfpm.native_libs: new backend to generate pkg deps for nfpm packages

Coverage Stats

613 of 767 new or added lines in 18 files covered. (79.92%)

1 existing line in 1 file now uncovered.

78600 of 97875 relevant lines covered (80.31%)

3.35 hits per line

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

61.63

/src/python/pants/backend/nfpm/native_libs/rules.py

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

from __future__ import annotations

from collections.abc import Iterable
from dataclasses import dataclass

from pants.backend.nfpm.field_sets import NfpmDebPackageFieldSet, NfpmRpmPackageFieldSet
from pants.backend.nfpm.fields.all import NfpmArchField
from pants.backend.nfpm.fields.deb import NfpmDebDependsField
from pants.backend.nfpm.fields.rpm import NfpmRpmDependsField, NfpmRpmProvidesField
from pants.backend.nfpm.util_rules.contents import (
    GetPackageFieldSetsForNfpmContentFileDepsRequest,
    get_package_field_sets_for_nfpm_content_file_deps,
)
from pants.backend.nfpm.util_rules.inject_config import (
    InjectedNfpmPackageFields,
    InjectNfpmPackageFieldsRequest,
)
from pants.backend.python.goals.package_pex_binary import PexBinaryFieldSet, package_pex_binary
from pants.backend.python.util_rules.pex import Pex, create_pex
from pants.backend.python.util_rules.pex_from_targets import PexFromTargetsRequest
from pants.engine.addresses import Address
from pants.engine.internals.selectors import concurrently
from pants.engine.rules import Rule, collect_rules, implicitly, rule
from pants.engine.target import Field, Target
from pants.engine.unions import UnionMembership, UnionRule

from .deb.rules import DebSearchForSonamesRequest, deb_search_for_sonames
from .deb.rules import rules as deb_rules
from .deb.utils import shlibdeps_filter_sonames
from .elfdeps.rules import RequestPexELFInfo, elfdeps_analyze_pex_wheels
from .elfdeps.rules import rules as elfdeps_rules
from .target_types import DebDistroCodenameField, DebDistroField
from .target_types import rules as target_types_rules


@dataclass(frozen=True)
class DebDependsFromPexRequest:
    target_pex: Pex
    distro: str
    distro_codename: str
    debian_arch: str


@dataclass(frozen=True)
class DebDependsInfo:
    requires: tuple[str, ...]


@rule
async def deb_depends_from_pex(request: DebDependsFromPexRequest) -> DebDependsInfo:
    # This rule partially replaces `dh_shlibdeps` + `dpkg-shlibdeps` in native deb builds.
    # `dpkg-shlibdeps` calculates deps that replace ${shlibs:<dep field>} vars in debian/control files.
    #   - By default, `dpkg-shlibdeps` puts the package deps in ${shlibs:Depends}.
    #   - When building an "Essential" package, it puts the deps in ${shlibs:Pre-Depends} instead.
    #   - If requested, some deps can also go in ${shlibs:Recommends} or ${shilbs:Suggests}.
    # This rule only calculates one list of deps (the equivalent of ${shlibs:Depends}).
    # Consuming rules are responsible for putting these deps in one or more nfpm package dep field(s).

    pex_elf_info = await elfdeps_analyze_pex_wheels(
        RequestPexELFInfo(request.target_pex), **implicitly()
    )

    sonames = shlibdeps_filter_sonames(so_info.soname for so_info in pex_elf_info.requires)

    packages_for_sonames = await deb_search_for_sonames(
        DebSearchForSonamesRequest(
            request.distro,
            request.distro_codename,
            request.debian_arch,
            sonames,
            from_best_so_files=True,
        )
    )

    # this blindly grabs all of them, but we really need to select only one so_file per soname and use those packages
    package_deps = {
        package
        for packages_for_soname in packages_for_sonames.packages_for_sonames
        for packages_per_so_file in packages_for_soname.packages_per_so_files
        for package in packages_per_so_file.packages
    }
    # TODO: Should there be a minimum version constraint for libc.so.6 based on so_info.version?
    return DebDependsInfo(requires=tuple(sorted(package_deps)))


@dataclass(frozen=True)
class RpmDependsFromPexRequest:
    target_pex: Pex


@dataclass(frozen=True)
class RpmDependsInfo:
    provides: tuple[str, ...]
    requires: tuple[str, ...]


@rule
async def rpm_depends_from_pex(request: RpmDependsFromPexRequest) -> RpmDependsInfo:
    # This rule provides a platform-agnostic replacement for `rpmdeps` in native rpm builds.
    pex_elf_info = await elfdeps_analyze_pex_wheels(
        RequestPexELFInfo(request.target_pex), **implicitly()
    )
    return RpmDependsInfo(
        provides=tuple(provided.so_info for provided in pex_elf_info.provides),
        requires=tuple(required.so_info for required in pex_elf_info.requires),
    )


async def _get_pex_deps_of_content_file_targets(
    address: Address, union_membership: UnionMembership
) -> tuple[Pex, ...]:
    """Get all pex_binary targets that are (transitive) deps of nfpm_content_file targets."""
    package_field_sets_for_nfpm_content_files = (
        await get_package_field_sets_for_nfpm_content_file_deps(
            GetPackageFieldSetsForNfpmContentFileDepsRequest([address], [PexBinaryFieldSet]),
            **implicitly(),
        )
    )
    pex_binary_field_sets = package_field_sets_for_nfpm_content_files.package_field_sets
    build_pex_requests = await concurrently(
        package_pex_binary(field_set, **implicitly())
        for field_set in pex_binary_field_sets.field_sets
    )
    pex_binaries = await concurrently(
        create_pex(**implicitly({pex_request.request: PexFromTargetsRequest}))
        for pex_request in build_pex_requests
    )
    return pex_binaries


class NativeLibsNfpmPackageFieldsRequest(InjectNfpmPackageFieldsRequest):
    # low priority to allow in-repo plugins to override/correct injected dependencies.
    priority = 2

    @classmethod
    def is_applicable(cls, target: Target) -> bool:
        return any(
            field_set_type.is_applicable(target)
            for field_set_type in (NfpmDebPackageFieldSet, NfpmRpmPackageFieldSet)
        )


@rule
async def inject_native_libs_dependencies_in_package_fields(
    request: NativeLibsNfpmPackageFieldsRequest,
    union_membership: UnionMembership,
) -> InjectedNfpmPackageFields:
    address = request.target.address

    fields: list[Field] = list(request.injected_fields.values())

    # TODO: support scanning more packaged binaries than just pex_binaries.

    pex_binaries = await _get_pex_deps_of_content_file_targets(address, union_membership)
    if not pex_binaries:
        return InjectedNfpmPackageFields(fields, address=address)

    if NfpmDebPackageFieldSet.is_applicable(request.target):
        # This is like running deb helper shlib-deps.
        deb_depends_infos = await concurrently(
            deb_depends_from_pex(
                DebDependsFromPexRequest(
                    pex,
                    request.get_field(DebDistroField).value or "",
                    request.get_field(DebDistroCodenameField).value or "",
                    request.get_field(NfpmArchField).value or "",
                )
            )
            for pex in pex_binaries
        )
        depends = list(request.get_field(NfpmDebDependsField).value or ())
        for deb_depends_info in deb_depends_infos:
            depends.extend(deb_depends_info.requires)
        fields.append(NfpmDebDependsField(depends, address=address))

    elif NfpmRpmPackageFieldSet.is_applicable(request.target):
        # This is like running rpmdeps -> elfdeps.
        rpm_depends_infos = await concurrently(
            rpm_depends_from_pex(RpmDependsFromPexRequest(pex)) for pex in pex_binaries
        )
        provides = list(request.get_field(NfpmRpmProvidesField).value or ())
        depends = list(request.get_field(NfpmRpmDependsField).value or ())
        for rpm_depends_info in rpm_depends_infos:
            provides.extend(rpm_depends_info.provides)
            depends.extend(rpm_depends_info.requires)
        fields.extend(
            [
                NfpmRpmProvidesField(provides, address=address),
                NfpmRpmDependsField(depends, address=address),
            ]
        )

    return InjectedNfpmPackageFields(fields, address=address)


def rules() -> Iterable[Rule | UnionRule]:
    return (
        *deb_rules(),
        *elfdeps_rules(),
        *target_types_rules(),
        *collect_rules(),
        UnionRule(InjectNfpmPackageFieldsRequest, NativeLibsNfpmPackageFieldsRequest),
    )

1	# Copyright 2025 Pants project contributors (see CONTRIBUTORS.md).
2	# Licensed under the Apache License, Version 2.0 (see LICENSE).
3
4	from __future__ import annotations	1✔
5
6	from collections.abc import Iterable	1✔
7	from dataclasses import dataclass	1✔
8
9	from pants.backend.nfpm.field_sets import NfpmDebPackageFieldSet, NfpmRpmPackageFieldSet	1✔
10	from pants.backend.nfpm.fields.all import NfpmArchField	1✔
11	from pants.backend.nfpm.fields.deb import NfpmDebDependsField	1✔
12	from pants.backend.nfpm.fields.rpm import NfpmRpmDependsField, NfpmRpmProvidesField	1✔
13	from pants.backend.nfpm.util_rules.contents import (	1✔
14	GetPackageFieldSetsForNfpmContentFileDepsRequest,
15	get_package_field_sets_for_nfpm_content_file_deps,
16	)
17	from pants.backend.nfpm.util_rules.inject_config import (	1✔
18	InjectedNfpmPackageFields,
19	InjectNfpmPackageFieldsRequest,
20	)
21	from pants.backend.python.goals.package_pex_binary import PexBinaryFieldSet, package_pex_binary	1✔
22	from pants.backend.python.util_rules.pex import Pex, create_pex	1✔
23	from pants.backend.python.util_rules.pex_from_targets import PexFromTargetsRequest	1✔
24	from pants.engine.addresses import Address	1✔
25	from pants.engine.internals.selectors import concurrently	1✔
26	from pants.engine.rules import Rule, collect_rules, implicitly, rule	1✔
27	from pants.engine.target import Field, Target	1✔
28	from pants.engine.unions import UnionMembership, UnionRule	1✔
29
30	from .deb.rules import DebSearchForSonamesRequest, deb_search_for_sonames	1✔
31	from .deb.rules import rules as deb_rules	1✔
32	from .deb.utils import shlibdeps_filter_sonames	1✔
33	from .elfdeps.rules import RequestPexELFInfo, elfdeps_analyze_pex_wheels	1✔
34	from .elfdeps.rules import rules as elfdeps_rules	1✔
35	from .target_types import DebDistroCodenameField, DebDistroField	1✔
36	from .target_types import rules as target_types_rules	1✔
37
38
39	@dataclass(frozen=True)	1✔
40	class DebDependsFromPexRequest:	1✔
41	target_pex: Pex	1✔
42	distro: str	1✔
43	distro_codename: str	1✔
44	debian_arch: str	1✔
45
46
47	@dataclass(frozen=True)	1✔
48	class DebDependsInfo:	1✔
49	requires: tuple[str, ...]	1✔
50
51
52	@rule	1✔
53	async def deb_depends_from_pex(request: DebDependsFromPexRequest) -> DebDependsInfo:	1✔
54	# This rule partially replaces `dh_shlibdeps` + `dpkg-shlibdeps` in native deb builds.
55	# `dpkg-shlibdeps` calculates deps that replace ${shlibs:<dep field>} vars in debian/control files.
56	# - By default, `dpkg-shlibdeps` puts the package deps in ${shlibs:Depends}.
57	# - When building an "Essential" package, it puts the deps in ${shlibs:Pre-Depends} instead.
58	# - If requested, some deps can also go in ${shlibs:Recommends} or ${shilbs:Suggests}.
59	# This rule only calculates one list of deps (the equivalent of ${shlibs:Depends}).
60	# Consuming rules are responsible for putting these deps in one or more nfpm package dep field(s).
61
NEW 62	pex_elf_info = await elfdeps_analyze_pex_wheels(	×
63	RequestPexELFInfo(request.target_pex), **implicitly()
64	)
65
NEW 66	sonames = shlibdeps_filter_sonames(so_info.soname for so_info in pex_elf_info.requires)	×
67
NEW 68	packages_for_sonames = await deb_search_for_sonames(	×
69	DebSearchForSonamesRequest(
70	request.distro,
71	request.distro_codename,
72	request.debian_arch,
73	sonames,
74	from_best_so_files=True,
75	)
76	)
77
78	# this blindly grabs all of them, but we really need to select only one so_file per soname and use those packages
NEW 79	package_deps = {	×
80	package
81	for packages_for_soname in packages_for_sonames.packages_for_sonames
82	for packages_per_so_file in packages_for_soname.packages_per_so_files
83	for package in packages_per_so_file.packages
84	}
85	# TODO: Should there be a minimum version constraint for libc.so.6 based on so_info.version?
NEW 86	return DebDependsInfo(requires=tuple(sorted(package_deps)))	×
87
88
89	@dataclass(frozen=True)	1✔
90	class RpmDependsFromPexRequest:	1✔
91	target_pex: Pex	1✔
92
93
94	@dataclass(frozen=True)	1✔
95	class RpmDependsInfo:	1✔
96	provides: tuple[str, ...]	1✔
97	requires: tuple[str, ...]	1✔
98
99
100	@rule	1✔
101	async def rpm_depends_from_pex(request: RpmDependsFromPexRequest) -> RpmDependsInfo:	1✔
102	# This rule provides a platform-agnostic replacement for `rpmdeps` in native rpm builds.
NEW 103	pex_elf_info = await elfdeps_analyze_pex_wheels(	×
104	RequestPexELFInfo(request.target_pex), **implicitly()
105	)
NEW 106	return RpmDependsInfo(	×
107	provides=tuple(provided.so_info for provided in pex_elf_info.provides),
108	requires=tuple(required.so_info for required in pex_elf_info.requires),
109	)
110
111
112	async def _get_pex_deps_of_content_file_targets(	1✔
113	address: Address, union_membership: UnionMembership
114	) -> tuple[Pex, ...]:
115	"""Get all pex_binary targets that are (transitive) deps of nfpm_content_file targets."""
NEW 116	package_field_sets_for_nfpm_content_files = (	×
117	await get_package_field_sets_for_nfpm_content_file_deps(
118	GetPackageFieldSetsForNfpmContentFileDepsRequest([address], [PexBinaryFieldSet]),
119	**implicitly(),
120	)
121	)
NEW 122	pex_binary_field_sets = package_field_sets_for_nfpm_content_files.package_field_sets	×
NEW 123	build_pex_requests = await concurrently(	×
124	package_pex_binary(field_set, **implicitly())
125	for field_set in pex_binary_field_sets.field_sets
126	)
NEW 127	pex_binaries = await concurrently(	×
128	create_pex(**implicitly({pex_request.request: PexFromTargetsRequest}))
129	for pex_request in build_pex_requests
130	)
NEW 131	return pex_binaries	×
132
133
134	class NativeLibsNfpmPackageFieldsRequest(InjectNfpmPackageFieldsRequest):	1✔
135	# low priority to allow in-repo plugins to override/correct injected dependencies.
136	priority = 2	1✔
137
138	@classmethod	1✔
139	def is_applicable(cls, target: Target) -> bool:	1✔
NEW 140	return any(	×
141	field_set_type.is_applicable(target)
142	for field_set_type in (NfpmDebPackageFieldSet, NfpmRpmPackageFieldSet)
143	)
144
145
146	@rule	1✔
147	async def inject_native_libs_dependencies_in_package_fields(	1✔
148	request: NativeLibsNfpmPackageFieldsRequest,
149	union_membership: UnionMembership,
150	) -> InjectedNfpmPackageFields:
NEW 151	address = request.target.address	×
152
NEW 153	fields: list[Field] = list(request.injected_fields.values())	×
154
155	# TODO: support scanning more packaged binaries than just pex_binaries.
156
NEW 157	pex_binaries = await _get_pex_deps_of_content_file_targets(address, union_membership)	×
NEW 158	if not pex_binaries:	×
NEW 159	return InjectedNfpmPackageFields(fields, address=address)	×
160
NEW 161	if NfpmDebPackageFieldSet.is_applicable(request.target):	×
162	# This is like running deb helper shlib-deps.
NEW 163	deb_depends_infos = await concurrently(	×
164	deb_depends_from_pex(
165	DebDependsFromPexRequest(
166	pex,
167	request.get_field(DebDistroField).value or "",
168	request.get_field(DebDistroCodenameField).value or "",
169	request.get_field(NfpmArchField).value or "",
170	)
171	)
172	for pex in pex_binaries
173	)
NEW 174	depends = list(request.get_field(NfpmDebDependsField).value or ())	×
NEW 175	for deb_depends_info in deb_depends_infos:	×
NEW 176	depends.extend(deb_depends_info.requires)	×
NEW 177	fields.append(NfpmDebDependsField(depends, address=address))	×
178
NEW 179	elif NfpmRpmPackageFieldSet.is_applicable(request.target):	×
180	# This is like running rpmdeps -> elfdeps.
NEW 181	rpm_depends_infos = await concurrently(	×
182	rpm_depends_from_pex(RpmDependsFromPexRequest(pex)) for pex in pex_binaries
183	)
NEW 184	provides = list(request.get_field(NfpmRpmProvidesField).value or ())	×
NEW 185	depends = list(request.get_field(NfpmRpmDependsField).value or ())	×
NEW 186	for rpm_depends_info in rpm_depends_infos:	×
NEW 187	provides.extend(rpm_depends_info.provides)	×
NEW 188	depends.extend(rpm_depends_info.requires)	×
NEW 189	fields.extend(	×
190	[
191	NfpmRpmProvidesField(provides, address=address),
192	NfpmRpmDependsField(depends, address=address),
193	]
194	)
195
NEW 196	return InjectedNfpmPackageFields(fields, address=address)	×
197
198
199	def rules() -> Iterable[Rule \| UnionRule]:	1✔
200	return (	1✔
201	*deb_rules(),
202	*elfdeps_rules(),
203	*target_types_rules(),
204	*collect_rules(),
205	UnionRule(InjectNfpmPackageFieldsRequest, NativeLibsNfpmPackageFieldsRequest),
206	)

pantsbuild / pants / 19123999670

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