19120576134

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

Build # 19120576134

Build Type

Pull #22861

github

Committed by

web-flow

Commit Message

Merge 9479b1cb5 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

81.58

/src/python/pants/backend/nfpm/native_libs/elfdeps/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 json
from collections.abc import Iterable, Mapping
from dataclasses import dataclass
from dataclasses import field as dataclass_field
from pathlib import PurePath

from pants.backend.nfpm.native_libs.elfdeps.subsystem import rules as subsystem_rules
from pants.backend.nfpm.native_libs.elfdeps.subsystem import setup_elfdeps_analyze_wheels_tool
from pants.backend.python.util_rules.pex import Pex, PexProcess, VenvPexProcess
from pants.backend.python.util_rules.pex_cli import PexPEX
from pants.engine.process import ProcessResult, execute_process_or_raise
from pants.engine.rules import Rule, collect_rules, concurrently, implicitly, rule
from pants.util.logging import LogLevel


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


@dataclass(frozen=True, order=True)
class SOInfo:  # elfdeps should not be used in rules, so this holds the same data as elfdeps.SOInfo.
    soname: str
    version: str
    marker: str
    # so_info combines soname+version+marker using whatever standard elfdeps follows.
    so_info: str = dataclass_field(compare=False)


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

    def __init__(
        self, provides: Iterable[Mapping[str, str]], requires: Iterable[Mapping[str, str]]
    ):
        object.__setattr__(
            self, "provides", tuple(sorted(SOInfo(**so_info) for so_info in provides))
        )
        object.__setattr__(
            self, "requires", tuple(sorted(SOInfo(**so_info) for so_info in requires))
        )


@rule(
    desc="Analyze ELF (native lib) dependencies of wheels in a PEX",
    level=LogLevel.DEBUG,
)
async def elfdeps_analyze_pex_wheels(request: RequestPexELFInfo, pex_pex: PexPEX) -> PexELFInfo:
    wheel_repo_dir = str(PurePath(request.target_pex.name).with_suffix(".wheel_repo"))

    extracted_wheels, elfdeps_analyze_wheels_tool = await concurrently(
        execute_process_or_raise(
            **implicitly(
                PexProcess(
                    pex=Pex(
                        digest=pex_pex.digest,
                        name=pex_pex.exe,
                        python=request.target_pex.python,
                    ),
                    argv=[
                        request.target_pex.name,
                        "repository",
                        "extract",
                        "--dest-dir",
                        wheel_repo_dir,
                    ],
                    input_digest=request.target_pex.digest,
                    output_directories=(wheel_repo_dir,),
                    extra_env={"PEX_MODULE": "pex.tools"},
                    description=f"Extract wheels from {request.target_pex.name}",
                    level=LogLevel.DEBUG,
                )
            )
        ),
        setup_elfdeps_analyze_wheels_tool(**implicitly()),
    )

    result: ProcessResult = await execute_process_or_raise(
        **implicitly(
            VenvPexProcess(
                elfdeps_analyze_wheels_tool.pex,
                argv=(wheel_repo_dir,),
                input_digest=extracted_wheels.output_digest,
                description=f"Calculate ELF provides+requires for wheels in pex {request.target_pex.name}",
                level=LogLevel.DEBUG,
            )
        )
    )

    pex_elf_info = json.loads(result.stdout)
    return PexELFInfo(
        provides=pex_elf_info["provides"],
        requires=pex_elf_info["requires"],
    )


