19881177903

Committed 03 Dec 2025 03:19AM UTC coverage: 80.282% (-0.009%) from 80.291%

Build # 19881177903

Build Type

Pull #22904

github

Committed by

web-flow

Commit Message

Merge 606163c53 into 1f2bc5397

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

Run Details

179 of 230 new or added lines in 9 files covered. (77.83%)

3 existing lines in 2 files now uncovered.

78565 of 97861 relevant lines covered (80.28%)

3.35 hits per line

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

68.0

/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.nfpm.native_libs.deb.subsystem import DebSubsystem
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
from pants.engine.process import ProcessResult, execute_process_or_raise
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
    "/lib/*-linux-*/",
    "/usr/lib/*-linux-*/",
    # Then look for a generic system libs
    "/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,
    deb_subsystem: DebSubsystem,
) -> 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}",
                    ),
                )
            )
        ),
    )

    distro_package_search_url = deb_subsystem.options.distro_package_search_urls[request.distro]

    # Raising an error means we gave up retrying because the server is unavailable.
    result: ProcessResult = await execute_process_or_raise(
        **implicitly(
            VenvPexProcess(
                venv_pex,
                argv=(
                    script_content.path,
                    f"--user-agent-suffix=pants/{VERSION}",
                    f"--search-url={distro_package_search_url}",
                    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,
            )
        )
    )

    packages = json.loads(result.stdout)
    if result.stderr:
        logger.warning(result.stderr.decode("utf-8"))

    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 (
        *DebSubsystem.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	1✔
5
6	import importlib.metadata	1✔
7	import json	1✔
8	import logging	1✔
9	import sys	1✔
10	from collections.abc import Iterable, Mapping	1✔
11	from dataclasses import dataclass, replace	1✔
12	from pathlib import PurePath	1✔
13
14	from pants.backend.nfpm.native_libs.deb.subsystem import DebSubsystem	1✔
15	from pants.backend.python.util_rules.pex import PexRequest, VenvPexProcess, create_venv_pex	1✔
16	from pants.backend.python.util_rules.pex_environment import PythonExecutable	1✔
17	from pants.backend.python.util_rules.pex_requirements import PexRequirements	1✔
18	from pants.engine.fs import CreateDigest, FileContent	1✔
19	from pants.engine.internals.native_engine import UnionRule	1✔
20	from pants.engine.internals.selectors import concurrently	1✔
21	from pants.engine.intrinsics import create_digest	1✔
22	from pants.engine.process import ProcessResult, execute_process_or_raise	1✔
23	from pants.engine.rules import Rule, collect_rules, implicitly, rule	1✔
24	from pants.init.import_util import find_matching_distributions	1✔
25	from pants.util.logging import LogLevel	1✔
26	from pants.util.resources import read_resource	1✔
27	from pants.version import VERSION	1✔
28
29	logger = logging.getLogger(__name__)	1✔
30
31	_NATIVE_LIBS_DEB_PACKAGE = "pants.backend.nfpm.native_libs.deb"	1✔
32	_SEARCH_FOR_SONAMES_SCRIPT = "search_for_sonames.py"	1✔
33	_PEX_NAME = "native_libs_deb.pex"	1✔
34
35
36	@dataclass(frozen=True)	1✔
37	class DebSearchForSonamesRequest:	1✔
38	distro: str	1✔
39	distro_codename: str	1✔
40	debian_arch: str	1✔
41	sonames: tuple[str, ...]	1✔
42	from_best_so_files: bool	1✔
43
44	def __init__(	1✔
45	self,
46	distro: str,
47	distro_codename: str,
48	debian_arch: str,
49	sonames: Iterable[str],
50	*,
51	from_best_so_files: bool = False,
52	):
53	object.__setattr__(self, "distro", distro)	1✔
54	object.__setattr__(self, "distro_codename", distro_codename)	1✔
55	object.__setattr__(self, "debian_arch", debian_arch)	1✔
56	object.__setattr__(self, "sonames", tuple(sorted(sonames)))	1✔
57	object.__setattr__(self, "from_best_so_files", from_best_so_files)	1✔
58
59
60	@dataclass(frozen=True)	1✔
61	class DebPackagesPerSoFile:	1✔
62	so_file: str	1✔
63	packages: tuple[str, ...]	1✔
64
65	def __init__(self, so_file: str, packages: Iterable[str]):	1✔
66	object.__setattr__(self, "so_file", so_file)	1✔
67	object.__setattr__(self, "packages", tuple(sorted(packages)))	1✔
68
69
70	_TYPICAL_LD_PATH_PATTERNS = (	1✔
71	# platform specific system libs (like libc) get selected first
72	"/lib/-linux-/",
73	"/usr/lib/-linux-/",
74	# Then look for a generic system libs
75	"/lib/",
76	"/usr/lib/",
77	# Anything else has to be added manually to dependencies.
78	# These rules cannot use symbols or shlibs metadata to inform package selection.
79	)
80
81
82	@dataclass(frozen=True)	1✔
83	class DebPackagesForSoname:	1✔
84	soname: str	1✔
85	packages_per_so_files: tuple[DebPackagesPerSoFile, ...]	1✔
86
87	def __init__(self, soname: str, packages_per_so_files: Iterable[DebPackagesPerSoFile]):	1✔
88	object.__setattr__(self, "soname", soname)	1✔
89	object.__setattr__(self, "packages_per_so_files", tuple(packages_per_so_files))	1✔
90
91	@property	1✔
92	def from_best_so_files(self) -> DebPackagesForSoname:	1✔
93	"""Pick best so_files from packages_for_so_files using a simplified ld.so-like algorithm.
94
95	The most preferred is first. This is NOT a recursive match; Only match if direct child of
96	ld_path_patt dir. Anything that uses a subdir like /usr/lib/<app>/lib.so. uses rpath to
97	prefer the app's libs over system libs. If this vastly simplified form of ld.so-style
98	matching does not select the correct libs, then the package(s) that provide the shared lib
99	should be added manually to the nfpm requires field.
100	"""
NEW 101	if len(self.packages_per_so_files) <= 1: # shortcut; no filtering required for 0-1 results.	×
NEW 102	return self	×
103
NEW 104	remaining = list(self.packages_per_so_files)	×
105
NEW 106	packages_per_so_files = []	×
NEW 107	for ld_path_patt in _TYPICAL_LD_PATH_PATTERNS:	×
NEW 108	for packages_per_so_file in remaining[:]:	×
NEW 109	if PurePath(packages_per_so_file.so_file).parent.match(ld_path_patt):	×
NEW 110	packages_per_so_files.append(packages_per_so_file)	×
NEW 111	remaining.remove(packages_per_so_file)	×
112
NEW 113	return replace(self, packages_per_so_files=tuple(packages_per_so_files))	×
114
115
116	@dataclass(frozen=True)	1✔
117	class DebPackagesForSonames:	1✔
118	packages_for_sonames: tuple[DebPackagesForSoname, ...]	1✔
119
120	@classmethod	1✔
121	def from_dict(cls, raw: Mapping[str, Mapping[str, Iterable[str]]]) -> DebPackagesForSonames:	1✔
122	return cls(	1✔
123	tuple(
124	DebPackagesForSoname(
125	soname,
126	(
127	DebPackagesPerSoFile(so_file, packages)
128	for so_file, packages in files_to_packages.items()
129	),
130	)
131	for soname, files_to_packages in raw.items()
132	)
133	)
134
135	@property	1✔
136	def from_best_so_files(self) -> DebPackagesForSonames:	1✔
NEW 137	packages = []	×
NEW 138	for packages_for_soname in self.packages_for_sonames:	×
NEW 139	packages.append(packages_for_soname.from_best_so_files)	×
NEW 140	return DebPackagesForSonames(tuple(packages))	×
141
142
143	@rule	1✔
144	async def deb_search_for_sonames(	1✔
145	request: DebSearchForSonamesRequest,
146	deb_subsystem: DebSubsystem,
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	distro_package_search_url = deb_subsystem.options.distro_package_search_urls[request.distro]	×
191
192	# Raising an error means we gave up retrying because the server is unavailable.
NEW 193	result: ProcessResult = await execute_process_or_raise(	×
194	**implicitly(
195	VenvPexProcess(
196	venv_pex,
197	argv=(
198	script_content.path,
199	f"--user-agent-suffix=pants/{VERSION}",
200	f"--search-url={distro_package_search_url}",
201	f"--distro={request.distro}",
202	f"--distro-codename={request.distro_codename}",
203	f"--arch={request.debian_arch}",
204	*request.sonames,
205	),
206	input_digest=script_digest,
207	description=f"Search deb packages for sonames: {request.sonames}",
208	level=LogLevel.DEBUG,
209	)
210	)
211	)
212
NEW 213	packages = json.loads(result.stdout)	×
NEW 214	if result.stderr:	×
NEW 215	logger.warning(result.stderr.decode("utf-8"))	×
216
NEW 217	deb_packages_for_sonames = DebPackagesForSonames.from_dict(packages)	×
NEW 218	if request.from_best_so_files:	×
NEW 219	return deb_packages_for_sonames.from_best_so_files	×
NEW 220	return deb_packages_for_sonames	×
221
222
223	def rules() -> Iterable[Rule \| UnionRule]:	1✔
224	return (	1✔
225	*DebSubsystem.rules(),
226	*collect_rules(),
227	)

pantsbuild / pants / 19881177903

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