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

pantsbuild / pants / 26080722777

19 May 2026 06:37AM UTC coverage: 52.106% (-11.5%) from 63.597%
26080722777

Pull #23250

github

web-flow
Merge 63ec06323 into 2693df832
Pull Request #23250: Feature: Add generic option to docker image

12 of 50 new or added lines in 3 files covered. (24.0%)

5382 existing lines in 201 files now uncovered.

32053 of 61515 relevant lines covered (52.11%)

1.04 hits per line

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

67.42
/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
2✔
5

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

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

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

20

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

25

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

30

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

34

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

62
        return _dec
2✔
63

64
    return _lockfile_metadata_version
2✔
65

66

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

70

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

74

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

88
    @classmethod
2✔
89
    def from_lockfile(
2✔
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!"
2✔
103
        in_metadata_block = False
2✔
104
        metadata_lines = []
2✔
105
        for line in lockfile.splitlines():
2✔
106
            if line == f"{delimeter} {BEGIN_LOCKFILE_HEADER}".encode():
2✔
107
                in_metadata_block = True
2✔
108
            elif line == f"{delimeter} {END_LOCKFILE_HEADER}".encode():
2✔
109
                break
2✔
110
            elif in_metadata_block:
2✔
111
                metadata_lines.append(line[len(delimeter) + 1 :])
2✔
112

113
        error_suffix = softwrap(
2✔
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:
2✔
UNCOV
120
            error_suffix += f" --resolve={resolve_name}"
×
121
        error_suffix += "`."
2✔
122

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

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

150
    @classmethod
2✔
151
    def from_json_dict(
2✔
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)
2✔
158
        if (cls.scope, version) not in _concrete_metadata_classes:
2✔
159
            raise InvalidLockfileError(
×
160
                f"Unknown lockfile metadata version {version} in {lockfile_description}. "
161
                f"{error_suffix}"
162
            )
163
        concrete_class = _concrete_metadata_classes[(cls.scope, version)]
2✔
164

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

171
        return concrete_class._from_json_dict(json_dict, lockfile_description, error_suffix)
2✔
172

173
    @classmethod
2✔
174
    def _from_json_dict(
2✔
175
        cls: type[_LockfileMetadataSubclass],
176
        json_dict: dict[Any, Any],
177
        lockfile_description: str,
178
        error_suffix: str,
179
    ) -> _LockfileMetadataSubclass:
180
        """Construct a `LockfileMetadata` subclass from the supplied JSON dict.
181

182
        *** Not implemented. Subclasses should override. ***
183

184

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

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

191
        raise NotImplementedError(
×
192
            "`LockfileMetadata._from_json_dict` should not be directly called."
193
        )
194

195
    def to_json(self, with_description: str | None = None) -> str:
2✔
UNCOV
196
        metadata_dict = self.__render_header_dict()
×
UNCOV
197
        if with_description is not None:
×
UNCOV
198
            metadata_dict["description"] = with_description
×
UNCOV
199
        return json.dumps(metadata_dict, ensure_ascii=True, indent=2)
×
200

201
    def add_header_to_lockfile(
2✔
202
        self, lockfile: bytes, *, delimeter: str, regenerate_command: str
203
    ) -> bytes:
UNCOV
204
        metadata_json = self.to_json()
×
UNCOV
205
        metadata_as_a_comment = "\n".join(f"{delimeter} {l}" for l in metadata_json.splitlines())
×
206

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

UNCOV
222
        return b"%b\n%b\n%b\n\n%b" % (
×
223
            regenerate_command_bytes,
224
            delimeter.encode(),
225
            header,
226
            lockfile,
227
        )
228

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

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

UNCOV
236
        attrs: dict[Any, tuple[Any, type]] = {}  # attr name -> (value, where we first saw it)
×
UNCOV
237
        for cls in reversed(self.__class__.__mro__[:-1]):
×
UNCOV
238
            new_attrs = cast(LockfileMetadata, cls).additional_header_attrs(self)
×
UNCOV
239
            for attr in new_attrs:
×
UNCOV
240
                if attr in attrs and attrs[attr][0] != new_attrs[attr]:
×
241
                    raise AssertionError(
×
242
                        softwrap(
243
                            f"""
244
                            Lockfile header attribute `{attr}` was returned by both
245
                            `{attrs[attr][1]}` and `{cls}`, returning different values. If these
246
                            classes return the same attribute, they must also return the same
247
                            value.
248
                            """
249
                        )
250
                    )
UNCOV
251
                attrs[attr] = new_attrs[attr], cls
×
252

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

255
    @classmethod
2✔
256
    def additional_header_attrs(cls, instance: LockfileMetadata) -> dict[Any, Any]:
2✔
UNCOV
257
        return {"version": instance.metadata_version()}
×
258

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

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

271

272
def calculate_invalidation_digest(requirements: Iterable[str]) -> str:
2✔
273
    """Returns an invalidation digest for the given requirements."""
274
    m = hashlib.sha256()
×
275
    inputs = {
×
276
        # `FrozenOrderedSet` deduplicates while keeping ordering, which speeds up the sorting if
277
        # the input was already sorted.
278
        "requirements": sorted(FrozenOrderedSet(requirements)),
279
    }
280
    m.update(json.dumps(inputs).encode("utf-8"))
×
281
    return m.hexdigest()
×
282

283

284
T = TypeVar("T")
2✔
285

286

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

290
    failure_reasons: set[T]
2✔
291

292
    def __init__(self, failure_reasons: Iterable[T] = ()):
2✔
293
        self.failure_reasons = set(failure_reasons)
2✔
294

295
    def __bool__(self):
2✔
296
        return not self.failure_reasons
2✔
297

298

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

307
    def get_metadata(key: str, type_: type[T], coerce: Callable[[Any], T] | None) -> T:
2✔
308
        val: Any
309
        try:
2✔
310
            val = metadata[key]
2✔
311
        except KeyError:
×
312
            raise InvalidLockfileError(
×
313
                softwrap(
314
                    f"""
315
                    Required key `{key}` is not present in metadata header for
316
                    {lockfile_description}. {error_suffix}
317
                    """
318
                )
319
            )
320

321
        if not coerce:
2✔
322
            if isinstance(val, type_):
×
323
                return val
×
324

325
            raise InvalidLockfileError(
×
326
                softwrap(
327
                    f"""
328
                    Metadata value `{key}` in {lockfile_description} must
329
                    be a {type(type_).__name__}. {error_suffix}
330
                    """
331
                )
332
            )
333
        else:
334
            try:
2✔
335
                return coerce(val)
2✔
336
            except Exception:
×
337
                raise InvalidLockfileError(
×
338
                    softwrap(
339
                        f"""
340
                        Metadata value `{key}` in {lockfile_description} must be able to
341
                        be converted to a {type(type_).__name__}. {error_suffix}
342
                        """
343
                    )
344
                )
345

346
    return get_metadata
2✔
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

© 2026 Coveralls, Inc