def rules() -> Iterable[Rule]:
    return (
        *subsystem_rules(),
        *collect_rules(),
    )

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	2✔
5
6	import json	2✔
7	from collections.abc import Iterable, Mapping	2✔
8	from dataclasses import dataclass	2✔
9	from dataclasses import field as dataclass_field	2✔
10	from pathlib import PurePath	2✔
11
12	from pants.backend.nfpm.native_libs.elfdeps.subsystem import rules as subsystem_rules	2✔
13	from pants.backend.nfpm.native_libs.elfdeps.subsystem import setup_elfdeps_analyze_wheels_tool	2✔
14	from pants.backend.python.util_rules.pex import Pex, PexProcess, VenvPexProcess	2✔
15	from pants.backend.python.util_rules.pex_cli import PexPEX	2✔
16	from pants.engine.process import ProcessResult, execute_process_or_raise	2✔
17	from pants.engine.rules import Rule, collect_rules, concurrently, implicitly, rule	2✔
18	from pants.util.logging import LogLevel	2✔
19
20
21	@dataclass(frozen=True)	2✔
22	class RequestPexELFInfo:	2✔
23	target_pex: Pex	2✔
24
25
26	@dataclass(frozen=True, order=True)	2✔
27	class SOInfo: # elfdeps should not be used in rules, so this holds the same data as elfdeps.SOInfo.	2✔
28	soname: str	2✔
29	version: str	2✔
30	marker: str	2✔
31	# so_info combines soname+version+marker using whatever standard elfdeps follows.
32	so_info: str = dataclass_field(compare=False)	2✔
33
34
35	@dataclass(frozen=True)	2✔
36	class PexELFInfo:	2✔
37	provides: tuple[SOInfo, ...]	2✔
38	requires: tuple[SOInfo, ...]	2✔
39
40	def __init__(	2✔
41	self, provides: Iterable[Mapping[str, str]], requires: Iterable[Mapping[str, str]]
42	):
NEW 43	object.__setattr__(	×
44	self, "provides", tuple(sorted(SOInfo(**so_info) for so_info in provides))
45	)
NEW 46	object.__setattr__(	×
47	self, "requires", tuple(sorted(SOInfo(**so_info) for so_info in requires))
48	)
49
50
51	@rule(	2✔
52	desc="Analyze ELF (native lib) dependencies of wheels in a PEX",
53	level=LogLevel.DEBUG,
54	)
55	async def elfdeps_analyze_pex_wheels(request: RequestPexELFInfo, pex_pex: PexPEX) -> PexELFInfo:	2✔
NEW 56	wheel_repo_dir = str(PurePath(request.target_pex.name).with_suffix(".wheel_repo"))	×
57
NEW 58	extracted_wheels, elfdeps_analyze_wheels_tool = await concurrently(	×
59	execute_process_or_raise(
60	**implicitly(
61	PexProcess(
62	pex=Pex(
63	digest=pex_pex.digest,
64	name=pex_pex.exe,
65	python=request.target_pex.python,
66	),
67	argv=[
68	request.target_pex.name,
69	"repository",
70	"extract",
71	"--dest-dir",
72	wheel_repo_dir,
73	],
74	input_digest=request.target_pex.digest,
75	output_directories=(wheel_repo_dir,),
76	extra_env={"PEX_MODULE": "pex.tools"},
77	description=f"Extract wheels from {request.target_pex.name}",
78	level=LogLevel.DEBUG,
79	)
80	)
81	),
82	setup_elfdeps_analyze_wheels_tool(**implicitly()),
83	)
84
NEW 85	result: ProcessResult = await execute_process_or_raise(	×
86	**implicitly(
87	VenvPexProcess(
88	elfdeps_analyze_wheels_tool.pex,
89	argv=(wheel_repo_dir,),
90	input_digest=extracted_wheels.output_digest,
91	description=f"Calculate ELF provides+requires for wheels in pex {request.target_pex.name}",
92	level=LogLevel.DEBUG,
93	)
94	)
95	)
96
NEW 97	pex_elf_info = json.loads(result.stdout)	×
NEW 98	return PexELFInfo(	×
99	provides=pex_elf_info["provides"],
100	requires=pex_elf_info["requires"],
101	)
102
103
104	def rules() -> Iterable[Rule]:	2✔
105	return (	2✔
106	*subsystem_rules(),
107	*collect_rules(),
108	)

pantsbuild / pants / 19120576134

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