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

67.68

/src/python/pants/backend/nfpm/native_libs/deb/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 importlib.metadata
import json
import logging
import sys
from collections.abc import Iterable, Mapping
from dataclasses import dataclass, replace
from pathlib import PurePath

from pants.backend.python.util_rules.pex import PexRequest, VenvPexProcess, create_venv_pex
from pants.backend.python.util_rules.pex_environment import PythonExecutable
from pants.backend.python.util_rules.pex_requirements import PexRequirements
from pants.engine.fs import CreateDigest, FileContent
from pants.engine.internals.native_engine import UnionRule
from pants.engine.internals.selectors import concurrently
from pants.engine.intrinsics import create_digest, execute_process
from pants.engine.process import FallibleProcessResult
from pants.engine.rules import Rule, collect_rules, implicitly, rule
from pants.init.import_util import find_matching_distributions
from pants.util.logging import LogLevel
from pants.util.resources import read_resource
from pants.version import VERSION

logger = logging.getLogger(__name__)

_NATIVE_LIBS_DEB_PACKAGE = "pants.backend.nfpm.native_libs.deb"
_SEARCH_FOR_SONAMES_SCRIPT = "search_for_sonames.py"
_PEX_NAME = "native_libs_deb.pex"


@dataclass(frozen=True)
class DebSearchForSonamesRequest:
    distro: str
    distro_codename: str
    debian_arch: str
    sonames: tuple[str, ...]
    from_best_so_files: bool

    def __init__(
        self,
        distro: str,
        distro_codename: str,
        debian_arch: str,
        sonames: Iterable[str],
        *,
        from_best_so_files: bool = False,
    ):
        object.__setattr__(self, "distro", distro)
        object.__setattr__(self, "distro_codename", distro_codename)
        object.__setattr__(self, "debian_arch", debian_arch)
        object.__setattr__(self, "sonames", tuple(sorted(sonames)))
        object.__setattr__(self, "from_best_so_files", from_best_so_files)


@dataclass(frozen=True)
class DebPackagesPerSoFile:
    so_file: str
    packages: tuple[str, ...]

    def __init__(self, so_file: str, packages: Iterable[str]):
        object.__setattr__(self, "so_file", so_file)
        object.__setattr__(self, "packages", tuple(sorted(packages)))


_TYPICAL_LD_PATH_PATTERNS = (
    # platform specific system libs (like libc) get selected first
    # "/usr/local/lib/*-linux-*/",
    "/lib/*-linux-*/",
    "/usr/lib/*-linux-*/",
    # Then look for a generic system libs
    # "/usr/local/lib/",
    "/lib/",
    "/usr/lib/",
    # Anything else has to be added manually to dependencies.
    # These rules cannot use symbols or shlibs metadata to inform package selection.
)


@dataclass(frozen=True)
class DebPackagesForSoname:
    soname: str
    packages_per_so_files: tuple[DebPackagesPerSoFile, ...]

    def __init__(self, soname: str, packages_per_so_files: Iterable[DebPackagesPerSoFile]):
        object.__setattr__(self, "soname", soname)
        object.__setattr__(self, "packages_per_so_files", tuple(packages_per_so_files))

    @property
    def from_best_so_files(self) -> DebPackagesForSoname:
        """Pick best so_files from packages_for_so_files using a simplified ld.so-like algorithm.

        The most preferred is first. This is NOT a recursive match; Only match if direct child of
        ld_path_patt dir. Anything that uses a subdir like /usr/lib/<app>/lib*.so.* uses rpath to
        prefer the app's libs over system libs. If this vastly simplified form of ld.so-style
        matching does not select the correct libs, then the package(s) that provide the shared lib
        should be added manually to the nfpm requires field.
        """
        if len(self.packages_per_so_files) <= 1:  # shortcut; no filtering required for 0-1 results.
            return self

        remaining = list(self.packages_per_so_files)

        packages_per_so_files = []
        for ld_path_patt in _TYPICAL_LD_PATH_PATTERNS:
            for packages_per_so_file in remaining[:]:
                if PurePath(packages_per_so_file.so_file).parent.match(ld_path_patt):
                    packages_per_so_files.append(packages_per_so_file)
                    remaining.remove(packages_per_so_file)

        return replace(self, packages_per_so_files=tuple(packages_per_so_files))


