19186107394

Committed 08 Nov 2025 01:49AM UTC coverage: 80.298% (-0.01%) from 80.31%

Build # 19186107394

Build Type

push

github

Committed by

web-flow

Commit Message

nfpm: support chaining multiple `InjectNfpmPackageFieldsRequest`s (sorted w/ simple priority integer) (#22864)

This extends then `pants.backend.experimental.nfpm` backend's plugin
API, allowing multiple implementations of this polymorphic-rule plugin
hook (before this change, only one implementation was allowed):

- `inject_nfpm_package_fields(InjectNfpmPackageFieldsRequest) -> InjectedNfpmPackageFields`

Unlike many polymorphic/union rules, each rule runs sequentially instead
of concurrently. This is conceptually similar to middleware, because the
request passed to each rule includes the results of the previous (lower
priority) rule. So, one rule can override or extend fields that were
injected in previous rules, not just the original fields from the
BUILD-file-based `nfpm_*_package` target.

I tried several approaches to sorting the request classes, but most were
error prone or simply didn't work. Heuristic-based sorting proved to be
complex and didn't work in all of my test cases (eg one heuristic was
looking at the class's module to make pants-provided requests lower
priority). A simple `priority` integer is far simpler and more reliable,
so that's what this implements.

Another thing that didn't work was adding `__lt__` as a `@classmethod`
on the abstract `InjectNfpmPackageFieldsRequest`. That fails because
`sorted()` requires an instance method, not a class method. The only way
I found to add an `__lt__` method on a class is via a metaclass. So,
this adds a `_PrioritizedSortableClassMetaclass` to
`InjectNfpmPackageFieldsRequest` allowing for simple `OneRequest <
TwoRequest` sorting of the class type like this:

