18812500213

Committed 26 Oct 2025 03:42AM UTC coverage: 80.284% (+0.005%) from 80.279%

Build # 18812500213

Build Type

Pull #22804

github

Committed by

web-flow

Commit Message

Merge 2a56fdb46 into 4834308dc

Pull Request Pull Request #22804: test_shell_command: use correct default cache scope for a test's environment

Run Details

29 of 31 new or added lines in 2 files covered. (93.55%)

1314 existing lines in 64 files now uncovered.

77900 of 97030 relevant lines covered (80.28%)

3.35 hits per line

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

83.45

/src/python/pants/backend/nfpm/field_sets.py

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

from __future__ import annotations

from abc import ABCMeta, abstractmethod
from dataclasses import dataclass
from typing import Any, ClassVar

from pants.backend.nfpm.config import NfpmContent, NfpmFileInfo, OctalInt
from pants.backend.nfpm.fields.all import (
    NfpmOutputPathField,
    NfpmPackageMtimeField,
    NfpmPackageNameField,
)
from pants.backend.nfpm.fields.contents import (
    NfpmContentDirDstField,
    NfpmContentDstField,
    NfpmContentFileGroupField,
    NfpmContentFileModeField,
    NfpmContentFileMtimeField,
    NfpmContentFileOwnerField,
    NfpmContentFileSourceField,
    NfpmContentSrcField,
    NfpmContentSymlinkDstField,
    NfpmContentSymlinkSrcField,
    NfpmContentTypeField,
)
from pants.backend.nfpm.fields.rpm import NfpmRpmGhostContents
from pants.backend.nfpm.fields.scripts import NfpmPackageScriptsField
from pants.backend.nfpm.target_types import APK_FIELDS, ARCHLINUX_FIELDS, DEB_FIELDS, RPM_FIELDS
from pants.core.goals.package import PackageFieldSet
from pants.engine.fs import FileEntry
from pants.engine.internals.native_engine import Field
from pants.engine.rules import collect_rules
from pants.engine.target import DescriptionField, FieldSet, Target
from pants.engine.unions import UnionRule, union
from pants.util.frozendict import FrozenDict
from pants.util.ordered_set import FrozenOrderedSet


@dataclass(frozen=True)
class NfpmPackageFieldSet(PackageFieldSet, metaclass=ABCMeta):
    packager: ClassVar[str]
    extension: ClassVar[str]
    output_path: NfpmOutputPathField
    package_name: NfpmPackageNameField
    mtime: NfpmPackageMtimeField
    description: DescriptionField
    scripts: NfpmPackageScriptsField

    def nfpm_config(
        self,
        tgt: Target,
        injected_fields: FrozenDict[type[Field], Field],
        *,
        default_mtime: str | None,
    ) -> dict[str, Any]:
        config: dict[str, Any] = {
            # pants handles any globbing before passing contents to nFPM.
            "disable_globbing": True,
            "contents": [],
            "mtime": self.mtime.normalized_value(default_mtime),
        }

        def fill_nested(_nfpm_alias: str, value: Any) -> None:
            # handle nested fields (eg: deb.triggers, rpm.compression, maintainer)
            keys = _nfpm_alias.split(".")

            cfg = config
            for key in keys[:-1]:
                # NB: if key == "[]" then it is an array (.contents).
                # We can safely ignore .contents because contents fields are on
                # the nfpm content targets, not on nfpm package targets, so
                # they will not be in NfpmPackageFieldSet.required_fields.
                # "contents" gets added to the config based on the dependencies field.
                cfg.setdefault(key, {})
                cfg = cfg[key]
            if isinstance(value, FrozenDict):
                value = dict(value)
            cfg[keys[-1]] = value

        for field in self.required_fields:
            # NB: This assumes that nfpm fields have a str 'nfpm_alias' attribute.
            if not hasattr(field, "nfpm_alias"):
                # Ignore field that is not defined in the nfpm backend.
                continue
            # nfpm_alias is a "." concatenated series of nfpm.yaml dict keys.
            nfpm_alias: str = getattr(field, "nfpm_alias", "")
            if not nfpm_alias:
                # field opted out of being included in this config (like dependencies)
                continue

            field_value = injected_fields.get(field, tgt[field]).value
            # NB: This assumes that nfpm fields have 'none_is_valid_value=False'.
            if not field.required and field_value is None:
                # Omit any undefined optional values unless default applied.
                # A default ensures field_value will not be None. So, the pants interface
                # will be stable even if nFPM changes any defaults.
                continue

            fill_nested(nfpm_alias, field_value)

        for script_type, script_src in self.scripts.normalized_value.items():
            nfpm_alias = self.scripts.nfpm_aliases[script_type]
            fill_nested(nfpm_alias, script_src)

        description = self.description.value
        if description:
            config["description"] = description

        return config


