• Home
  • Features
  • Pricing
  • Docs
  • Announcements
  • Sign In

pantsbuild / pants / 19015773527

02 Nov 2025 05:33PM UTC coverage: 17.872% (-62.4%) from 80.3%
19015773527

Pull #22816

github

web-flow
Merge a12d75757 into 6c024e162
Pull Request #22816: Update Pants internal Python to 3.14

4 of 5 new or added lines in 3 files covered. (80.0%)

28452 existing lines in 683 files now uncovered.

9831 of 55007 relevant lines covered (17.87%)

0.18 hits per line

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

40.77
/src/python/pants/core/util_rules/lockfile_metadata.py
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
1✔
5

6
import hashlib
1✔
7
import json
1✔
8
from collections.abc import Callable, Iterable
1✔
9
from dataclasses import dataclass
1✔
10
from enum import Enum
1✔
11
from typing import Any, ClassVar, Generic, TypeVar, cast
1✔
12

13
from pants.util.docutil import bin_name
1✔
14
from pants.util.ordered_set import FrozenOrderedSet
1✔
15
from pants.util.strutil import softwrap
1✔
16

17
BEGIN_LOCKFILE_HEADER = "--- BEGIN PANTS LOCKFILE METADATA: DO NOT EDIT OR REMOVE ---"
1✔
18
END_LOCKFILE_HEADER = "--- END PANTS LOCKFILE METADATA ---"
1✔
19

20

21
class LockfileScope(Enum):
1✔
22
    JVM = "jvm"
1✔
23
    PYTHON = "python"
1✔
24

25

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

30

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

34

35
def lockfile_metadata_registrar(scope: LockfileScope) -> Callable[[int], RegisterClassForVersion]:
1✔
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(
1✔
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]:
1✔
48
            # Only frozen dataclasses may be registered as lockfile metadata:
49
            cls_dataclass_params = getattr(cls, "__dataclass_params__", None)
1✔
50
            if not cls_dataclass_params or not cls_dataclass_params.frozen:
1✔
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
1✔
60
            return cls
1✔
61

62
        return _dec
1✔
63

64
    return _lockfile_metadata_version
1✔
65

66

67
class InvalidLockfileError(Exception):
1✔
68
    pass
1✔
69

70

71
class NoLockfileMetadataBlock(InvalidLockfileError):
1✔
72
    pass
1✔
73

74

75
@dataclass(frozen=True)
1✔
76
class LockfileMetadata:
1✔
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]
1✔
87

88
    @classmethod