```diff
     inject_nfpm_config_request_types = union_membership.get(InjectNfpmPackageFieldsRequest)
     applicable_inject_nfpm_config_request_types = tuple(
-        request
-        for request in inject_nfpm_config_request_types
-        if request.is_applicable(target)
+        sorted(
+            request_type
+            for requ... (continued)

Run Details

41 of 74 new or added lines in 2 files covered. (55.41%)

1 existing line in 1 file now uncovered.

78098 of 97260 relevant lines covered (80.3%)

3.36 hits per line

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

69.33

/src/python/pants/backend/nfpm/util_rules/inject_config_test.py

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

from __future__ import annotations

from textwrap import dedent
from typing import Any

import pytest

from pants.backend.nfpm.fields.all import NfpmPlatformField
from pants.backend.nfpm.fields.version import NfpmVersionField, NfpmVersionReleaseField
from pants.backend.nfpm.target_types import target_types as nfpm_target_types
from pants.backend.nfpm.target_types_rules import rules as nfpm_target_types_rules
from pants.backend.nfpm.util_rules.inject_config import (
    InjectedNfpmPackageFields,
    InjectNfpmPackageFieldsRequest,
    NfpmPackageTargetWrapper,
)
from pants.backend.nfpm.util_rules.inject_config import rules as nfpm_inject_config_rules
from pants.engine.internals.native_engine import Address, Field
from pants.engine.rules import QueryRule, rule
from pants.engine.unions import UnionRule
from pants.testutil.rule_runner import RuleRunner

_PKG_NAME = "pkg"


class PluginInjectFieldsRequest(InjectNfpmPackageFieldsRequest):
    @classmethod
    def is_applicable(cls, _) -> bool:
        return True


@rule
async def inject_nfpm_package_fields_plugin(
    request: PluginInjectFieldsRequest,
) -> InjectedNfpmPackageFields:
    address = request.target.address
    # preserve fields from earlier rules in chain
    fields: list[Field] = list(request.injected_fields.values())
    if NfpmVersionField not in request.injected_fields:
        fields.extend(
            [
                NfpmVersionField("9.8.7-dev+git", address),
                NfpmVersionReleaseField(6, address),
            ]
        )
    return InjectedNfpmPackageFields(fields, address=address)


class SubclassPluginInjectFieldsRequest(PluginInjectFieldsRequest):
    pass


class AnotherSubclassPluginInjectFieldsRequest(PluginInjectFieldsRequest):
    pass


@rule
async def inject_nfpm_package_fields_subclass(
    request: SubclassPluginInjectFieldsRequest,
) -> InjectedNfpmPackageFields:
    address = request.target.address
    # preserve fields from earlier rules in chain
    fields: list[Field] = list(request.injected_fields.values())
    if not fields or NfpmVersionReleaseField in request.injected_fields:
        release = 0
        if NfpmVersionReleaseField in request.injected_fields:
            old_release = request.injected_fields[NfpmVersionReleaseField].value
            assert old_release is not None
            release = 10 + old_release
        fields.append(NfpmVersionReleaseField(release, address))
    return InjectedNfpmPackageFields(fields, address=address)


class HighPriorityInjectFieldsRequest(InjectNfpmPackageFieldsRequest):
    priority = 100

    @classmethod
    def is_applicable(cls, _) -> bool:
        return True


@rule
async def inject_nfpm_package_fields_high_priority(
    request: HighPriorityInjectFieldsRequest,
) -> InjectedNfpmPackageFields:
    address = request.target.address
    # preserve fields from earlier rules in chain
    fields: list[Field] = list(request.injected_fields.values())
    if not fields or NfpmVersionField not in request.injected_fields:
        fields.extend(
            [
                NfpmVersionField("9.9.9-dev+git", address),
                NfpmVersionReleaseField(9, address),
            ]
        )
    # This high priority implementation wants to force Platform to always be "foobar"
    # even if a lower priority rule injected NfpmPlatformField with a different value.
    fields.append(NfpmPlatformField("foobar", address))
    return InjectedNfpmPackageFields(fields, address=address)


@pytest.mark.parametrize(
    "a,b,expected_lt",
    (
        # HighPriority* > SubclassPlugin* > Plugin*
        (HighPriorityInjectFieldsRequest, SubclassPluginInjectFieldsRequest, False),
        (HighPriorityInjectFieldsRequest, PluginInjectFieldsRequest, False),
        (SubclassPluginInjectFieldsRequest, HighPriorityInjectFieldsRequest, True),
        (SubclassPluginInjectFieldsRequest, PluginInjectFieldsRequest, False),
        (PluginInjectFieldsRequest, HighPriorityInjectFieldsRequest, True),
        (PluginInjectFieldsRequest, SubclassPluginInjectFieldsRequest, True),
        # Subclass* > AnotherSubclass* : with same priority, sort uses name of module+class
        (SubclassPluginInjectFieldsRequest, AnotherSubclassPluginInjectFieldsRequest, False),
        (AnotherSubclassPluginInjectFieldsRequest, SubclassPluginInjectFieldsRequest, True),
        # self is equal which is not less than
        (HighPriorityInjectFieldsRequest, HighPriorityInjectFieldsRequest, False),
    ),
)
def test_nfpm_package_fields_request_class_sort(a: Any, b: Any, expected_lt):
    ret = a < b
    assert ret == expected_lt, f"a.priority={a.priority} b.priority={b.priority}"


@pytest.fixture
def rule_runner() -> RuleRunner:
    rule_runner = RuleRunner(
        target_types=[
            *nfpm_target_types(),
        ],
        rules=[
            *nfpm_target_types_rules(),
            *nfpm_inject_config_rules(),
            inject_nfpm_package_fields_plugin,
            inject_nfpm_package_fields_subclass,
            inject_nfpm_package_fields_high_priority,
            UnionRule(InjectNfpmPackageFieldsRequest, PluginInjectFieldsRequest),
            UnionRule(InjectNfpmPackageFieldsRequest, SubclassPluginInjectFieldsRequest),
            UnionRule(InjectNfpmPackageFieldsRequest, HighPriorityInjectFieldsRequest),
            QueryRule(InjectedNfpmPackageFields, (NfpmPackageTargetWrapper,)),
        ],
    )
    rule_runner.set_options([], env_inherit={"PATH", "PYENV_ROOT", "HOME"})
    return rule_runner


@pytest.mark.parametrize(
    "packager",
    (
        "apk",
        "archlinux",
        "deb",
        "rpm",
    ),
)
def test_determine_injected_nfpm_package_fields(rule_runner: RuleRunner, packager: str) -> None:
    packager = "deb"
    rule_runner.write_files(
        {
            "BUILD": dedent(
                f"""
                nfpm_{packager}_package(
                    name="{_PKG_NAME}",
                    description="A {packager} package",
                    package_name="{_PKG_NAME}",
                    version="",  # the plugin should provide this
                    {"" if packager != "deb" else 'maintainer="Foo Bar <deb@example.com>",'}
                    dependencies=[],
                )
                """
            ),
        }
    )
    target = rule_runner.get_target(Address("", target_name=_PKG_NAME))
    result = rule_runner.request(InjectedNfpmPackageFields, [NfpmPackageTargetWrapper(target)])
    field_values = result.field_values
    assert len(field_values) == 3
    assert field_values[NfpmVersionField].value == "9.8.7-dev+git"  # (Plugin*)
    assert field_values[NfpmVersionReleaseField].value == 16  # 6 (Plugin*) + 10 (SubclassPlugin*)
    assert field_values[NfpmPlatformField].value == "foobar"  # (HighPriority*)

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	from textwrap import dedent	1✔
7	from typing import Any	1✔
8
9	import pytest	1✔
10
11	from pants.backend.nfpm.fields.all import NfpmPlatformField	1✔
12	from pants.backend.nfpm.fields.version import NfpmVersionField, NfpmVersionReleaseField	1✔
13	from pants.backend.nfpm.target_types import target_types as nfpm_target_types	1✔
14	from pants.backend.nfpm.target_types_rules import rules as nfpm_target_types_rules	1✔
15	from pants.backend.nfpm.util_rules.inject_config import (	1✔
16	InjectedNfpmPackageFields,
17	InjectNfpmPackageFieldsRequest,
18	NfpmPackageTargetWrapper,
19	)
20	from pants.backend.nfpm.util_rules.inject_config import rules as nfpm_inject_config_rules	1✔
21	from pants.engine.internals.native_engine import Address, Field	1✔
22	from pants.engine.rules import QueryRule, rule	1✔
23	from pants.engine.unions import UnionRule	1✔
24	from pants.testutil.rule_runner import RuleRunner	1✔
25
26	_PKG_NAME = "pkg"	1✔
27
28
29	class PluginInjectFieldsRequest(InjectNfpmPackageFieldsRequest):	1✔
30	@classmethod	1✔
31	def is_applicable(cls, _) -> bool:	1✔
32	return True	×
33
34
35	@rule	1✔
36	async def inject_nfpm_package_fields_plugin(	1✔
37	request: PluginInjectFieldsRequest,
38	) -> InjectedNfpmPackageFields:
39	address = request.target.address	×
40	# preserve fields from earlier rules in chain
NEW 41	fields: list[Field] = list(request.injected_fields.values())	×
NEW 42	if NfpmVersionField not in request.injected_fields:	×
NEW 43	fields.extend(	×
44	[
45	NfpmVersionField("9.8.7-dev+git", address),
46	NfpmVersionReleaseField(6, address),
47	]
48	)
UNCOV 49	return InjectedNfpmPackageFields(fields, address=address)	×
50
51
52	class SubclassPluginInjectFieldsRequest(PluginInjectFieldsRequest):	1✔
53	pass	1✔
54
55
56	class AnotherSubclassPluginInjectFieldsRequest(PluginInjectFieldsRequest):	1✔
57	pass	1✔
58
59
60	@rule	1✔
61	async def inject_nfpm_package_fields_subclass(	1✔
62	request: SubclassPluginInjectFieldsRequest,
63	) -> InjectedNfpmPackageFields:
NEW 64	address = request.target.address	×
65	# preserve fields from earlier rules in chain
NEW 66	fields: list[Field] = list(request.injected_fields.values())	×
NEW 67	if not fields or NfpmVersionReleaseField in request.injected_fields:	×
NEW 68	release = 0	×
NEW 69	if NfpmVersionReleaseField in request.injected_fields:	×
NEW 70	old_release = request.injected_fields[NfpmVersionReleaseField].value	×
NEW 71	assert old_release is not None	×
NEW 72	release = 10 + old_release	×
NEW 73	fields.append(NfpmVersionReleaseField(release, address))	×
NEW 74	return InjectedNfpmPackageFields(fields, address=address)	×
75
76
77	class HighPriorityInjectFieldsRequest(InjectNfpmPackageFieldsRequest):	1✔
78	priority = 100	1✔
79
80	@classmethod	1✔
81	def is_applicable(cls, _) -> bool:	1✔
NEW 82	return True	×
83
84
85	@rule	1✔
86	async def inject_nfpm_package_fields_high_priority(	1✔
87	request: HighPriorityInjectFieldsRequest,
88	) -> InjectedNfpmPackageFields:
NEW 89	address = request.target.address	×
90	# preserve fields from earlier rules in chain
NEW 91	fields: list[Field] = list(request.injected_fields.values())	×
NEW 92	if not fields or NfpmVersionField not in request.injected_fields:	×
NEW 93	fields.extend(	×
94	[
95	NfpmVersionField("9.9.9-dev+git", address),
96	NfpmVersionReleaseField(9, address),
97	]
98	)
99	# This high priority implementation wants to force Platform to always be "foobar"
100	# even if a lower priority rule injected NfpmPlatformField with a different value.
NEW 101	fields.append(NfpmPlatformField("foobar", address))	×
NEW 102	return InjectedNfpmPackageFields(fields, address=address)	×
103
104
105	@pytest.mark.parametrize(	1✔
106	"a,b,expected_lt",
107	(
108	# HighPriority* > SubclassPlugin* > Plugin*
109	(HighPriorityInjectFieldsRequest, SubclassPluginInjectFieldsRequest, False),
110	(HighPriorityInjectFieldsRequest, PluginInjectFieldsRequest, False),
111	(SubclassPluginInjectFieldsRequest, HighPriorityInjectFieldsRequest, True),
112	(SubclassPluginInjectFieldsRequest, PluginInjectFieldsRequest, False),
113	(PluginInjectFieldsRequest, HighPriorityInjectFieldsRequest, True),
114	(PluginInjectFieldsRequest, SubclassPluginInjectFieldsRequest, True),
115	# Subclass* > AnotherSubclass* : with same priority, sort uses name of module+class
116	(SubclassPluginInjectFieldsRequest, AnotherSubclassPluginInjectFieldsRequest, False),
117	(AnotherSubclassPluginInjectFieldsRequest, SubclassPluginInjectFieldsRequest, True),
118	# self is equal which is not less than
119	(HighPriorityInjectFieldsRequest, HighPriorityInjectFieldsRequest, False),
120	),
121	)
122	def test_nfpm_package_fields_request_class_sort(a: Any, b: Any, expected_lt):	1✔
123	ret = a < b	1✔
124	assert ret == expected_lt, f"a.priority={a.priority} b.priority={b.priority}"	1✔
125
126
127	@pytest.fixture	1✔
128	def rule_runner() -> RuleRunner:	1✔
129	rule_runner = RuleRunner(	1✔
130	target_types=[
131	*nfpm_target_types(),
132	],
133	rules=[
134	*nfpm_target_types_rules(),
135	*nfpm_inject_config_rules(),
136	inject_nfpm_package_fields_plugin,
137	inject_nfpm_package_fields_subclass,
138	inject_nfpm_package_fields_high_priority,
139	UnionRule(InjectNfpmPackageFieldsRequest, PluginInjectFieldsRequest),
140	UnionRule(InjectNfpmPackageFieldsRequest, SubclassPluginInjectFieldsRequest),
141	UnionRule(InjectNfpmPackageFieldsRequest, HighPriorityInjectFieldsRequest),
142	QueryRule(InjectedNfpmPackageFields, (NfpmPackageTargetWrapper,)),
143	],
144	)
145	rule_runner.set_options([], env_inherit={"PATH", "PYENV_ROOT", "HOME"})	1✔
146	return rule_runner	1✔
147
148
149	@pytest.mark.parametrize(	1✔
150	"packager",
151	(
152	"apk",
153	"archlinux",
154	"deb",
155	"rpm",
156	),
157	)
158	def test_determine_injected_nfpm_package_fields(rule_runner: RuleRunner, packager: str) -> None:	1✔
159	packager = "deb"	1✔
160	rule_runner.write_files(	1✔
161	{
162	"BUILD": dedent(
163	f"""
164	nfpm_{packager}_package(
165	name="{_PKG_NAME}",
166	description="A {packager} package",
167	package_name="{_PKG_NAME}",
168	version="", # the plugin should provide this
169	{"" if packager != "deb" else 'maintainer="Foo Bar <deb@example.com>",'}
170	dependencies=[],
171	)
172	"""
173	),
174	}
175	)
176	target = rule_runner.get_target(Address("", target_name=_PKG_NAME))	1✔
177	result = rule_runner.request(InjectedNfpmPackageFields, [NfpmPackageTargetWrapper(target)])	1✔
178	field_values = result.field_values	1✔
179	assert len(field_values) == 3	1✔
180	assert field_values[NfpmVersionField].value == "9.8.7-dev+git" # (Plugin*)	1✔
181	assert field_values[NfpmVersionReleaseField].value == 16 # 6 (Plugin) + 10 (SubclassPlugin)	1✔
182	assert field_values[NfpmPlatformField].value == "foobar" # (HighPriority*)	1✔

pantsbuild / pants / 19186107394

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