@dataclass(frozen=True)
class NfpmApkPackageFieldSet(NfpmPackageFieldSet):
    packager = "apk"
    extension = f".{packager}"
    required_fields = APK_FIELDS


# noinspection DuplicatedCode
@dataclass(frozen=True)
class NfpmArchlinuxPackageFieldSet(NfpmPackageFieldSet):
    packager = "archlinux"
    # NB: uses *.tar.zst (unlike the other packagers where packager == file extension)
    extension = ".tar.zst"
    required_fields = ARCHLINUX_FIELDS


# noinspection DuplicatedCode
@dataclass(frozen=True)
class NfpmDebPackageFieldSet(NfpmPackageFieldSet):
    packager = "deb"
    extension = f".{packager}"
    required_fields = DEB_FIELDS


# noinspection DuplicatedCode
@dataclass(frozen=True)
class NfpmRpmPackageFieldSet(NfpmPackageFieldSet):
    packager = "rpm"
    extension = f".{packager}"
    required_fields = RPM_FIELDS
    ghost_contents: NfpmRpmGhostContents

    def nfpm_config(
        self,
        tgt: Target,
        injected_fields: FrozenDict[type[Field], Field],
        *,
        default_mtime: str | None,
    ) -> dict[str, Any]:
        config = super().nfpm_config(tgt, injected_fields, default_mtime=default_mtime)
        config["contents"].extend(self.ghost_contents.nfpm_contents)
        return config


NFPM_PACKAGE_FIELD_SET_TYPES: FrozenOrderedSet[type[NfpmPackageFieldSet]] = FrozenOrderedSet(
    (
        NfpmApkPackageFieldSet,
        NfpmArchlinuxPackageFieldSet,
        NfpmDebPackageFieldSet,
        NfpmRpmPackageFieldSet,
    )
)


@union
@dataclass(frozen=True)
class NfpmContentFieldSet(FieldSet, metaclass=ABCMeta):
    owner: NfpmContentFileOwnerField
    group: NfpmContentFileGroupField
    mode: NfpmContentFileModeField
    mtime: NfpmContentFileMtimeField

    @abstractmethod
    def nfpm_config(
        self, *, content_sandbox_files: dict[str, FileEntry], default_mtime: str | None = None
    ) -> NfpmContent:
        pass

    def file_info(
        self, default_is_executable: bool | None = None, default_mtime: str | None = None
    ) -> NfpmFileInfo:
        mode = self.mode.value
        if mode is None and default_is_executable is not None:
            # NB: The execute bit is the only mode bit we can safely get from the sandbox.
            # If we don't pass an explicit mode, nFPM will try to use the sandboxed file's mode.
            mode = 0o755 if default_is_executable else 0o644

        return NfpmFileInfo(
            owner=self.owner.value,
            group=self.group.value,
            mode=OctalInt(mode) if mode is not None else mode,
            mtime=self.mtime.normalized_value(default_mtime),
        )


@dataclass(frozen=True)
class NfpmContentDirFieldSet(NfpmContentFieldSet):
    required_fields = (NfpmContentDirDstField,)

    dst: NfpmContentDirDstField

    def nfpm_config(
        self, *, content_sandbox_files: dict[str, FileEntry], default_mtime: str | None = None
    ) -> NfpmContent:
        return NfpmContent(
            type="dir",
            dst=self.dst.value,
            file_info=self.file_info(default_mtime=default_mtime),
        )