1✔
89
    def from_lockfile(
1✔
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
        """
UNCOV
102
        assert cls is not LockfileMetadata, "Call me on a subclass!"
×
UNCOV
103
        in_metadata_block = False
×
UNCOV
104
        metadata_lines = []
×
UNCOV
105
        for line in lockfile.splitlines():
×
UNCOV
106
            if line == f"{delimeter} {BEGIN_LOCKFILE_HEADER}".encode():
×
UNCOV
107
                in_metadata_block = True
×
UNCOV
108
            elif line == f"{delimeter} {END_LOCKFILE_HEADER}".encode():
×
UNCOV
109
                break
×
UNCOV
110
            elif in_metadata_block:
×
UNCOV
111
                metadata_lines.append(line[len(delimeter) + 1 :])
×
112

UNCOV
113
        error_suffix = softwrap(
×
114
            f"""
115
            To resolve this error, you will need to regenerate the lockfile by running
116
            `{bin_name()} generate-lockfiles
117
            """
118
        )
UNCOV
119
        if resolve_name:
×
UNCOV
120
            error_suffix += f" --resolve={resolve_name}"
×
UNCOV
121
        error_suffix += "`."
×
122

UNCOV
123
        if lockfile_path is not None and resolve_name is not None:
×
124
            lockfile_description = f"the lockfile `{lockfile_path}` for `{resolve_name}`"
×
UNCOV
125
        elif lockfile_path is not None:
×
126
            lockfile_description = f"the lockfile `{lockfile_path}`"
×
UNCOV
127
        elif resolve_name is not None:
×
UNCOV
128
            lockfile_description = f"the lockfile for `{resolve_name}`"
×
129
        else:
UNCOV
130
            lockfile_description = "this lockfile"
×
131

UNCOV
132
        if not metadata_lines:
×
UNCOV
133
            raise NoLockfileMetadataBlock(
×
134
                f"Could not find a Pants metadata block in {lockfile_description}. {error_suffix}"
135
            )
UNCOV
136
        try:
×
UNCOV
137
            json_dict = json.loads(b"\n".join(metadata_lines))
×
UNCOV
138
        except json.decoder.JSONDecodeError:
×
UNCOV
139
            raise InvalidLockfileError(
×
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
            )
UNCOV
148
        return cls.from_json_dict(json_dict, lockfile_description, error_suffix)
×
149

150
    @classmethod
1✔
151
    def from_json_dict(
1✔
152
        cls: type[_LockfileMetadataSubclass],
153
        json_dict: dict[Any, Any],
154
        lockfile_description: str,
155
        error_suffix: str,
156
    ) -> _LockfileMetadataSubclass:
UNCOV
157
        version = json_dict.get("version", 1)
×
UNCOV
158
        concrete_class = _concrete_metadata_classes[(cls.scope, version)]
×
159

UNCOV
160
        assert issubclass(concrete_class, cls)
×
UNCOV
161
        assert concrete_class.scope == cls.scope, (
×
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

UNCOV
166
        return concrete_class._from_json_dict(json_dict, lockfile_description, error_suffix)
×
167

168
    @classmethod
1✔
169
    def _from_json_dict(
1✔
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:
1✔
UNCOV
191
        metadata_dict = self.__render_header_dict()
×
UNCOV
192
        if with_description is not None:
×
193
            metadata_dict["description"] = with_description
×
UNCOV
194
        return json.dumps(metadata_dict, ensure_ascii=True, indent=2)
×
195

196
    def add_header_to_lockfile(
1✔
197
        self, lockfile: bytes, *, delimeter: str, regenerate_command: str
198
    ) -> bytes:
UNCOV
199
        metadata_json = self.to_json()
×
UNCOV
200
        metadata_as_a_comment = "\n".join(f"{delimeter} {l}" for l in metadata_json.splitlines())
×
201

UNCOV
202
        regenerate_command_bytes = "\n".join(
×
203
            [
204
                f"{delimeter} This lockfile was autogenerated by Pants. To regenerate, run:",
205
                delimeter,
206
                f"{delimeter}    {regenerate_command}",
207
            ]
208
        ).encode()
UNCOV
209
        header = "\n".join(
×
210
            [
211
                f"{delimeter} {BEGIN_LOCKFILE_HEADER}",
212
                metadata_as_a_comment,
213
                f"{delimeter} {END_LOCKFILE_HEADER}",
214
            ]
215
        ).encode("ascii")
216

UNCOV
217
        return b"%b\n%b\n%b\n\n%b" % (
×
218
            regenerate_command_bytes,
219
            delimeter.encode(),
220
            header,
221
            lockfile,
222
        )
223

224
    def __render_header_dict(self) -> dict[Any, Any]:
1✔
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

UNCOV
231
        attrs: dict[Any, tuple[Any, type]] = {}  # attr name -> (value, where we first saw it)
×
UNCOV
232
        for cls in reversed(self.__class__.__mro__[:-1]):
×
UNCOV
233
            new_attrs = cast(LockfileMetadata, cls).additional_header_attrs(self)
×
UNCOV
234
            for attr in new_attrs:
×
UNCOV
235
                if attr in attrs and attrs[attr][0] != new_attrs[attr]:
×
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
                    )
UNCOV
246
                attrs[attr] = new_attrs[attr], cls
×
247

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

250
    @classmethod
1✔
251
    def additional_header_attrs(cls, instance: LockfileMetadata) -> dict[Any, Any]:
1✔
UNCOV
252
        return {"version": instance.metadata_version()}
×
253

254
    def metadata_version(self):
1✔
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
        """
UNCOV
260
        for (scope, ver), cls in _concrete_metadata_classes.items():
×
261
            # Note that we do exact version matches so that authors can subclass earlier versions.
UNCOV
262
            if type(self) is cls:
×
UNCOV
263
                return ver
×
264
        raise ValueError("Trying to serialize an unregistered `LockfileMetadata` subclass.")
×
265

266

267
def calculate_invalidation_digest(requirements: Iterable[str]) -> str:
1✔
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")
1✔
280

281

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

285
    failure_reasons: set[T]
1✔
286

287
    def __init__(self, failure_reasons: Iterable[T] = ()):
1✔
UNCOV
288
        self.failure_reasons = set(failure_reasons)
×
289

290
    def __bool__(self):
1✔
UNCOV
291
        return not self.failure_reasons
×
292

293

294
def _get_metadata(
1✔
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

UNCOV
302
    def get_metadata(key: str, type_: type[T], coerce: Callable[[Any], T] | None) -> T:
×
303
        val: Any
UNCOV
304
        try:
×
UNCOV
305
            val = metadata[key]
×
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

UNCOV
316
        if not coerce:
×
UNCOV
317
            if isinstance(val, type_):
×
UNCOV
318
                return val
×
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:
UNCOV
329
            try:
×
UNCOV
330
                return coerce(val)
×
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

UNCOV
341
    return get_metadata
×
STATUS · Troubleshooting · Open an Issue · Sales · Support · CAREERS · ENTERPRISE · START FREE · SCHEDULE DEMO
ANNOUNCEMENTS · TWITTER · TOS & SLA · Supported CI Services · What's a CI service? · Automated Testing

© 2025 Coveralls, Inc