19562031062

Committed 21 Nov 2025 06:24AM UTC coverage: 80.287% (-0.008%) from 80.295%

Build # 19562031062

Build Type

Pull #22904

github

Committed by

web-flow

Commit Message

Merge 299d34458 into 70dc9fe34

Pull Request Pull Request #22904: nfpm.native_libs: deb search_for_sonames script to find pkg deps

Run Details

160 of 209 new or added lines in 5 files covered. (76.56%)

15 existing lines in 1 file now uncovered.

78545 of 97830 relevant lines covered (80.29%)

3.36 hits per line

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

63.33

/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

import re
from collections.abc import Iterable
from dataclasses import dataclass

from pants.backend.nfpm.field_sets import NfpmRpmPackageFieldSet
from pants.backend.nfpm.fields.rpm import NfpmRpmDependsField, NfpmRpmProvidesField
from pants.backend.nfpm.native_libs.deb.rules import (
    DebSearchForSonamesRequest,
    deb_search_for_sonames,
)
from pants.backend.nfpm.native_libs.deb.rules import rules as deb_rules
from pants.backend.nfpm.native_libs.elfdeps.rules import RequestPexELFInfo, elfdeps_analyze_pex
from pants.backend.nfpm.native_libs.elfdeps.rules import rules as elfdeps_rules
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


