20147226056

Committed 11 Dec 2025 08:58PM UTC coverage: 78.827% (-1.5%) from 80.293%

Build # 20147226056

Build Type

push

github

Committed by

web-flow

Commit Message

Forwarded the `style` and `complete-platform` args from pants.toml to PEX (#22910)

## Context

After Apple switched to the `arm64` architecture, some package
publishers stopped releasing `x86_64` variants of their packages for
`darwin`. As a result, generating a universal lockfile now fails because
no single package version is compatible with both `x86_64` and `arm64`
on `darwin`.

The solution is to use the `--style` and `--complete-platform` flags
with PEX. For example:
```
pex3 lock create \
    --style strict \
    --complete-platform 3rdparty/platforms/manylinux_2_28_aarch64.json \
    --complete-platform 3rdparty/platforms/macosx_26_0_arm64.json \
    -r 3rdparty/python/requirements_pyarrow.txt \
    -o python-pyarrow.lock
```

See the Slack discussion here:
https://pantsbuild.slack.com/archives/C046T6T9U/p1760098582461759

## Reproduction

* `BUILD`
```
python_requirement(
    name="awswrangler",
    requirements=["awswrangler==3.12.1"],
    resolve="awswrangler",
)
```
* Run `pants generate-lockfiles --resolve=awswrangler` on macOS with an
`arm64` CPU
```
pip: ERROR: Cannot install awswrangler==3.12.1 because these package versions have conflicting dependencies.
pip: ERROR: ResolutionImpossible: for help visit https://pip.pypa.io/en/latest/topics/dependency-resolution/#dealing-with-dependency-conflicts
pip:  
pip:  The conflict is caused by:
pip:      awswrangler 3.12.1 depends on pyarrow<18.0.0 and >=8.0.0; sys_platform == "darwin" and platform_machine == "x86_64"
pip:      awswrangler 3.12.1 depends on pyarrow<21.0.0 and >=18.0.0; sys_platform != "darwin" or platform_machine != "x86_64"
pip:  
pip:  Additionally, some packages in these conflicts have no matching distributions available for your environment:
pip:      pyarrow
pip:  
pip:  To fix this you could try to:
pip:  1. loosen the range of package versions you've specified
pip:  2. remove package versions to allow pip to attempt to solve the dependency conflict
```

## Implementation
... (continued)

Run Details

77 of 100 new or added lines in 6 files covered. (77.0%)

868 existing lines in 42 files now uncovered.

74471 of 94474 relevant lines covered (78.83%)

3.18 hits per line

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

87.69

/src/python/pants/core/util_rules/lockfile_metadata.py

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

from __future__ import annotations

import hashlib
import json
from collections.abc import Callable, Iterable
from dataclasses import dataclass
from enum import Enum
from typing import Any, ClassVar, Generic, TypeVar, cast

from pants.util.docutil import bin_name
from pants.util.ordered_set import FrozenOrderedSet
from pants.util.strutil import softwrap

BEGIN_LOCKFILE_HEADER = "--- BEGIN PANTS LOCKFILE METADATA: DO NOT EDIT OR REMOVE ---"
END_LOCKFILE_HEADER = "--- END PANTS LOCKFILE METADATA ---"


class LockfileScope(Enum):
    JVM = "jvm"
    PYTHON = "python"


_LockfileMetadataSubclass = TypeVar("_LockfileMetadataSubclass", bound="LockfileMetadata")
# N.B. the value type is `type[_LockfileMetadataSubclass]`
_concrete_metadata_classes: dict[tuple[LockfileScope, int], type] = {}


# Registrar types (pre-declaring to avoid repetition)
RegisterClassForVersion = Callable[[type["LockfileMetadata"]], type["LockfileMetadata"]]


def lockfile_metadata_registrar(scope: LockfileScope) -> Callable[[int], RegisterClassForVersion]:
    """Decorator factory -- returns a decorator that can be used to register Lockfile metadata
    version subclasses belonging to a specific `LockfileScope`."""

    def _lockfile_metadata_version(
        version: int,
    ) -> RegisterClassForVersion:
        """Decorator to register a Lockfile metadata version subclass with a given version number.

        The class must be a frozen dataclass
        """

        def _dec(cls: type[LockfileMetadata]) -> type[LockfileMetadata]:
            # Only frozen dataclasses may be registered as lockfile metadata:
            cls_dataclass_params = getattr(cls, "__dataclass_params__", None)
            if not cls_dataclass_params or not cls_dataclass_params.frozen:
                raise ValueError(
                    softwrap(
                        """
                        Classes registered with `_lockfile_metadata_version` may only be
                        frozen dataclasses
                        """
                    )
                )
            _concrete_metadata_classes[(scope, version)] = cls
            return cls

        return _dec

    return _lockfile_metadata_version


class InvalidLockfileError(Exception):
    pass


class NoLockfileMetadataBlock(InvalidLockfileError):
    pass


@dataclass(frozen=True)
class LockfileMetadata:
    """Base class for metadata that is attached to a given lockfile.

    This class provides the external API for serializing, deserializing, and validating the
    contents of individual lockfiles. New versions of metadata implement a concrete subclass and
    provide deserialization and validation logic, along with specialist serialization logic.

    To construct an instance of the most recent concrete subclass, call `LockfileMetadata.new()`.
    """

    scope: ClassVar[LockfileScope]

    @classmethod
    def from_lockfile(
        cls: type[_LockfileMetadataSubclass],
        lockfile: bytes,
        lockfile_path: str | None = None,
        resolve_name: str | None = None,
        *,
        delimeter: str,
    ) -> _LockfileMetadataSubclass:
        """Parse and return the metadata from the lockfile's header.

        This shouldn't be called on `LockfileMetadata`, but rather on the "base class" for your
        metadata class. See the existing callers for an example.
        """
        assert cls is not LockfileMetadata, "Call me on a subclass!"
        in_metadata_block = False
        metadata_lines = []
        for line in lockfile.splitlines():
            if line == f"{delimeter} {BEGIN_LOCKFILE_HEADER}".encode():
                in_metadata_block = True
            elif line == f"{delimeter} {END_LOCKFILE_HEADER}".encode():
                break
            elif in_metadata_block:
                metadata_lines.append(line[len(delimeter) + 1 :])

        error_suffix = softwrap(
            f"""
            To resolve this error, you will need to regenerate the lockfile by running
            `{bin_name()} generate-lockfiles
            """
        )
        if resolve_name:
            error_suffix += f" --resolve={resolve_name}"
        error_suffix += "`."

        if lockfile_path is not None and resolve_name is not None:
            lockfile_description = f"the lockfile `{lockfile_path}` for `{resolve_name}`"
        elif lockfile_path is not None:
            lockfile_description = f"the lockfile `{lockfile_path}`"
        elif resolve_name is not None:
            lockfile_description = f"the lockfile for `{resolve_name}`"
        else:
            lockfile_description = "this lockfile"

        if not metadata_lines:
            raise NoLockfileMetadataBlock(
                f"Could not find a Pants metadata block in {lockfile_description}. {error_suffix}"
            )
        try:
            json_dict = json.loads(b"\n".join(metadata_lines))
        except json.decoder.JSONDecodeError:
            raise InvalidLockfileError(
                softwrap(
                    f"""
                    Metadata header in {lockfile_description} is not a valid JSON string and can't
                    be decoded.
                    """
                )
                + error_suffix
            )
        return cls.from_json_dict(json_dict, lockfile_description, error_suffix)

    @classmethod
    def from_json_dict(
        cls: type[_LockfileMetadataSubclass],
        json_dict: dict[Any, Any],
        lockfile_description: str,
        error_suffix: str,
    ) -> _LockfileMetadataSubclass:
        version = json_dict.get("version", 1)
        concrete_class = _concrete_metadata_classes[(cls.scope, version)]

        assert issubclass(concrete_class, cls)
        assert concrete_class.scope == cls.scope, (
            "The class used to call `from_json_dict` has a different scope than what was "
            f"expected given the metadata. Expected '{cls.scope}', got '{concrete_class.scope}'",
        )

        return concrete_class._from_json_dict(json_dict, lockfile_description, error_suffix)

    @classmethod
    def _from_json_dict(
        cls: type[_LockfileMetadataSubclass],
        json_dict: dict[Any, Any],
        lockfile_description: str,
        error_suffix: str,
    ) -> _LockfileMetadataSubclass:
        """Construct a `LockfileMetadata` subclass from the supplied JSON dict.

        *** Not implemented. Subclasses should override. ***


        `lockfile_description` is a detailed, human-readable description of the lockfile, which can
        be read by the user to figure out which lockfile is broken in case of an error.

        `error_suffix` is a string describing how to fix the lockfile.
        """

        raise NotImplementedError(
            "`LockfileMetadata._from_json_dict` should not be directly called."
        )

    def to_json(self, with_description: str | None = None) -> str:
        metadata_dict = self.__render_header_dict()
        if with_description is not None:
            metadata_dict["description"] = with_description
        return json.dumps(metadata_dict, ensure_ascii=True, indent=2)

    def add_header_to_lockfile(
        self, lockfile: bytes, *, delimeter: str, regenerate_command: str
    ) -> bytes:
        metadata_json = self.to_json()
        metadata_as_a_comment = "\n".join(f"{delimeter} {l}" for l in metadata_json.splitlines())

        regenerate_command_bytes = "\n".join(
            [
                f"{delimeter} This lockfile was autogenerated by Pants. To regenerate, run:",
                delimeter,
                f"{delimeter}    {regenerate_command}",
            ]
        ).encode()
        header = "\n".join(
            [
                f"{delimeter} {BEGIN_LOCKFILE_HEADER}",
                metadata_as_a_comment,
                f"{delimeter} {END_LOCKFILE_HEADER}",
            ]
        ).encode("ascii")

        return b"%b\n%b\n%b\n\n%b" % (
            regenerate_command_bytes,
            delimeter.encode(),
            header,
            lockfile,
        )

    def __render_header_dict(self) -> dict[Any, Any]:
        """Produce a dictionary to be serialized into the lockfile header.

        Each class should implement a class method called `additional_header_attrs`, which returns a
        `dict` containing the metadata attributes that should be stored in the lockfile.
        """

        attrs: dict[Any, tuple[Any, type]] = {}  # attr name -> (value, where we first saw it)
        for cls in reversed(self.__class__.__mro__[:-1]):
            new_attrs = cast(LockfileMetadata, cls).additional_header_attrs(self)
            for attr in new_attrs:
                if attr in attrs and attrs[attr][0] != new_attrs[attr]:
                    raise AssertionError(
                        softwrap(
                            f"""
                            Lockfile header attribute `{attr}` was returned by both
                            `{attrs[attr][1]}` and `{cls}`, returning different values. If these
                            classes return the same attribute, they must also return the same
                            value.
                            """
                        )
                    )
                attrs[attr] = new_attrs[attr], cls

        return {key: val[0] for key, val in attrs.items()}

    @classmethod
    def additional_header_attrs(cls, instance: LockfileMetadata) -> dict[Any, Any]:
        return {"version": instance.metadata_version()}

    def metadata_version(self):
        """Returns the version number for this metadata class, or raises an exception.

        To avoid raising an exception, ensure the subclass is decorated with
        `lockfile_metadata_version`
        """
        for (scope, ver), cls in _concrete_metadata_classes.items():
            # Note that we do exact version matches so that authors can subclass earlier versions.
            if type(self) is cls:
                return ver
        raise ValueError("Trying to serialize an unregistered `LockfileMetadata` subclass.")


def calculate_invalidation_digest(requirements: Iterable[str]) -> str:
    """Returns an invalidation digest for the given requirements."""
    m = hashlib.sha256()
    inputs = {
        # `FrozenOrderedSet` deduplicates while keeping ordering, which speeds up the sorting if
        # the input was already sorted.
        "requirements": sorted(FrozenOrderedSet(requirements)),
    }
    m.update(json.dumps(inputs).encode("utf-8"))
    return m.hexdigest()


T = TypeVar("T")


class LockfileMetadataValidation(Generic[T]):
    """Boolean-like value which additionally carries reasons why a validation failed."""

    failure_reasons: set[T]

    def __init__(self, failure_reasons: Iterable[T] = ()):
        self.failure_reasons = set(failure_reasons)

    def __bool__(self):
        return not self.failure_reasons


def _get_metadata(
    metadata: dict[Any, Any],
    lockfile_description: str,
    error_suffix: str,
) -> Callable[[str, type[T], Callable[[Any], T] | None], T]:
    """Returns a function that will get a given key from the `metadata` dict, and optionally do some
    verification and post-processing to return a value of the correct type."""

    def get_metadata(key: str, type_: type[T], coerce: Callable[[Any], T] | None) -> T:
        val: Any
        try:
            val = metadata[key]
        except KeyError:
            raise InvalidLockfileError(
                softwrap(
                    f"""
                    Required key `{key}` is not present in metadata header for
                    {lockfile_description}. {error_suffix}
                    """
                )
            )

        if not coerce:
            if isinstance(val, type_):
                return val

            raise InvalidLockfileError(
                softwrap(
                    f"""
                    Metadata value `{key}` in {lockfile_description} must
                    be a {type(type_).__name__}. {error_suffix}
                    """
                )
            )
        else:
            try:
                return coerce(val)
            except Exception:
                raise InvalidLockfileError(
                    softwrap(
                        f"""
                        Metadata value `{key}` in {lockfile_description} must be able to
                        be converted to a {type(type_).__name__}. {error_suffix}
                        """
                    )
                )

    return get_metadata

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	11✔
5
6	import hashlib	11✔
7	import json	11✔
8	from collections.abc import Callable, Iterable	11✔
9	from dataclasses import dataclass	11✔
10	from enum import Enum	11✔
11	from typing import Any, ClassVar, Generic, TypeVar, cast	11✔
12
13	from pants.util.docutil import bin_name	11✔
14	from pants.util.ordered_set import FrozenOrderedSet	11✔
15	from pants.util.strutil import softwrap	11✔
16
17	BEGIN_LOCKFILE_HEADER = "--- BEGIN PANTS LOCKFILE METADATA: DO NOT EDIT OR REMOVE ---"	11✔
18	END_LOCKFILE_HEADER = "--- END PANTS LOCKFILE METADATA ---"	11✔
19
20
21	class LockfileScope(Enum):	11✔
22	JVM = "jvm"	11✔
23	PYTHON = "python"	11✔
24
25
26	_LockfileMetadataSubclass = TypeVar("_LockfileMetadataSubclass", bound="LockfileMetadata")	11✔
27	# N.B. the value type is `type[_LockfileMetadataSubclass]`
28	_concrete_metadata_classes: dict[tuple[LockfileScope, int], type] = {}	11✔
29
30
31	# Registrar types (pre-declaring to avoid repetition)
32	RegisterClassForVersion = Callable[[type["LockfileMetadata"]], type["LockfileMetadata"]]	11✔
33
34
35	def lockfile_metadata_registrar(scope: LockfileScope) -> Callable[[int], RegisterClassForVersion]:	11✔
36	"""Decorator factory -- returns a decorator that can be used to register Lockfile metadata
37	version subclasses belonging to a specific `LockfileScope`."""
38
39	def _lockfile_metadata_version(	11✔
40	version: int,
41	) -> RegisterClassForVersion:
42	"""Decorator to register a Lockfile metadata version subclass with a given version number.
43
44	The class must be a frozen dataclass
45	"""
46
47	def _dec(cls: type[LockfileMetadata]) -> type[LockfileMetadata]:	11✔
48	# Only frozen dataclasses may be registered as lockfile metadata:
49	cls_dataclass_params = getattr(cls, "__dataclass_params__", None)	11✔
50	if not cls_dataclass_params or not cls_dataclass_params.frozen:	11✔
51	raise ValueError(	×
52	softwrap(
53	"""
54	Classes registered with `_lockfile_metadata_version` may only be
55	frozen dataclasses
56	"""
57	)
58	)
59	_concrete_metadata_classes[(scope, version)] = cls	11✔
60	return cls	11✔
61
62	return _dec	11✔
63
64	return _lockfile_metadata_version	11✔
65
66
67	class InvalidLockfileError(Exception):	11✔
68	pass	11✔
69
70
71	class NoLockfileMetadataBlock(InvalidLockfileError):	11✔
72	pass	11✔
73
74
75	@dataclass(frozen=True)	11✔
76	class LockfileMetadata:	11✔
77	"""Base class for metadata that is attached to a given lockfile.
78
79	This class provides the external API for serializing, deserializing, and validating the
80	contents of individual lockfiles. New versions of metadata implement a concrete subclass and
81	provide deserialization and validation logic, along with specialist serialization logic.
82
83	To construct an instance of the most recent concrete subclass, call `LockfileMetadata.new()`.
84	"""
85
86	scope: ClassVar[LockfileScope]	11✔
87
88	@classmethod	11✔
89	def from_lockfile(	11✔
90	cls: type[_LockfileMetadataSubclass],
91	lockfile: bytes,
92	lockfile_path: str \| None = None,
93	resolve_name: str \| None = None,
94	*,
95	delimeter: str,
96	) -> _LockfileMetadataSubclass:
97	"""Parse and return the metadata from the lockfile's header.
98
99	This shouldn't be called on `LockfileMetadata`, but rather on the "base class" for your
100	metadata class. See the existing callers for an example.
101	"""
102	assert cls is not LockfileMetadata, "Call me on a subclass!"	9✔
103	in_metadata_block = False	9✔
104	metadata_lines = []	9✔
105	for line in lockfile.splitlines():	9✔
106	if line == f"{delimeter} {BEGIN_LOCKFILE_HEADER}".encode():	9✔
107	in_metadata_block = True	9✔
108	elif line == f"{delimeter} {END_LOCKFILE_HEADER}".encode():	9✔
109	break	9✔
110	elif in_metadata_block:	9✔
111	metadata_lines.append(line[len(delimeter) + 1 :])	9✔
112
113	error_suffix = softwrap(	9✔
114	f"""
115	To resolve this error, you will need to regenerate the lockfile by running
116	`{bin_name()} generate-lockfiles
117	"""
118	)
119	if resolve_name:	9✔
120	error_suffix += f" --resolve={resolve_name}"	1✔
121	error_suffix += "`."	9✔
122
123	if lockfile_path is not None and resolve_name is not None:	9✔
124	lockfile_description = f"the lockfile `{lockfile_path}` for `{resolve_name}`"	×
125	elif lockfile_path is not None:	9✔
126	lockfile_description = f"the lockfile `{lockfile_path}`"	×
127	elif resolve_name is not None:	9✔
128	lockfile_description = f"the lockfile for `{resolve_name}`"	1✔
129	else:
130	lockfile_description = "this lockfile"	9✔
131
132	if not metadata_lines:	9✔
133	raise NoLockfileMetadataBlock(	1✔
134	f"Could not find a Pants metadata block in {lockfile_description}. {error_suffix}"
135	)
136	try:	9✔
137	json_dict = json.loads(b"\n".join(metadata_lines))	9✔
138	except json.decoder.JSONDecodeError:	1✔
139	raise InvalidLockfileError(	1✔
140	softwrap(
141	f"""
142	Metadata header in {lockfile_description} is not a valid JSON string and can't
143	be decoded.
144	"""
145	)
146	+ error_suffix
147	)
148	return cls.from_json_dict(json_dict, lockfile_description, error_suffix)	9✔
149
150	@classmethod	11✔
151	def from_json_dict(	11✔
152	cls: type[_LockfileMetadataSubclass],
153	json_dict: dict[Any, Any],
154	lockfile_description: str,
155	error_suffix: str,
156	) -> _LockfileMetadataSubclass:
157	version = json_dict.get("version", 1)	9✔
158	concrete_class = _concrete_metadata_classes[(cls.scope, version)]	9✔
159
160	assert issubclass(concrete_class, cls)	9✔
161	assert concrete_class.scope == cls.scope, (	9✔
162	"The class used to call `from_json_dict` has a different scope than what was "
163	f"expected given the metadata. Expected '{cls.scope}', got '{concrete_class.scope}'",
164	)
165
166	return concrete_class._from_json_dict(json_dict, lockfile_description, error_suffix)	9✔
167
168	@classmethod	11✔
169	def _from_json_dict(	11✔
170	cls: type[_LockfileMetadataSubclass],
171	json_dict: dict[Any, Any],
172	lockfile_description: str,
173	error_suffix: str,
174	) -> _LockfileMetadataSubclass:
175	"""Construct a `LockfileMetadata` subclass from the supplied JSON dict.
176
177	* Not implemented. Subclasses should override. *
178
179
180	`lockfile_description` is a detailed, human-readable description of the lockfile, which can
181	be read by the user to figure out which lockfile is broken in case of an error.
182
183	`error_suffix` is a string describing how to fix the lockfile.
184	"""
185
186	raise NotImplementedError(	×
187	"`LockfileMetadata._from_json_dict` should not be directly called."
188	)
189
190	def to_json(self, with_description: str \| None = None) -> str:	11✔
191	metadata_dict = self.__render_header_dict()	4✔
192	if with_description is not None:	4✔
193	metadata_dict["description"] = with_description	×
194	return json.dumps(metadata_dict, ensure_ascii=True, indent=2)	4✔
195
196	def add_header_to_lockfile(	11✔
197	self, lockfile: bytes, *, delimeter: str, regenerate_command: str
198	) -> bytes:
199	metadata_json = self.to_json()	4✔
200	metadata_as_a_comment = "\n".join(f"{delimeter} {l}" for l in metadata_json.splitlines())	4✔
201
202	regenerate_command_bytes = "\n".join(	4✔
203	[
204	f"{delimeter} This lockfile was autogenerated by Pants. To regenerate, run:",
205	delimeter,
206	f"{delimeter} {regenerate_command}",
207	]
208	).encode()
209	header = "\n".join(	4✔
210	[
211	f"{delimeter} {BEGIN_LOCKFILE_HEADER}",
212	metadata_as_a_comment,
213	f"{delimeter} {END_LOCKFILE_HEADER}",
214	]
215	).encode("ascii")
216
217	return b"%b\n%b\n%b\n\n%b" % (	4✔
218	regenerate_command_bytes,
219	delimeter.encode(),
220	header,
221	lockfile,
222	)
223
224	def __render_header_dict(self) -> dict[Any, Any]:	11✔
225	"""Produce a dictionary to be serialized into the lockfile header.
226
227	Each class should implement a class method called `additional_header_attrs`, which returns a
228	`dict` containing the metadata attributes that should be stored in the lockfile.
229	"""
230
231	attrs: dict[Any, tuple[Any, type]] = {} # attr name -> (value, where we first saw it)	4✔
232	for cls in reversed(self.__class__.__mro__[:-1]):	4✔
233	new_attrs = cast(LockfileMetadata, cls).additional_header_attrs(self)	4✔
234	for attr in new_attrs:	4✔
235	if attr in attrs and attrs[attr][0] != new_attrs[attr]:	4✔
236	raise AssertionError(	×
237	softwrap(
238	f"""
239	Lockfile header attribute `{attr}` was returned by both
240	`{attrs[attr][1]}` and `{cls}`, returning different values. If these
241	classes return the same attribute, they must also return the same
242	value.
243	"""
244	)
245	)
246	attrs[attr] = new_attrs[attr], cls	4✔
247
248	return {key: val[0] for key, val in attrs.items()}	4✔
249
250	@classmethod	11✔
251	def additional_header_attrs(cls, instance: LockfileMetadata) -> dict[Any, Any]:	11✔
252	return {"version": instance.metadata_version()}	4✔
253
254	def metadata_version(self):	11✔
255	"""Returns the version number for this metadata class, or raises an exception.
256
257	To avoid raising an exception, ensure the subclass is decorated with
258	`lockfile_metadata_version`
259	"""
260	for (scope, ver), cls in _concrete_metadata_classes.items():	4✔
261	# Note that we do exact version matches so that authors can subclass earlier versions.
262	if type(self) is cls:	4✔
263	return ver	4✔
264	raise ValueError("Trying to serialize an unregistered `LockfileMetadata` subclass.")	×
265
266
267	def calculate_invalidation_digest(requirements: Iterable[str]) -> str:	11✔
268	"""Returns an invalidation digest for the given requirements."""
UNCOV 269	m = hashlib.sha256()	×
UNCOV 270	inputs = {	×
271	# `FrozenOrderedSet` deduplicates while keeping ordering, which speeds up the sorting if
272	# the input was already sorted.
273	"requirements": sorted(FrozenOrderedSet(requirements)),
274	}
UNCOV 275	m.update(json.dumps(inputs).encode("utf-8"))	×
UNCOV 276	return m.hexdigest()	×
277
278
279	T = TypeVar("T")	11✔
280
281
282	class LockfileMetadataValidation(Generic[T]):	11✔
283	"""Boolean-like value which additionally carries reasons why a validation failed."""
284
285	failure_reasons: set[T]	11✔
286
287	def __init__(self, failure_reasons: Iterable[T] = ()):	11✔
288	self.failure_reasons = set(failure_reasons)	9✔
289
290	def __bool__(self):	11✔
291	return not self.failure_reasons	9✔
292
293
294	def _get_metadata(	11✔
295	metadata: dict[Any, Any],
296	lockfile_description: str,
297	error_suffix: str,
298	) -> Callable[[str, type[T], Callable[[Any], T] \| None], T]:
299	"""Returns a function that will get a given key from the `metadata` dict, and optionally do some
300	verification and post-processing to return a value of the correct type."""
301
302	def get_metadata(key: str, type_: type[T], coerce: Callable[[Any], T] \| None) -> T:	9✔
303	val: Any
304	try:	9✔
305	val = metadata[key]	9✔
306	except KeyError:	×
307	raise InvalidLockfileError(	×
308	softwrap(
309	f"""
310	Required key `{key}` is not present in metadata header for
311	{lockfile_description}. {error_suffix}
312	"""
313	)
314	)
315
316	if not coerce:	9✔
317	if isinstance(val, type_):	1✔
318	return val	1✔
319
320	raise InvalidLockfileError(	×
321	softwrap(
322	f"""
323	Metadata value `{key}` in {lockfile_description} must
324	be a {type(type_).__name__}. {error_suffix}
325	"""
326	)
327	)
328	else:
329	try:	9✔
330	return coerce(val)	9✔
331	except Exception:	×
332	raise InvalidLockfileError(	×
333	softwrap(
334	f"""
335	Metadata value `{key}` in {lockfile_description} must be able to
336	be converted to a {type(type_).__name__}. {error_suffix}
337	"""
338	)
339	)
340
341	return get_metadata	9✔

pantsbuild / pants / 20147226056

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