@dataclass(frozen=True)
class DebPackagesForSonames:
    packages_for_sonames: tuple[DebPackagesForSoname, ...]

    @classmethod
    def from_dict(cls, raw: Mapping[str, Mapping[str, Iterable[str]]]) -> DebPackagesForSonames:
        return cls(
            tuple(
                DebPackagesForSoname(
                    soname,
                    (
                        DebPackagesPerSoFile(so_file, packages)
                        for so_file, packages in files_to_packages.items()
                    ),
                )
                for soname, files_to_packages in raw.items()
            )
        )

    @property
    def from_best_so_files(self) -> DebPackagesForSonames:
        packages = []
        for packages_for_soname in self.packages_for_sonames:
            packages.append(packages_for_soname.from_best_so_files)
        return DebPackagesForSonames(tuple(packages))


@rule
async def deb_search_for_sonames(
    request: DebSearchForSonamesRequest,
) -> DebPackagesForSonames:
    script = read_resource(_NATIVE_LIBS_DEB_PACKAGE, _SEARCH_FOR_SONAMES_SCRIPT)
    if not script:
        raise ValueError(
            f"Unable to find source of {_SEARCH_FOR_SONAMES_SCRIPT!r} in {_NATIVE_LIBS_DEB_PACKAGE}"
        )

    script_content = FileContent(
        path=_SEARCH_FOR_SONAMES_SCRIPT, content=script, is_executable=True
    )

    # Pull python and requirements versions from the pants venv since that is what the script is tested with.
    pants_python = PythonExecutable.fingerprinted(
        sys.executable, ".".join(map(str, sys.version_info[:3])).encode("utf8")
    )
    distributions_in_pants_venv: list[importlib.metadata.Distribution] = list(
        find_matching_distributions(None)
    )
    constraints = tuple(f"{dist.name}=={dist.version}" for dist in distributions_in_pants_venv)
    requirements = {  # requirements (and transitive deps) are constrained to the versions in the pants venv
        "aiohttp",
        "aiohttp-retry",
        "beautifulsoup4",
    }

    script_digest, venv_pex = await concurrently(
        create_digest(CreateDigest([script_content])),
        create_venv_pex(
            **implicitly(
                PexRequest(
                    output_filename=_PEX_NAME,
                    internal_only=True,
                    python=pants_python,
                    requirements=PexRequirements(
                        requirements,
                        constraints_strings=constraints,
                        description_of_origin=f"Requirements for {_PEX_NAME}:{_SEARCH_FOR_SONAMES_SCRIPT}",
                    ),
                )
            )
        ),
    )

    result: FallibleProcessResult = await execute_process(
        **implicitly(
            VenvPexProcess(
                venv_pex,
                argv=(
                    script_content.path,
                    f"--user-agent-suffix=pants/{VERSION}",
                    f"--distro={request.distro}",
                    f"--distro-codename={request.distro_codename}",
                    f"--arch={request.debian_arch}",
                    *request.sonames,
                ),
                input_digest=script_digest,
                description=f"Search deb packages for sonames: {request.sonames}",
                level=LogLevel.DEBUG,
            )
        )
    )

    if result.exit_code == 0:
        packages = json.loads(result.stdout)
    else:
        # The search API returns 200 even if no results were found.
        # A 4xx or 5xx error means we gave up retrying because the server is unavailable.
        # TODO: Should this raise an error instead of just a warning?
        logger.warning(result.stderr.decode("utf-8"))
        packages = {}

    deb_packages_for_sonames = DebPackagesForSonames.from_dict(packages)
    if request.from_best_so_files:
        return deb_packages_for_sonames.from_best_so_files
    return deb_packages_for_sonames


def rules() -> Iterable[Rule | UnionRule]:
    return 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 importlib.metadata	2✔
