18252174847

Committed 05 Oct 2025 01:36AM UTC coverage: 43.382% (-36.9%) from 80.261%

Build # 18252174847

Build Type

push

github

Committed by

web-flow

Commit Message

run tests on mac arm (#22717)

Just doing the minimal to pull forward the x86_64 pattern.

ref #20993

Run Details

25776 of 59416 relevant lines covered (43.38%)

1.3 hits per line

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

62.59

/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	3✔
5
6	from abc import ABCMeta, abstractmethod	3✔
7	from dataclasses import dataclass	3✔
8	from typing import Any, ClassVar	3✔
9
10	from pants.backend.nfpm.config import NfpmContent, NfpmFileInfo, OctalInt	3✔
11	from pants.backend.nfpm.fields.all import (	3✔
12	NfpmOutputPathField,
13	NfpmPackageMtimeField,
14	NfpmPackageNameField,
15	)
16	from pants.backend.nfpm.fields.contents import (	3✔
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	3✔
30	from pants.backend.nfpm.fields.scripts import NfpmPackageScriptsField	3✔
31	from pants.backend.nfpm.target_types import APK_FIELDS, ARCHLINUX_FIELDS, DEB_FIELDS, RPM_FIELDS	3✔
32	from pants.core.goals.package import PackageFieldSet	3✔
33	from pants.engine.fs import FileEntry	3✔
34	from pants.engine.internals.native_engine import Field	3✔
35	from pants.engine.rules import collect_rules	3✔
36	from pants.engine.target import DescriptionField, FieldSet, Target	3✔
37	from pants.engine.unions import UnionRule, union	3✔
38	from pants.util.frozendict import FrozenDict	3✔
39	from pants.util.ordered_set import FrozenOrderedSet	3✔
40
41
42	@dataclass(frozen=True)	3✔
43	class NfpmPackageFieldSet(PackageFieldSet, metaclass=ABCMeta):	3✔
44	packager: ClassVar[str]	3✔
45	extension: ClassVar[str]	3✔
46	output_path: NfpmOutputPathField	3✔
47	package_name: NfpmPackageNameField	3✔
48	mtime: NfpmPackageMtimeField	3✔
49	description: DescriptionField	3✔
50	scripts: NfpmPackageScriptsField	3✔
51
52	def nfpm_config(	3✔
53	self,
54	tgt: Target,
55	injected_fields: FrozenDict[type[Field], Field],
56	*,
57	default_mtime: str \| None,
58	) -> dict[str, Any]:
59	config: dict[str, Any] = {	×
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
66	def fill_nested(_nfpm_alias: str, value: Any) -> None:	×
67	# handle nested fields (eg: deb.triggers, rpm.compression, maintainer)
68	keys = _nfpm_alias.split(".")	×
69
70	cfg = config	×
71	for key in keys[:-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.
77	cfg.setdefault(key, {})	×
78	cfg = cfg[key]	×
79	if isinstance(value, FrozenDict):	×
80	value = dict(value)	×
81	cfg[keys[-1]] = value	×
82
83	for field in self.required_fields:	×
84	# NB: This assumes that nfpm fields have a str 'nfpm_alias' attribute.
85	if not hasattr(field, "nfpm_alias"):	×
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.
89	nfpm_alias: str = getattr(field, "nfpm_alias", "")	×
90	if not nfpm_alias:	×
91	# field opted out of being included in this config (like dependencies)
92	continue	×
93
94	field_value = injected_fields.get(field, tgt[field]).value	×
95	# NB: This assumes that nfpm fields have 'none_is_valid_value=False'.
96	if not field.required and field_value is None:	×
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.
100	continue	×
101
102	fill_nested(nfpm_alias, field_value)	×
103
104	for script_type, script_src in self.scripts.normalized_value.items():	×
105	nfpm_alias = self.scripts.nfpm_aliases[script_type]	×
106	fill_nested(nfpm_alias, script_src)	×
107
108	description = self.description.value	×
109	if description:	×
110	config["description"] = description	×
111
112	return config	×
113
114
115	@dataclass(frozen=True)	3✔
116	class NfpmApkPackageFieldSet(NfpmPackageFieldSet):	3✔
117	packager = "apk"	3✔
118	extension = f".{packager}"	3✔
119	required_fields = APK_FIELDS	3✔
120
121
122	# noinspection DuplicatedCode
123	@dataclass(frozen=True)	3✔
124	class NfpmArchlinuxPackageFieldSet(NfpmPackageFieldSet):	3✔
125	packager = "archlinux"	3✔
126	# NB: uses *.tar.zst (unlike the other packagers where packager == file extension)
127	extension = ".tar.zst"	3✔
128	required_fields = ARCHLINUX_FIELDS	3✔
129
130
131	# noinspection DuplicatedCode
132	@dataclass(frozen=True)	3✔
133	class NfpmDebPackageFieldSet(NfpmPackageFieldSet):	3✔
134	packager = "deb"	3✔
135	extension = f".{packager}"	3✔
136	required_fields = DEB_FIELDS	3✔
137
138
139	# noinspection DuplicatedCode
140	@dataclass(frozen=True)	3✔
141	class NfpmRpmPackageFieldSet(NfpmPackageFieldSet):	3✔
142	packager = "rpm"	3✔
143	extension = f".{packager}"	3✔
144	required_fields = RPM_FIELDS	3✔
145	ghost_contents: NfpmRpmGhostContents	3✔
146
147	def nfpm_config(	3✔
148	self,
149	tgt: Target,
150	injected_fields: FrozenDict[type[Field], Field],
151	*,
152	default_mtime: str \| None,
153	) -> dict[str, Any]:
154	config = super().nfpm_config(tgt, injected_fields, default_mtime=default_mtime)	×
155	config["contents"].extend(self.ghost_contents.nfpm_contents)	×
156	return config	×
157
158
159	NFPM_PACKAGE_FIELD_SET_TYPES: FrozenOrderedSet[type[NfpmPackageFieldSet]] = FrozenOrderedSet(	3✔
160	(
161	NfpmApkPackageFieldSet,
162	NfpmArchlinuxPackageFieldSet,
163	NfpmDebPackageFieldSet,
164	NfpmRpmPackageFieldSet,
165	)
166	)
167
168
169	@union	3✔
170	@dataclass(frozen=True)	3✔
171	class NfpmContentFieldSet(FieldSet, metaclass=ABCMeta):	3✔
172	owner: NfpmContentFileOwnerField	3✔
173	group: NfpmContentFileGroupField	3✔
174	mode: NfpmContentFileModeField	3✔
175	mtime: NfpmContentFileMtimeField	3✔
176
177	@abstractmethod	3✔
178	def nfpm_config(	3✔
179	self, *, content_sandbox_files: dict[str, FileEntry], default_mtime: str \| None = None
180	) -> NfpmContent:
181	pass	×
182
183	def file_info(	3✔
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)	3✔
201	class NfpmContentDirFieldSet(NfpmContentFieldSet):	3✔
202	required_fields = (NfpmContentDirDstField,)	3✔
203
204	dst: NfpmContentDirDstField	3✔
205
206	def nfpm_config(	3✔
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)	3✔
217	class NfpmContentSymlinkFieldSet(NfpmContentFieldSet):	3✔
218	required_fields = (NfpmContentSymlinkDstField,)	3✔
219
220	src: NfpmContentSymlinkSrcField	3✔
221	dst: NfpmContentSymlinkDstField	3✔
222
223	def nfpm_config(	3✔
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)	3✔
235	class NfpmContentFileFieldSet(NfpmContentFieldSet):	3✔
236	required_fields = (NfpmContentDstField,)	3✔
237
238	source: NfpmContentFileSourceField	3✔
239	src: NfpmContentSrcField	3✔
240	dst: NfpmContentDstField	3✔
241	content_type: NfpmContentTypeField	3✔
242
243	class InvalidTarget(Exception):	3✔
244	pass	3✔
245
246	class SrcMissingFomSandbox(Exception):	3✔
247	pass	3✔
248
249	def nfpm_config(	3✔
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(	3✔
279	(
280	NfpmContentDirFieldSet,
281	NfpmContentSymlinkFieldSet,
282	NfpmContentFileFieldSet,
283	)
284	)
285
286
287	def rules():	3✔
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 / 18252174847

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