@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(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 NfpmRpmPackageFieldSet.is_applicable(target)


@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 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(),
        *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	import re	1✔
7	from collections.abc import Iterable	1✔
8	from dataclasses import dataclass	1✔
9
10	from pants.backend.nfpm.field_sets import NfpmRpmPackageFieldSet	1✔
11	from pants.backend.nfpm.fields.rpm import NfpmRpmDependsField, NfpmRpmProvidesField	1✔
12	from pants.backend.nfpm.native_libs.deb.rules import (	1✔
13	DebSearchForSonamesRequest,
14	deb_search_for_sonames,
15	)
16	from pants.backend.nfpm.native_libs.deb.rules import rules as deb_rules	1✔
17	from pants.backend.nfpm.native_libs.elfdeps.rules import RequestPexELFInfo, elfdeps_analyze_pex	1✔
18	from pants.backend.nfpm.native_libs.elfdeps.rules import rules as elfdeps_rules	1✔
19	from pants.backend.nfpm.util_rules.contents import (	1✔
20	GetPackageFieldSetsForNfpmContentFileDepsRequest,
21	get_package_field_sets_for_nfpm_content_file_deps,
22	)
23	from pants.backend.nfpm.util_rules.inject_config import (	1✔
24	InjectedNfpmPackageFields,
25	InjectNfpmPackageFieldsRequest,
26	)
27	from pants.backend.python.goals.package_pex_binary import PexBinaryFieldSet, package_pex_binary	1✔
28	from pants.backend.python.util_rules.pex import Pex, create_pex	1✔
29	from pants.backend.python.util_rules.pex_from_targets import PexFromTargetsRequest	1✔
30	from pants.engine.addresses import Address	1✔
31	from pants.engine.internals.selectors import concurrently	1✔
32	from pants.engine.rules import Rule, collect_rules, implicitly, rule	1✔
33	from pants.engine.target import Field, Target	1✔
34	from pants.engine.unions import UnionMembership, UnionRule	1✔
35
36
37	@dataclass(frozen=True)	1✔
38	class RpmDependsFromPexRequest:	1✔
39	target_pex: Pex	1✔
40
41
42	@dataclass(frozen=True)	1✔
43	class RpmDependsInfo:	1✔
44	provides: tuple[str, ...]	1✔
45	requires: tuple[str, ...]	1✔
46
47
48	@rule	1✔
49	async def rpm_depends_from_pex(request: RpmDependsFromPexRequest) -> RpmDependsInfo:	1✔
50	# This rule provides a platform-agnostic replacement for `rpmdeps` in native rpm builds.
UNCOV 51	pex_elf_info = await elfdeps_analyze_pex(RequestPexELFInfo(request.target_pex), **implicitly())	×
UNCOV 52	return RpmDependsInfo(	×
53	provides=tuple(provided.so_info for provided in pex_elf_info.provides),
54	requires=tuple(required.so_info for required in pex_elf_info.requires),
55	)
56
57
58	async def _get_pex_deps_of_content_file_targets(	1✔
59	address: Address, union_membership: UnionMembership
60	) -> tuple[Pex, ...]:
61	"""Get all pex_binary targets that are (transitive) deps of nfpm_content_file targets."""
UNCOV 62	package_field_sets_for_nfpm_content_files = (	×
63	await get_package_field_sets_for_nfpm_content_file_deps(
64	GetPackageFieldSetsForNfpmContentFileDepsRequest([address], [PexBinaryFieldSet]),
65	**implicitly(),
66	)
67	)
68	pex_binary_field_sets = package_field_sets_for_nfpm_content_files.package_field_sets	×
UNCOV 69	build_pex_requests = await concurrently(	×
70	package_pex_binary(field_set, **implicitly())
71	for field_set in pex_binary_field_sets.field_sets
72	)
UNCOV 73	pex_binaries = await concurrently(	×
74	create_pex(**implicitly({pex_request.request: PexFromTargetsRequest}))
75	for pex_request in build_pex_requests
76	)
UNCOV 77	return pex_binaries	×
78
79
80	class NativeLibsNfpmPackageFieldsRequest(InjectNfpmPackageFieldsRequest):	1✔
81	# low priority to allow in-repo plugins to override/correct injected dependencies.
82	priority = 2	1✔
83
84	@classmethod	1✔
85	def is_applicable(cls, target: Target) -> bool:	1✔
UNCOV 86	return NfpmRpmPackageFieldSet.is_applicable(target)	×
87
88
89	@rule	1✔
90	async def inject_native_libs_dependencies_in_package_fields(	1✔
91	request: NativeLibsNfpmPackageFieldsRequest,
92	union_membership: UnionMembership,
93	) -> InjectedNfpmPackageFields:
UNCOV 94	address = request.target.address	×
95
96	fields: list[Field] = list(request.injected_fields.values())	×
97
98	# TODO: support scanning more packaged binaries than just pex_binaries.
99
UNCOV 100	pex_binaries = await _get_pex_deps_of_content_file_targets(address, union_membership)	×
101	if not pex_binaries:	×
UNCOV 102	return InjectedNfpmPackageFields(fields, address=address)	×
103
104	if NfpmRpmPackageFieldSet.is_applicable(request.target):	×
105	# This is like running rpmdeps -> elfdeps.
106	rpm_depends_infos = await concurrently(	×
107	rpm_depends_from_pex(RpmDependsFromPexRequest(pex)) for pex in pex_binaries
108	)
109	provides = list(request.get_field(NfpmRpmProvidesField).value or ())	×
UNCOV 110	depends = list(request.get_field(NfpmRpmDependsField).value or ())	×
UNCOV 111	for rpm_depends_info in rpm_depends_infos:	×
UNCOV 112	provides.extend(rpm_depends_info.provides)	×
UNCOV 113	depends.extend(rpm_depends_info.requires)	×
UNCOV 114	fields.extend(	×
115	[
116	NfpmRpmProvidesField(provides, address=address),
117	NfpmRpmDependsField(depends, address=address),
118	]
119	)
120
121	return InjectedNfpmPackageFields(fields, address=address)	×
122
123
124	def rules() -> Iterable[Rule \| UnionRule]:	1✔
125	return (	1✔
126	*deb_rules(),
127	*elfdeps_rules(),
128	*collect_rules(),
129	UnionRule(InjectNfpmPackageFieldsRequest, NativeLibsNfpmPackageFieldsRequest),
130	)

pantsbuild / pants / 19562031062

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