20632486505

Committed 01 Jan 2026 04:21AM UTC coverage: 43.231% (-37.1%) from 80.281%

Build # 20632486505

Build Type

Pull #22962

github

Committed by

web-flow

Commit Message

Merge 08d5c63b0 into f52ab6675

Pull Request Pull Request #22962: Bump the gha-deps group across 1 directory with 6 updates

Run Details

26122 of 60424 relevant lines covered (43.23%)

0.86 hits per line

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

0.0

/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_tool
from pants.backend.python.util_rules.pex import Pex, PexProcess, VenvPexProcess
from pants.backend.python.util_rules.pex_cli import PexPEX
from pants.core.util_rules.system_binaries import UnzipBinary
from pants.engine.process import Process, 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 and files in a PEX",
    level=LogLevel.DEBUG,
)
async def elfdeps_analyze_pex(
    request: RequestPexELFInfo, pex_pex: PexPEX, unzip: UnzipBinary
) -> PexELFInfo:
    wheel_repo_dir = str(PurePath(request.target_pex.name).with_suffix(".wheel_repo"))
    contents_dir = str(PurePath(request.target_pex.name).with_suffix(".wheel_repo"))

    extracted_wheels, extracted_files, elfdeps_analyze_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,
                )
            )
        ),
        execute_process_or_raise(
            **implicitly(
                # pex-tools can only extract wheels and PEX-INFO,
                # so unzip remaining files from pex manually
                # (even though that means relying on pex implementation details)
                Process(
                    argv=[
                        unzip.path,
                        request.target_pex.name,
                        "-x",  # exclude wheels
                        ".deps/*",  # this ties us to this pex implementation detail :(
                        "-d",
                        wheel_repo_dir,
                    ],
                    input_digest=request.target_pex.digest,
                    output_directories=(contents_dir,),
                    description=f"Extract non-wheel files from {request.target_pex.name}",
                    level=LogLevel.DEBUG,
                ),
            )
        ),
        setup_elfdeps_analyze_tool(**implicitly()),
    )

    process_results: tuple[ProcessResult, ProcessResult] = await concurrently(
        execute_process_or_raise(
            **implicitly(
                VenvPexProcess(
                    elfdeps_analyze_tool.pex,
                    argv=("--mode", "wheels", 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,
                )
            )
        ),
        execute_process_or_raise(
            **implicitly(
                VenvPexProcess(
                    elfdeps_analyze_tool.pex,
                    argv=("--mode", "files", contents_dir),
                    input_digest=extracted_files.output_digest,
                    description=f"Calculate ELF provides+requires for non-wheel files in pex {request.target_pex.name}",
                    level=LogLevel.DEBUG,
                )
            )
        ),
    )

    pex_elf_info: dict[str, list[dict[str, str]]] = {"provides": [], "requires": []}
    for process_result in process_results:
        result = json.loads(process_result.stdout)
        pex_elf_info["provides"].extend(result["provides"])
        pex_elf_info["requires"].extend(result["requires"])

    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	×
5
6	import json	×
7	from collections.abc import Iterable, Mapping	×
8	from dataclasses import dataclass	×
9	from dataclasses import field as dataclass_field	×
10	from pathlib import PurePath	×
11
12	from pants.backend.nfpm.native_libs.elfdeps.subsystem import rules as subsystem_rules	×
13	from pants.backend.nfpm.native_libs.elfdeps.subsystem import setup_elfdeps_analyze_tool	×
14	from pants.backend.python.util_rules.pex import Pex, PexProcess, VenvPexProcess	×
15	from pants.backend.python.util_rules.pex_cli import PexPEX	×
16	from pants.core.util_rules.system_binaries import UnzipBinary	×
17	from pants.engine.process import Process, ProcessResult, execute_process_or_raise	×
18	from pants.engine.rules import Rule, collect_rules, concurrently, implicitly, rule	×
19	from pants.util.logging import LogLevel	×
20
21
22	@dataclass(frozen=True)	×
23	class RequestPexELFInfo:	×
24	target_pex: Pex	×
25
26
27	@dataclass(frozen=True, order=True)	×
28	class SOInfo: # elfdeps should not be used in rules, so this holds the same data as elfdeps.SOInfo.	×
29	soname: str	×
30	version: str	×
31	marker: str	×
32	# so_info combines soname+version+marker using whatever standard elfdeps follows.
33	so_info: str = dataclass_field(compare=False)	×
34
35
36	@dataclass(frozen=True)	×
37	class PexELFInfo:	×
38	provides: tuple[SOInfo, ...]	×
39	requires: tuple[SOInfo, ...]	×
40
41	def __init__(	×
42	self, provides: Iterable[Mapping[str, str]], requires: Iterable[Mapping[str, str]]
43	):
44	object.__setattr__(	×
45	self, "provides", tuple(sorted({SOInfo(**so_info) for so_info in provides}))
46	)
47	object.__setattr__(	×
48	self, "requires", tuple(sorted({SOInfo(**so_info) for so_info in requires}))
49	)
50
51
52	@rule(	×
53	desc="Analyze ELF (native lib) dependencies of wheels and files in a PEX",
54	level=LogLevel.DEBUG,
55	)
56	async def elfdeps_analyze_pex(	×
57	request: RequestPexELFInfo, pex_pex: PexPEX, unzip: UnzipBinary
58	) -> PexELFInfo:
59	wheel_repo_dir = str(PurePath(request.target_pex.name).with_suffix(".wheel_repo"))	×
60	contents_dir = str(PurePath(request.target_pex.name).with_suffix(".wheel_repo"))	×
61
62	extracted_wheels, extracted_files, elfdeps_analyze_tool = await concurrently(	×
63	execute_process_or_raise(
64	**implicitly(
65	PexProcess(
66	pex=Pex(
67	digest=pex_pex.digest,
68	name=pex_pex.exe,
69	python=request.target_pex.python,
70	),
71	argv=[
72	request.target_pex.name,
73	"repository",
74	"extract",
75	"--dest-dir",
76	wheel_repo_dir,
77	],
78	input_digest=request.target_pex.digest,
79	output_directories=(wheel_repo_dir,),
80	extra_env={"PEX_MODULE": "pex.tools"},
81	description=f"Extract wheels from {request.target_pex.name}",
82	level=LogLevel.DEBUG,
83	)
84	)
85	),
86	execute_process_or_raise(
87	**implicitly(
88	# pex-tools can only extract wheels and PEX-INFO,
89	# so unzip remaining files from pex manually
90	# (even though that means relying on pex implementation details)
91	Process(
92	argv=[
93	unzip.path,
94	request.target_pex.name,
95	"-x", # exclude wheels
96	".deps/*", # this ties us to this pex implementation detail :(
97	"-d",
98	wheel_repo_dir,
99	],
100	input_digest=request.target_pex.digest,
101	output_directories=(contents_dir,),
102	description=f"Extract non-wheel files from {request.target_pex.name}",
103	level=LogLevel.DEBUG,
104	),
105	)
106	),
107	setup_elfdeps_analyze_tool(**implicitly()),
108	)
109
110	process_results: tuple[ProcessResult, ProcessResult] = await concurrently(	×
111	execute_process_or_raise(
112	**implicitly(
113	VenvPexProcess(
114	elfdeps_analyze_tool.pex,
115	argv=("--mode", "wheels", wheel_repo_dir),
116	input_digest=extracted_wheels.output_digest,
117	description=f"Calculate ELF provides+requires for wheels in pex {request.target_pex.name}",
118	level=LogLevel.DEBUG,
119	)
120	)
121	),
122	execute_process_or_raise(
123	**implicitly(
124	VenvPexProcess(
125	elfdeps_analyze_tool.pex,
126	argv=("--mode", "files", contents_dir),
127	input_digest=extracted_files.output_digest,
128	description=f"Calculate ELF provides+requires for non-wheel files in pex {request.target_pex.name}",
129	level=LogLevel.DEBUG,
130	)
131	)
132	),
133	)
134
135	pex_elf_info: dict[str, list[dict[str, str]]] = {"provides": [], "requires": []}	×
136	for process_result in process_results:	×
137	result = json.loads(process_result.stdout)	×
138	pex_elf_info["provides"].extend(result["provides"])	×
139	pex_elf_info["requires"].extend(result["requires"])	×
140
141	return PexELFInfo(	×
142	provides=pex_elf_info["provides"],
143	requires=pex_elf_info["requires"],
144	)
145
146
147	def rules() -> Iterable[Rule]:	×
148	return (	×
149	*subsystem_rules(),
150	*collect_rules(),
151	)

pantsbuild / pants / 20632486505

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