19187435615

Committed 08 Nov 2025 03:38AM UTC coverage: 80.298% (-0.002%) from 80.3%

Build # 19187435615

Build Type

Pull #22862

github

Committed by

web-flow

Commit Message

Merge a76a55651 into a20fa8db6

Pull Request Pull Request #22862: [pants-ng] Initial implementation of contextual config

Run Details

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
41	fields: list[Field] = list(request.injected_fields.values())	×
42	if NfpmVersionField not in request.injected_fields:	×
43	fields.extend(	×
44	[
45	NfpmVersionField("9.8.7-dev+git", address),
46	NfpmVersionReleaseField(6, address),
47	]
48	)
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:
64	address = request.target.address	×
65	# preserve fields from earlier rules in chain
66	fields: list[Field] = list(request.injected_fields.values())	×
67	if not fields or NfpmVersionReleaseField in request.injected_fields:	×
68	release = 0	×
69	if NfpmVersionReleaseField in request.injected_fields:	×
70	old_release = request.injected_fields[NfpmVersionReleaseField].value	×
71	assert old_release is not None	×
72	release = 10 + old_release	×
73	fields.append(NfpmVersionReleaseField(release, address))	×
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✔
82	return True	×
83
84
85	@rule	1✔
86	async def inject_nfpm_package_fields_high_priority(	1✔
87	request: HighPriorityInjectFieldsRequest,
88	) -> InjectedNfpmPackageFields:
89	address = request.target.address	×
90	# preserve fields from earlier rules in chain
91	fields: list[Field] = list(request.injected_fields.values())	×
92	if not fields or NfpmVersionField not in request.injected_fields:	×
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.
101	fields.append(NfpmPlatformField("foobar", address))	×
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 / 19187435615

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