19529437518

Committed 20 Nov 2025 07:44AM UTC coverage: 78.884% (-1.4%) from 80.302%

Build # 19529437518

Build Type

push

github

Committed by

web-flow

Commit Message

nfpm.native_libs: Add RPM package depends from packaged pex_binaries (#22899)

## PR Series Overview

This is the second in a series of PRs that introduces a new backend:
`pants.backend.npm.native_libs`
Initially, the backend will be available as:
`pants.backend.experimental.nfpm.native_libs`

I proposed this new backend (originally named `bindeps`) in discussion
#22396.

This backend will inspect ELF bin/lib files (like `lib*.so`) in packaged
contents (for this PR series, only in `pex_binary` targets) to identify
package dependency metadata and inject that metadata on the relevant
`nfpm_deb_package` or `nfpm_rpm_package` targets. Effectively, it will
provide an approximation of these native packager features:
- `rpm`: `rpmdeps` + `elfdeps`
- `deb`: `dh_shlibdeps` + `dpkg-shlibdeps` (These substitute
`${shlibs:Depends}` in debian control files have)

### Goal: Host-agnostic package builds

This pants backend is designed to be host-agnostic, like
[nFPM](https://nfpm.goreleaser.com/).

Native packaging tools are often restricted to a single release of a
single distro. Unlike native package builders, this new pants backend
does not use any of those distro-specific or distro-release-specific
utilities or local package databases. This new backend should be able to
run (help with building deb and rpm packages) anywhere that pants can
run (MacOS, rpm linux distros, deb linux distros, other linux distros,
docker, ...).

### Previous PRs in series

- #22873

## PR Overview

This PR adds rules in `nfpm.native_libs` to add package dependency
metadata to `nfpm_rpm_package`. The 2 new rules are:

- `inject_native_libs_dependencies_in_package_fields`:

    - An implementation of the polymorphic rule `inject_nfpm_package_fields`.
      This rule is low priority (`priority = 2`) so that in-repo plugins can
      override/augment what it injects. (See #22864)

    - Rule logic overview:
        - find any pex_binaries that will be packaged in an `nfpm_rpm_package`
   ... (continued)

Run Details

96 of 118 new or added lines in 3 files covered. (81.36%)

910 existing lines in 53 files now uncovered.

73897 of 93678 relevant lines covered (78.88%)

3.21 hits per line

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

84.21

/src/python/pants/backend/docker/utils.py

# Copyright 2021 Pants project contributors (see CONTRIBUTORS.md).
# Licensed under the Apache License, Version 2.0 (see LICENSE).

from __future__ import annotations

import difflib
import os.path
import re
from collections.abc import Callable, Iterable, Iterator, Sequence
from fnmatch import fnmatch
from typing import Self, TypeVar

from pants.help.maybe_color import MaybeColor
from pants.util.ordered_set import FrozenOrderedSet

_T = TypeVar("_T", bound="KeyValueSequenceUtil")


image_ref_regexp = re.compile(
    r"""
    ^
    # Optional registry.
    ((?P<registry>[^/:_ ]+:?[^/:_ ]*)/)?
    # Repository.
    (?P<repository>[^:@ \t\n\r\f\v]+)
    # Optionally with `:tag`.
    (:(?P<tag>[^@ ]+))?
    # Optionally with `@digest`.
    (@(?P<digest>\S+))?
    $
    """,
    re.VERBOSE,
)


class KeyValueSequenceUtil(FrozenOrderedSet[str]):
    @classmethod
    def from_strings(cls: type[_T], *strings: str, duplicates_must_match: bool = False) -> _T:
        """Takes all `KEY`/`KEY=VALUE` strings and dedupes by `KEY`.

        The last seen `KEY` wins in case of duplicates, unless `duplicates_must_match` is `True`, in
        which case all `VALUE`s must be equal, if present.
        """

        key_to_entry_and_value: dict[str, tuple[str, str | None]] = {}
        for entry in strings:
            key, has_value, value = entry.partition("=")
            if not duplicates_must_match:
                # Note that last entry with the same key wins.
                key_to_entry_and_value[key] = (entry, value if has_value else None)
            else:
                prev_entry, prev_value = key_to_entry_and_value.get(key, (None, None))
                if prev_entry is None:
                    # Not seen before.
                    key_to_entry_and_value[key] = (entry, value if has_value else None)
                elif not has_value:
                    # Seen before, no new value, so keep existing.
                    pass
                elif prev_value is None:
                    # Update value.
                    key_to_entry_and_value[key] = (entry, value)
                elif prev_value != value:
                    # Seen before with a different value.
                    raise ValueError(
                        f"{cls.__name__}: duplicated {key!r} with different values: "
                        f"{prev_value!r} != {value!r}."
                    )

        deduped_entries = sorted(
            entry_and_value[0] for entry_and_value in key_to_entry_and_value.values()
        )
        return cls(FrozenOrderedSet(deduped_entries))

    @classmethod
    def from_dict(cls, value: dict[str, str | None]) -> Self:
        def encode_kv(k: str, v: str | None) -> str:
            if v is None:
                return k
            return "=".join((k, v))

        return cls(FrozenOrderedSet(sorted(encode_kv(k, v) for k, v in value.items())))

    def to_dict(
        self, default: Callable[[str], str | None] = lambda x: None
    ) -> dict[str, str | None]:
        return {
            key: value if has_value else default(key)
            for key, has_value, value in [pair.partition("=") for pair in self]
        }


def suggest_renames(
    tentative_paths: Iterable[str], actual_files: Sequence[str], actual_dirs: Sequence[str]
) -> Iterator[tuple[str, str]]:
    """Return each pair of `tentative_paths` matched to the best possible match of `actual_paths`
    that are not an exact match.

    A pair of `(tentative_path, "")` means there were no possible match to find in the
    `actual_paths`, while a pair of `("", actual_path)` indicates a file in the build context that
    is not taking part in any `COPY` instruction.
    """

    actual_paths = (*actual_files, *actual_dirs)
    referenced: dict[str, set[str] | bool] = {}

    def reference(path: str) -> None:
        """Track which actual files has been referenced either explicitly by a tentative path, or as
        a suggested rename."""
        if path in actual_dirs:
            referenced[path] = True
        else:
            dirname = os.path.dirname(path)
            refs = referenced.setdefault(dirname, set())
            if isinstance(refs, set):
                refs.add(path)

        # Recalculate possible matches, to avoid suggesting the same file twice.
        nonlocal actual_paths
        actual_paths = tuple(get_unreferenced(actual_files, actual_dirs))

    def is_referenced(path: str, dirname: str | None = None) -> bool:
        """Check the list of referenced files to see if `path` has been flagged.

        Walks up the directory tree in case there is a recursive flag on one of the parent
        directories.
        """
        if dirname is None:
            dirname = os.path.dirname(path)
        refs = referenced.get(dirname, set())
        if isinstance(refs, bool):
            return refs
        if path in refs:
            return True
        parentdir = os.path.dirname(dirname)
        if parentdir:
            return is_referenced(path, parentdir)
        return False

    def get_unreferenced(files: Sequence[str] = (), dirs: Sequence[str] = ()) -> Iterator[str]:
        unreferenced_files = tuple(path for path in files if not is_referenced(path))
        yield from unreferenced_files
        for path in dirs:
            if not any(filename.startswith(path + "/") for filename in unreferenced_files):
                # Skip paths where we don't have any unreferenced files any longer.
                continue
            if not is_referenced(path, path):
                yield path

    def get_matches(path: str) -> tuple[str, ...]:
        is_pattern = any(all(c in path for c in cs) for cs in ["*", "?", "[]"])
        if not is_pattern:
            return (path,) if path in actual_paths else ()
        #
        # NOTICE: There is a slight difference in the pattern syntax used for the Dockerfile `COPY`
        # instruction, than what is implemented by the `fnmatch` function in Python.
        # https://docs.docker.com/engine/reference/builder/#copy which is implemented using
        # https://golang.org/pkg/path/filepath#Match compared to
        # https://docs.python.org/3/library/fnmatch.html#fnmatch.fnmatch
        #
        # For best experience when using globs, this should be addressed, but for now, I'll settle
        # for a "close enough" approximation to get this moving.
        return tuple(p for p in actual_paths if fnmatch(p, path))

    # Go over exact matches first, so we don't target them as possible matches for renames.
    unmatched_paths = set()
    for path in tentative_paths:
        matches = get_matches(path)
        for match in matches:
            reference(match)
        if not matches:
            unmatched_paths.add(path)

    # List unknown files, possibly with a rename suggestion.
    for path in sorted(unmatched_paths):
        for suggestion in difflib.get_close_matches(path, actual_paths, n=1, cutoff=0.1):
            # Suggest rename to match what files there are.
            reference(suggestion)
            yield path, suggestion
            break
        else:
            # No match for this path.
            yield path, ""

    # List unused files.
    for path in sorted(get_unreferenced(actual_files)):
        yield "", path


def format_rename_suggestion(src_path: str, dst_path: str, *, colors: bool) -> str:
    """Given two paths, formats a line showing what to change in `src_path` to get to `dst_path`."""
    color = MaybeColor(colors)
    rem = color.maybe_red(src_path)
    add = color.maybe_green(dst_path)
    return f"{rem} => {add}"

1	# Copyright 2021 Pants project contributors (see CONTRIBUTORS.md).
2	# Licensed under the Apache License, Version 2.0 (see LICENSE).
3
4	from __future__ import annotations	8✔
5
6	import difflib	8✔
7	import os.path	8✔
8	import re	8✔
9	from collections.abc import Callable, Iterable, Iterator, Sequence	8✔
10	from fnmatch import fnmatch	8✔
11	from typing import Self, TypeVar	8✔
12
13	from pants.help.maybe_color import MaybeColor	8✔
14	from pants.util.ordered_set import FrozenOrderedSet	8✔
15
16	_T = TypeVar("_T", bound="KeyValueSequenceUtil")	8✔
17
18
19	image_ref_regexp = re.compile(	8✔
20	r"""
21	^
22	# Optional registry.
23	((?P<registry>[^/:_ ]+:?[^/:_ ]*)/)?
24	# Repository.
25	(?P<repository>[^:@ \t\n\r\f\v]+)
26	# Optionally with `:tag`.
27	(:(?P<tag>[^@ ]+))?
28	# Optionally with `@digest`.
29	(@(?P<digest>\S+))?
30	$
31	""",
32	re.VERBOSE,
33	)
34
35
36	class KeyValueSequenceUtil(FrozenOrderedSet[str]):	8✔
37	@classmethod	8✔
38	def from_strings(cls: type[_T], *strings: str, duplicates_must_match: bool = False) -> _T:	8✔
39	"""Takes all `KEY`/`KEY=VALUE` strings and dedupes by `KEY`.
40
41	The last seen `KEY` wins in case of duplicates, unless `duplicates_must_match` is `True`, in
42	which case all `VALUE`s must be equal, if present.
43	"""
44
45	key_to_entry_and_value: dict[str, tuple[str, str \| None]] = {}	2✔
46	for entry in strings:	2✔
47	key, has_value, value = entry.partition("=")	2✔
48	if not duplicates_must_match:	2✔
49	# Note that last entry with the same key wins.
50	key_to_entry_and_value[key] = (entry, value if has_value else None)	2✔
51	else:
52	prev_entry, prev_value = key_to_entry_and_value.get(key, (None, None))	×
53	if prev_entry is None:	×
54	# Not seen before.
55	key_to_entry_and_value[key] = (entry, value if has_value else None)	×
56	elif not has_value:	×
57	# Seen before, no new value, so keep existing.
58	pass	×
59	elif prev_value is None:	×
60	# Update value.
61	key_to_entry_and_value[key] = (entry, value)	×
62	elif prev_value != value:	×
63	# Seen before with a different value.
64	raise ValueError(	×
65	f"{cls.__name__}: duplicated {key!r} with different values: "
66	f"{prev_value!r} != {value!r}."
67	)
68
69	deduped_entries = sorted(	2✔
70	entry_and_value[0] for entry_and_value in key_to_entry_and_value.values()
71	)
72	return cls(FrozenOrderedSet(deduped_entries))	2✔
73
74	@classmethod	8✔
75	def from_dict(cls, value: dict[str, str \| None]) -> Self:	8✔
76	def encode_kv(k: str, v: str \| None) -> str:	×
77	if v is None:	×
78	return k	×
79	return "=".join((k, v))	×
80
81	return cls(FrozenOrderedSet(sorted(encode_kv(k, v) for k, v in value.items())))	×
82
83	def to_dict(	8✔
84	self, default: Callable[[str], str \| None] = lambda x: None
85	) -> dict[str, str \| None]:
UNCOV 86	return {	×
87	key: value if has_value else default(key)
88	for key, has_value, value in [pair.partition("=") for pair in self]
89	}
90
91
92	def suggest_renames(	8✔
93	tentative_paths: Iterable[str], actual_files: Sequence[str], actual_dirs: Sequence[str]
94	) -> Iterator[tuple[str, str]]:
95	"""Return each pair of `tentative_paths` matched to the best possible match of `actual_paths`
96	that are not an exact match.
97
98	A pair of `(tentative_path, "")` means there were no possible match to find in the
99	`actual_paths`, while a pair of `("", actual_path)` indicates a file in the build context that
100	is not taking part in any `COPY` instruction.
101	"""
102
103	actual_paths = (actual_files, actual_dirs)	3✔
104	referenced: dict[str, set[str] \| bool] = {}	3✔
105
106	def reference(path: str) -> None:	3✔
107	"""Track which actual files has been referenced either explicitly by a tentative path, or as
108	a suggested rename."""
109	if path in actual_dirs:	2✔
110	referenced[path] = True	1✔
111	else:
112	dirname = os.path.dirname(path)	2✔
113	refs = referenced.setdefault(dirname, set())	2✔
114	if isinstance(refs, set):	2✔
115	refs.add(path)	2✔
116
117	# Recalculate possible matches, to avoid suggesting the same file twice.
118	nonlocal actual_paths
119	actual_paths = tuple(get_unreferenced(actual_files, actual_dirs))	2✔
120
121	def is_referenced(path: str, dirname: str \| None = None) -> bool:	3✔
122	"""Check the list of referenced files to see if `path` has been flagged.
123
124	Walks up the directory tree in case there is a recursive flag on one of the parent
125	directories.
126	"""
127	if dirname is None:	2✔
128	dirname = os.path.dirname(path)	2✔
129	refs = referenced.get(dirname, set())	2✔
130	if isinstance(refs, bool):	2✔
131	return refs	1✔
132	if path in refs:	2✔
133	return True	2✔
134	parentdir = os.path.dirname(dirname)	2✔
135	if parentdir:	2✔
136	return is_referenced(path, parentdir)	2✔
137	return False	2✔
138
139	def get_unreferenced(files: Sequence[str] = (), dirs: Sequence[str] = ()) -> Iterator[str]:	3✔
140	unreferenced_files = tuple(path for path in files if not is_referenced(path))	3✔
141	yield from unreferenced_files	3✔
142	for path in dirs:	3✔
143	if not any(filename.startswith(path + "/") for filename in unreferenced_files):	2✔
144	# Skip paths where we don't have any unreferenced files any longer.
145	continue	2✔
146	if not is_referenced(path, path):	2✔
147	yield path	2✔
148
149	def get_matches(path: str) -> tuple[str, ...]:	3✔
150	is_pattern = any(all(c in path for c in cs) for cs in ["*", "?", "[]"])	3✔
151	if not is_pattern:	3✔
152	return (path,) if path in actual_paths else ()	3✔
153	#
154	# NOTICE: There is a slight difference in the pattern syntax used for the Dockerfile `COPY`
155	# instruction, than what is implemented by the `fnmatch` function in Python.
156	# https://docs.docker.com/engine/reference/builder/#copy which is implemented using
157	# https://golang.org/pkg/path/filepath#Match compared to
158	# https://docs.python.org/3/library/fnmatch.html#fnmatch.fnmatch
159	#
160	# For best experience when using globs, this should be addressed, but for now, I'll settle
161	# for a "close enough" approximation to get this moving.
162	return tuple(p for p in actual_paths if fnmatch(p, path))	1✔
163
164	# Go over exact matches first, so we don't target them as possible matches for renames.
165	unmatched_paths = set()	3✔
166	for path in tentative_paths:	3✔
167	matches = get_matches(path)	3✔
168	for match in matches:	3✔
169	reference(match)	2✔
170	if not matches:	3✔
171	unmatched_paths.add(path)	3✔
172
173	# List unknown files, possibly with a rename suggestion.
174	for path in sorted(unmatched_paths):	3✔
175	for suggestion in difflib.get_close_matches(path, actual_paths, n=1, cutoff=0.1):	3✔
176	# Suggest rename to match what files there are.
177	reference(suggestion)	2✔
178	yield path, suggestion	2✔
179	break	2✔
180	else:
181	# No match for this path.
182	yield path, ""	3✔
183
184	# List unused files.
185	for path in sorted(get_unreferenced(actual_files)):	3✔
186	yield "", path	2✔
187
188
189	def format_rename_suggestion(src_path: str, dst_path: str, *, colors: bool) -> str:	8✔
190	"""Given two paths, formats a line showing what to change in `src_path` to get to `dst_path`."""
191	color = MaybeColor(colors)	2✔
192	rem = color.maybe_red(src_path)	2✔
193	add = color.maybe_green(dst_path)	2✔
194	return f"{rem} => {add}"	2✔

pantsbuild / pants / 19529437518

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