@dataclass(frozen=True)
class NfpmContentSymlinkFieldSet(NfpmContentFieldSet):
    required_fields = (NfpmContentSymlinkDstField,)

    src: NfpmContentSymlinkSrcField
    dst: NfpmContentSymlinkDstField

    def nfpm_config(
        self, *, content_sandbox_files: dict[str, FileEntry], default_mtime: str | None = None
    ) -> NfpmContent:
        return NfpmContent(
            type="symlink",
            src=self.src.value,
            dst=self.dst.value,
            file_info=self.file_info(default_mtime=default_mtime),
        )


@dataclass(frozen=True)
class NfpmContentFileFieldSet(NfpmContentFieldSet):
    required_fields = (NfpmContentDstField,)

    source: NfpmContentFileSourceField
    src: NfpmContentSrcField
    dst: NfpmContentDstField
    content_type: NfpmContentTypeField

    class InvalidTarget(Exception):
        pass

    class SrcMissingFomSandbox(Exception):
        pass

    def nfpm_config(
        self, *, content_sandbox_files: dict[str, FileEntry], default_mtime: str | None = None
    ) -> NfpmContent:
        source: str | None = self.source.file_path
        src: str | None = self.src.file_path
        src_value: str | None = self.src.value
        dst: str = self.dst.value
        if source is not None and not src:
            # If defined, 'source' provides the default value for 'src'.
            src = source
            src_value = self.source.value
        if src is None:  # src is NOT required; prepare to raise an error.
            raise self.InvalidTarget()
        if src not in content_sandbox_files and src_value in content_sandbox_files:
            # A field's file_path assumes the field's value is relative to the BUILD file.
            # But, for packages the field's value can be relative to the build_root instead,
            # because a package's output_path can be any arbitrary build_root relative path.
            src = src_value
        sandbox_file: FileEntry | None = content_sandbox_files.get(src)
        if sandbox_file is None:
            raise self.SrcMissingFomSandbox()
        return NfpmContent(
            type=self.content_type.value,
            src=src,
            dst=dst,
            file_info=self.file_info(sandbox_file.is_executable, default_mtime),
        )


NFPM_CONTENT_FIELD_SET_TYPES: FrozenOrderedSet[type[NfpmContentFieldSet]] = FrozenOrderedSet(
    (
        NfpmContentDirFieldSet,
        NfpmContentSymlinkFieldSet,
        NfpmContentFileFieldSet,
    )
)


def rules():
    return [
        *collect_rules(),
        *(
            UnionRule(PackageFieldSet, field_set_type)
            for field_set_type in NFPM_PACKAGE_FIELD_SET_TYPES
        ),
        *(
            UnionRule(NfpmContentFieldSet, field_set_type)
            for field_set_type in NFPM_CONTENT_FIELD_SET_TYPES
        ),
    ]