7	import json	2✔
8	import logging	2✔
9	import sys	2✔
10	from collections.abc import Iterable, Mapping	2✔
11	from dataclasses import dataclass, replace	2✔
12	from pathlib import PurePath	2✔
13
14	from pants.backend.python.util_rules.pex import PexRequest, VenvPexProcess, create_venv_pex	2✔
15	from pants.backend.python.util_rules.pex_environment import PythonExecutable	2✔
16	from pants.backend.python.util_rules.pex_requirements import PexRequirements	2✔
17	from pants.engine.fs import CreateDigest, FileContent	2✔
18	from pants.engine.internals.native_engine import UnionRule	2✔
19	from pants.engine.internals.selectors import concurrently	2✔
20	from pants.engine.intrinsics import create_digest, execute_process	2✔
21	from pants.engine.process import FallibleProcessResult	2✔
22	from pants.engine.rules import Rule, collect_rules, implicitly, rule	2✔
23	from pants.init.import_util import find_matching_distributions	2✔
24	from pants.util.logging import LogLevel	2✔
25	from pants.util.resources import read_resource	2✔
26	from pants.version import VERSION	2✔
27
28	logger = logging.getLogger(__name__)	2✔
29
30	_NATIVE_LIBS_DEB_PACKAGE = "pants.backend.nfpm.native_libs.deb"	2✔
31	_SEARCH_FOR_SONAMES_SCRIPT = "search_for_sonames.py"	2✔
32	_PEX_NAME = "native_libs_deb.pex"	2✔
33
34
35	@dataclass(frozen=True)	2✔
36	class DebSearchForSonamesRequest:	2✔
37	distro: str	2✔
38	distro_codename: str	2✔
39	debian_arch: str	2✔
40	sonames: tuple[str, ...]	2✔
41	from_best_so_files: bool	2✔
42
43	def __init__(	2✔
44	self,
45	distro: str,
46	distro_codename: str,
47	debian_arch: str,
48	sonames: Iterable[str],
49	*,
50	from_best_so_files: bool = False,
51	):
52	object.__setattr__(self, "distro", distro)	1✔
53	object.__setattr__(self, "distro_codename", distro_codename)	1✔
54	object.__setattr__(self, "debian_arch", debian_arch)	1✔
55	object.__setattr__(self, "sonames", tuple(sorted(sonames)))	1✔
56	object.__setattr__(self, "from_best_so_files", from_best_so_files)	1✔
57
58
59	@dataclass(frozen=True)	2✔
60	class DebPackagesPerSoFile:	2✔
61	so_file: str	2✔
62	packages: tuple[str, ...]	2✔
63
64	def __init__(self, so_file: str, packages: Iterable[str]):	2✔
65	object.__setattr__(self, "so_file", so_file)	1✔
66	object.__setattr__(self, "packages", tuple(sorted(packages)))	1✔
67
68
69	_TYPICAL_LD_PATH_PATTERNS = (	2✔
70	# platform specific system libs (like libc) get selected first
71	# "/usr/local/lib/-linux-/",
72	"/lib/-linux-/",
73	"/usr/lib/-linux-/",
74	# Then look for a generic system libs
75	# "/usr/local/lib/",
76	"/lib/",
77	"/usr/lib/",
78	# Anything else has to be added manually to dependencies.
79	# These rules cannot use symbols or shlibs metadata to inform package selection.
80	)
81
82
83	@dataclass(frozen=True)	2✔
84	class DebPackagesForSoname:	2✔
85	soname: str	2✔
86	packages_per_so_files: tuple[DebPackagesPerSoFile, ...]	2✔
87
88	def __init__(self, soname: str, packages_per_so_files: Iterable[DebPackagesPerSoFile]):	2✔
89	object.__setattr__(self, "soname", soname)	1✔
90	object.__setattr__(self, "packages_per_so_files", tuple(packages_per_so_files))	1✔
91
92	@property	2✔
93	def from_best_so_files(self) -> DebPackagesForSoname:	2✔
94	"""Pick best so_files from packages_for_so_files using a simplified ld.so-like algorithm.
95
96	The most preferred is first. This is NOT a recursive match; Only match if direct child of
97	ld_path_patt dir. Anything that uses a subdir like /usr/lib/<app>/lib.so. uses rpath to
98	prefer the app's libs over system libs. If this vastly simplified form of ld.so-style
99	matching does not select the correct libs, then the package(s) that provide the shared lib
100	should be added manually to the nfpm requires field.
101	"""
NEW 102	if len(self.packages_per_so_files) <= 1: # shortcut; no filtering required for 0-1 results.	×
NEW 103	return self	×
104
NEW 105	remaining = list(self.packages_per_so_files)	×
106
NEW 107	packages_per_so_files = []	×
NEW 108	for ld_path_patt in _TYPICAL_LD_PATH_PATTERNS:	×
NEW 109	for packages_per_so_file in remaining[:]:	×
NEW 110	if PurePath(packages_per_so_file.so_file).parent.match(ld_path_patt):	×
NEW 111	packages_per_so_files.append(packages_per_so_file)	×
NEW 112	remaining.remove(packages_per_so_file)	×
113
NEW 114	return replace(self, packages_per_so_files=tuple(packages_per_so_files))	×
115
116
117	@dataclass(frozen=True)	2✔
118	class DebPackagesForSonames:	2✔
119	packages_for_sonames: tuple[DebPackagesForSoname, ...]	2✔
120
121	@classmethod	2✔
122	def from_dict(cls, raw: Mapping[str, Mapping[str, Iterable[str]]]) -> DebPackagesForSonames:	2✔
123	return cls(	1✔
124	tuple(
125	DebPackagesForSoname(
126	soname,
127	(
128	DebPackagesPerSoFile(so_file, packages)
129	for so_file, packages in files_to_packages.items()
130	),
131	)
132	for soname, files_to_packages in raw.items()
133	)
134	)
135
136	@property	2✔
137	def from_best_so_files(self) -> DebPackagesForSonames:	2✔
NEW 138	packages = []	×
NEW 139	for packages_for_soname in self.packages_for_sonames:	×
NEW 140	packages.append(packages_for_soname.from_best_so_files)	×
NEW 141	return DebPackagesForSonames(tuple(packages))	×
142
143
144	@rule	2✔
145	async def deb_search_for_sonames(	2✔
146	request: DebSearchForSonamesRequest,
147	) -> DebPackagesForSonames:
NEW 148	script = read_resource(_NATIVE_LIBS_DEB_PACKAGE, _SEARCH_FOR_SONAMES_SCRIPT)	×
NEW 149	if not script:	×
NEW 150	raise ValueError(	×
151	f"Unable to find source of {_SEARCH_FOR_SONAMES_SCRIPT!r} in {_NATIVE_LIBS_DEB_PACKAGE}"
152	)
153
NEW 154	script_content = FileContent(	×
155	path=_SEARCH_FOR_SONAMES_SCRIPT, content=script, is_executable=True
156	)
157
158	# Pull python and requirements versions from the pants venv since that is what the script is tested with.
NEW 159	pants_python = PythonExecutable.fingerprinted(	×
160	sys.executable, ".".join(map(str, sys.version_info[:3])).encode("utf8")
161	)
NEW 162	distributions_in_pants_venv: list[importlib.metadata.Distribution] = list(	×
163	find_matching_distributions(None)
164	)
NEW 165	constraints = tuple(f"{dist.name}=={dist.version}" for dist in distributions_in_pants_venv)	×
NEW 166	requirements = { # requirements (and transitive deps) are constrained to the versions in the pants venv	×
167	"aiohttp",
168	"aiohttp-retry",
169	"beautifulsoup4",
170	}
171
NEW 172	script_digest, venv_pex = await concurrently(	×
173	create_digest(CreateDigest([script_content])),
174	create_venv_pex(
175	**implicitly(
176	PexRequest(
177	output_filename=_PEX_NAME,
178	internal_only=True,
179	python=pants_python,
180	requirements=PexRequirements(
181	requirements,
182	constraints_strings=constraints,
183	description_of_origin=f"Requirements for {_PEX_NAME}:{_SEARCH_FOR_SONAMES_SCRIPT}",
184	),
185	)
186	)
187	),
188	)
189
NEW 190	result: FallibleProcessResult = await execute_process(	×
191	**implicitly(
192	VenvPexProcess(
193	venv_pex,
194	argv=(
195	script_content.path,
196	f"--user-agent-suffix=pants/{VERSION}",
197	f"--distro={request.distro}",
198	f"--distro-codename={request.distro_codename}",
199	f"--arch={request.debian_arch}",
200	*request.sonames,
201	),
202	input_digest=script_digest,
203	description=f"Search deb packages for sonames: {request.sonames}",
204	level=LogLevel.DEBUG,
205	)
206	)
207	)
208
NEW 209	if result.exit_code == 0:	×
NEW 210	packages = json.loads(result.stdout)	×
211	else:
212	# The search API returns 200 even if no results were found.
213	# A 4xx or 5xx error means we gave up retrying because the server is unavailable.
214	# TODO: Should this raise an error instead of just a warning?
NEW 215	logger.warning(result.stderr.decode("utf-8"))	×
NEW 216	packages = {}	×
217
NEW 218	deb_packages_for_sonames = DebPackagesForSonames.from_dict(packages)	×
NEW 219	if request.from_best_so_files:	×
NEW 220	return deb_packages_for_sonames.from_best_so_files	×
NEW 221	return deb_packages_for_sonames	×
222
223
224	def rules() -> Iterable[Rule \| UnionRule]:	2✔
225	return collect_rules()	2✔

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