1	# Copyright 2023 Pants project contributors (see CONTRIBUTORS.md).
2	# Licensed under the Apache License, Version 2.0 (see LICENSE).
3
4	from __future__ import annotations	4✔
5
6	from abc import ABCMeta, abstractmethod	4✔
7	from dataclasses import dataclass	4✔
8	from typing import Any, ClassVar	4✔
9
10	from pants.backend.nfpm.config import NfpmContent, NfpmFileInfo, OctalInt	4✔
11	from pants.backend.nfpm.fields.all import (	4✔
12	NfpmOutputPathField,
13	NfpmPackageMtimeField,
14	NfpmPackageNameField,
15	)
16	from pants.backend.nfpm.fields.contents import (	4✔
17	NfpmContentDirDstField,
18	NfpmContentDstField,
19	NfpmContentFileGroupField,
20	NfpmContentFileModeField,
21	NfpmContentFileMtimeField,
22	NfpmContentFileOwnerField,
23	NfpmContentFileSourceField,
24	NfpmContentSrcField,
25	NfpmContentSymlinkDstField,
26	NfpmContentSymlinkSrcField,
27	NfpmContentTypeField,
28	)
29	from pants.backend.nfpm.fields.rpm import NfpmRpmGhostContents	4✔
30	from pants.backend.nfpm.fields.scripts import NfpmPackageScriptsField	4✔
31	from pants.backend.nfpm.target_types import APK_FIELDS, ARCHLINUX_FIELDS, DEB_FIELDS, RPM_FIELDS	4✔
32	from pants.core.goals.package import PackageFieldSet	4✔
33	from pants.engine.fs import FileEntry	4✔
34	from pants.engine.internals.native_engine import Field	4✔
35	from pants.engine.rules import collect_rules	4✔
36	from pants.engine.target import DescriptionField, FieldSet, Target	4✔
37	from pants.engine.unions import UnionRule, union	4✔
38	from pants.util.frozendict import FrozenDict	4✔
39	from pants.util.ordered_set import FrozenOrderedSet	4✔
40
41
42	@dataclass(frozen=True)	4✔
43	class NfpmPackageFieldSet(PackageFieldSet, metaclass=ABCMeta):	4✔
44	packager: ClassVar[str]	4✔
45	extension: ClassVar[str]	4✔
46	output_path: NfpmOutputPathField	4✔
47	package_name: NfpmPackageNameField	4✔
48	mtime: NfpmPackageMtimeField	4✔
49	description: DescriptionField	4✔
50	scripts: NfpmPackageScriptsField	4✔
51
52	def nfpm_config(	4✔
53	self,
54	tgt: Target,
55	injected_fields: FrozenDict[type[Field], Field],
56	*,
57	default_mtime: str \| None,
58	) -> dict[str, Any]:
UNCOV 59	config: dict[str, Any] = {	1✔
60	# pants handles any globbing before passing contents to nFPM.
61	"disable_globbing": True,
62	"contents": [],
63	"mtime": self.mtime.normalized_value(default_mtime),
64	}
65
UNCOV 66	def fill_nested(_nfpm_alias: str, value: Any) -> None:	1✔
67	# handle nested fields (eg: deb.triggers, rpm.compression, maintainer)
UNCOV 68	keys = _nfpm_alias.split(".")	1✔
69
UNCOV 70	cfg = config	1✔
UNCOV 71	for key in keys[:-1]:	1✔
72	# NB: if key == "[]" then it is an array (.contents).
73	# We can safely ignore .contents because contents fields are on
74	# the nfpm content targets, not on nfpm package targets, so
75	# they will not be in NfpmPackageFieldSet.required_fields.
76	# "contents" gets added to the config based on the dependencies field.
UNCOV 77	cfg.setdefault(key, {})	1✔
UNCOV 78	cfg = cfg[key]	1✔
UNCOV 79	if isinstance(value, FrozenDict):	1✔
UNCOV 80	value = dict(value)	1✔
UNCOV 81	cfg[keys[-1]] = value	1✔
82
UNCOV 83	for field in self.required_fields:	1✔
84	# NB: This assumes that nfpm fields have a str 'nfpm_alias' attribute.
UNCOV 85	if not hasattr(field, "nfpm_alias"):	1✔
86	# Ignore field that is not defined in the nfpm backend.
87	continue	×
88	# nfpm_alias is a "." concatenated series of nfpm.yaml dict keys.
UNCOV 89	nfpm_alias: str = getattr(field, "nfpm_alias", "")	1✔
UNCOV 90	if not nfpm_alias:	1✔
91	# field opted out of being included in this config (like dependencies)
UNCOV 92	continue	1✔
93
UNCOV 94	field_value = injected_fields.get(field, tgt[field]).value	1✔
95	# NB: This assumes that nfpm fields have 'none_is_valid_value=False'.
UNCOV 96	if not field.required and field_value is None:	1✔
97	# Omit any undefined optional values unless default applied.
98	# A default ensures field_value will not be None. So, the pants interface
99	# will be stable even if nFPM changes any defaults.
UNCOV 100	continue	1✔
101
UNCOV 102	fill_nested(nfpm_alias, field_value)	1✔
103
UNCOV 104	for script_type, script_src in self.scripts.normalized_value.items():	1✔
UNCOV 105	nfpm_alias = self.scripts.nfpm_aliases[script_type]	1✔
UNCOV 106	fill_nested(nfpm_alias, script_src)	1✔
107
UNCOV 108	description = self.description.value	1✔
UNCOV 109	if description:	1✔
UNCOV 110	config["description"] = description	1✔
111
UNCOV 112	return config	1✔
113
114
115	@dataclass(frozen=True)	4✔
116	class NfpmApkPackageFieldSet(NfpmPackageFieldSet):	4✔
117	packager = "apk"	4✔
118	extension = f".{packager}"	4✔
119	required_fields = APK_FIELDS	4✔
120
121
122	# noinspection DuplicatedCode
123	@dataclass(frozen=True)	4✔
124	class NfpmArchlinuxPackageFieldSet(NfpmPackageFieldSet):	4✔
125	packager = "archlinux"	4✔
126	# NB: uses *.tar.zst (unlike the other packagers where packager == file extension)
127	extension = ".tar.zst"	4✔
128	required_fields = ARCHLINUX_FIELDS	4✔
129
130
131	# noinspection DuplicatedCode
132	@dataclass(frozen=True)	4✔
133	class NfpmDebPackageFieldSet(NfpmPackageFieldSet):	4✔
134	packager = "deb"	4✔
135	extension = f".{packager}"	4✔
136	required_fields = DEB_FIELDS	4✔
137
138
139	# noinspection DuplicatedCode
140	@dataclass(frozen=True)	4✔
141	class NfpmRpmPackageFieldSet(NfpmPackageFieldSet):	4✔
142	packager = "rpm"	4✔
143	extension = f".{packager}"	4✔
144	required_fields = RPM_FIELDS	4✔
145	ghost_contents: NfpmRpmGhostContents	4✔
146
147	def nfpm_config(	4✔
148	self,
149	tgt: Target,
150	injected_fields: FrozenDict[type[Field], Field],
151	*,
152	default_mtime: str \| None,
153	) -> dict[str, Any]:
UNCOV 154	config = super().nfpm_config(tgt, injected_fields, default_mtime=default_mtime)	1✔
UNCOV 155	config["contents"].extend(self.ghost_contents.nfpm_contents)	1✔
UNCOV 156	return config	1✔
157
158
159	NFPM_PACKAGE_FIELD_SET_TYPES: FrozenOrderedSet[type[NfpmPackageFieldSet]] = FrozenOrderedSet(	4✔
160	(
161	NfpmApkPackageFieldSet,
162	NfpmArchlinuxPackageFieldSet,
163	NfpmDebPackageFieldSet,
164	NfpmRpmPackageFieldSet,
165	)
166	)
167
168
169	@union	4✔
170	@dataclass(frozen=True)	4✔
171	class NfpmContentFieldSet(FieldSet, metaclass=ABCMeta):	4✔
172	owner: NfpmContentFileOwnerField	4✔
173	group: NfpmContentFileGroupField	4✔
174	mode: NfpmContentFileModeField	4✔
175	mtime: NfpmContentFileMtimeField	4✔
176
177	@abstractmethod	4✔
178	def nfpm_config(	4✔
179	self, *, content_sandbox_files: dict[str, FileEntry], default_mtime: str \| None = None
180	) -> NfpmContent:
181	pass	×
182
183	def file_info(	4✔
184	self, default_is_executable: bool \| None = None, default_mtime: str \| None = None
185	) -> NfpmFileInfo:
186	mode = self.mode.value	×
187	if mode is None and default_is_executable is not None:	×
188	# NB: The execute bit is the only mode bit we can safely get from the sandbox.
189	# If we don't pass an explicit mode, nFPM will try to use the sandboxed file's mode.
190	mode = 0o755 if default_is_executable else 0o644	×
191
192	return NfpmFileInfo(	×
193	owner=self.owner.value,
194	group=self.group.value,
195	mode=OctalInt(mode) if mode is not None else mode,
196	mtime=self.mtime.normalized_value(default_mtime),
197	)
198
199
200	@dataclass(frozen=True)	4✔
201	class NfpmContentDirFieldSet(NfpmContentFieldSet):	4✔
202	required_fields = (NfpmContentDirDstField,)	4✔
203
204	dst: NfpmContentDirDstField	4✔
205
206	def nfpm_config(	4✔
207	self, *, content_sandbox_files: dict[str, FileEntry], default_mtime: str \| None = None
208	) -> NfpmContent:
209	return NfpmContent(	×
210	type="dir",
211	dst=self.dst.value,
212	file_info=self.file_info(default_mtime=default_mtime),
213	)
214
215
216	@dataclass(frozen=True)	4✔
217	class NfpmContentSymlinkFieldSet(NfpmContentFieldSet):	4✔
218	required_fields = (NfpmContentSymlinkDstField,)	4✔
219
220	src: NfpmContentSymlinkSrcField	4✔
221	dst: NfpmContentSymlinkDstField	4✔
222
223	def nfpm_config(	4✔
224	self, *, content_sandbox_files: dict[str, FileEntry], default_mtime: str \| None = None
225	) -> NfpmContent:
226	return NfpmContent(	×
227	type="symlink",
228	src=self.src.value,
229	dst=self.dst.value,
230	file_info=self.file_info(default_mtime=default_mtime),
231	)
232
233
234	@dataclass(frozen=True)	4✔
235	class NfpmContentFileFieldSet(NfpmContentFieldSet):	4✔
236	required_fields = (NfpmContentDstField,)	4✔
237
238	source: NfpmContentFileSourceField	4✔
239	src: NfpmContentSrcField	4✔
240	dst: NfpmContentDstField	4✔
241	content_type: NfpmContentTypeField	4✔
242
243	class InvalidTarget(Exception):	4✔
244	pass	4✔
245
246	class SrcMissingFomSandbox(Exception):	4✔
247	pass	4✔
248
249	def nfpm_config(	4✔
250	self, *, content_sandbox_files: dict[str, FileEntry], default_mtime: str \| None = None
251	) -> NfpmContent:
252	source: str \| None = self.source.file_path	×
253	src: str \| None = self.src.file_path	×
254	src_value: str \| None = self.src.value	×
255	dst: str = self.dst.value	×
256	if source is not None and not src:	×
257	# If defined, 'source' provides the default value for 'src'.
258	src = source	×
259	src_value = self.source.value	×
260	if src is None: # src is NOT required; prepare to raise an error.	×
261	raise self.InvalidTarget()	×
262	if src not in content_sandbox_files and src_value in content_sandbox_files:	×
263	# A field's file_path assumes the field's value is relative to the BUILD file.
264	# But, for packages the field's value can be relative to the build_root instead,
265	# because a package's output_path can be any arbitrary build_root relative path.
266	src = src_value	×
267	sandbox_file: FileEntry \| None = content_sandbox_files.get(src)	×
268	if sandbox_file is None:	×
269	raise self.SrcMissingFomSandbox()	×
270	return NfpmContent(	×
271	type=self.content_type.value,
272	src=src,
273	dst=dst,
274	file_info=self.file_info(sandbox_file.is_executable, default_mtime),
275	)
276
277
278	NFPM_CONTENT_FIELD_SET_TYPES: FrozenOrderedSet[type[NfpmContentFieldSet]] = FrozenOrderedSet(	4✔
279	(
280	NfpmContentDirFieldSet,
281	NfpmContentSymlinkFieldSet,
282	NfpmContentFileFieldSet,
283	)
284	)
285
286
287	def rules():	4✔
288	return [	3✔
289	*collect_rules(),
290	*(
291	UnionRule(PackageFieldSet, field_set_type)
292	for field_set_type in NFPM_PACKAGE_FIELD_SET_TYPES
293	),
294	*(
295	UnionRule(NfpmContentFieldSet, field_set_type)
296	for field_set_type in NFPM_CONTENT_FIELD_SET_TYPES
297	),
298	]

pantsbuild / pants / 18812500213

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