26380816428

Committed 25 May 2026 02:57AM UTC coverage: 52.312% (-40.6%) from 92.89%

Build # 26380816428

Build Type

Pull #23368

github

Committed by

web-flow

Commit Message

Merge 7410b48e1 into 7b1060c81

Pull Request Pull Request #23368: Run Linux ARM CI on Depot runners (Cherry-pick of #23363)

Coverage Stats

31807 of 60802 relevant lines covered (52.31%)

1.05 hits per line

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

34.52

/src/python/pants/backend/python/util_rules/lockfile_diff.py

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

from __future__ import annotations

import itertools
import json
import logging
import tomllib
from collections.abc import Mapping
from dataclasses import dataclass
from typing import Any

from packaging.version import Version, parse

from pants.backend.python.util_rules.lockfile_metadata import LockfileFormat
from pants.backend.python.util_rules.pex_requirements import (
    LoadedLockfileRequest,
    Lockfile,
    load_lockfile,
    strip_comments_from_pex_json_lockfile,
)
from pants.base.exceptions import EngineError
from pants.core.goals.generate_lockfiles import LockfileDiff, LockfilePackages, PackageName
from pants.engine.fs import Digest
from pants.engine.intrinsics import get_digest_contents
from pants.engine.rules import implicitly
from pants.util.frozendict import FrozenDict

logger = logging.getLogger(__name__)


@dataclass(frozen=True, order=True)
class PythonRequirementVersion:
    _parsed: Version

    @classmethod
    def parse(cls, version: str) -> PythonRequirementVersion:
        return cls(parse(version))

    def __str__(self) -> str:
        return str(self._parsed)

    def __getattr__(self, key: str) -> Any:
        return getattr(self._parsed, key)


def _pex_lockfile_requirements(
    lockfile_data: Mapping[str, Any] | None, path: str | None = None
) -> LockfilePackages:
    if not lockfile_data:
        return LockfilePackages({})

    try:
        # Setup generators
        locked_resolves = (
            (
                (PackageName(r["project_name"]), PythonRequirementVersion.parse(r["version"]))
                for r in resolve["locked_requirements"]
            )
            for resolve in lockfile_data["locked_resolves"]
        )
        requirements = dict(itertools.chain.from_iterable(locked_resolves))
    except KeyError as e:
        if path:
            logger.warning(f"{path}: Failed to parse lockfile: {e}")

        requirements = {}

    return LockfilePackages(requirements)


def _uv_lockfile_requirements(
    lockfile_data: Mapping[str, Any] | None, path: str | None = None
) -> LockfilePackages:
    if not lockfile_data:
        return LockfilePackages({})

    requirements = {}
    for pkg in lockfile_data.get("package", []):
        try:
            name = pkg["name"]
            version = pkg.get("version")
            # Skip the synthetic virtual package, it's not interesting in diffs.
            if version is not None and not name.startswith("pants-lockfile-for-"):
                requirements[PackageName(name)] = PythonRequirementVersion.parse(version)
        except Exception as e:
            if path:
                logger.warning(f"{path}: Failed to parse package entry in lockfile: {e}")

    return LockfilePackages(requirements)


def _parse_lockfile_packages(
    content: bytes, lockfile_format: LockfileFormat, path: str | None = None
) -> LockfilePackages:
    """Parse the packages from lockfile content according to its format."""
    try:
        match lockfile_format:
            case LockfileFormat.PEX:
                # strip_comments_from_pex_json_lockfile is idempotent, so safe to call on
                # already-stripped content (e.g. when content was loaded via load_lockfile).
                stripped = strip_comments_from_pex_json_lockfile(content)
                data = FrozenDict.deep_freeze(json.loads(stripped))
                return _pex_lockfile_requirements(data, path)
            case LockfileFormat.UV:
                data = FrozenDict.deep_freeze(tomllib.loads(content.decode()))
                return _uv_lockfile_requirements(data, path)
            case LockfileFormat.CONSTRAINTS_DEPRECATED:
                # These can't meaningfully be diffed.
                return LockfilePackages({})
            case _:
                raise ValueError(f"Unrecognized lockfile format: {lockfile_format}")
    except Exception as e:
        if path:
            logger.debug(f"{path}: Failed to parse lockfile contents: {e}")
        return LockfilePackages({})


async def _generate_lockfile_diff(
    digest: Digest, resolve_name: str, path: str, new_format: LockfileFormat
) -> LockfileDiff:
    """Generate a diff between the newly generated lockfile and the existing one on disk.

    Handles all combinations of old vs. new and pex vs. uv lockfile formats.
    """
    new_digest_contents = await get_digest_contents(digest)
    new_content = next(c for c in new_digest_contents if c.path == path).content
    new_packages = _parse_lockfile_packages(new_content, new_format, path)

    old_packages = LockfilePackages({})
    try:
        loaded = await load_lockfile(
            LoadedLockfileRequest(
                Lockfile(
                    url=path,
                    url_description_of_origin="existing lockfile",
                    resolve_name=resolve_name,
                )
            ),
            **implicitly(),
        )
        old_content_entries = await get_digest_contents(loaded.lockfile_digest)
        old_content = next(iter(old_content_entries)).content
        old_packages = _parse_lockfile_packages(old_content, loaded.lockfile_format, path)
    except EngineError:
        # May fail if the file doesn't exist, which is expected the first time a new lockfile
        # is generated.
        pass

    return LockfileDiff.create(
        path=path,
        resolve_name=resolve_name,
        old=old_packages,
        new=new_packages,
    )

1	# Copyright 2022 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 itertools	2✔
7	import json	2✔
8	import logging	2✔
9	import tomllib	2✔
10	from collections.abc import Mapping	2✔
11	from dataclasses import dataclass	2✔
12	from typing import Any	2✔
13
14	from packaging.version import Version, parse	2✔
15
16	from pants.backend.python.util_rules.lockfile_metadata import LockfileFormat	2✔
17	from pants.backend.python.util_rules.pex_requirements import (	2✔
18	LoadedLockfileRequest,
19	Lockfile,
20	load_lockfile,
21	strip_comments_from_pex_json_lockfile,
22	)
23	from pants.base.exceptions import EngineError	2✔
24	from pants.core.goals.generate_lockfiles import LockfileDiff, LockfilePackages, PackageName	2✔
25	from pants.engine.fs import Digest	2✔
26	from pants.engine.intrinsics import get_digest_contents	2✔
27	from pants.engine.rules import implicitly	2✔
28	from pants.util.frozendict import FrozenDict	2✔
29
30	logger = logging.getLogger(__name__)	2✔
31
32
33	@dataclass(frozen=True, order=True)	2✔
34	class PythonRequirementVersion:	2✔
35	_parsed: Version	2✔
36
37	@classmethod	2✔
38	def parse(cls, version: str) -> PythonRequirementVersion:	2✔
39	return cls(parse(version))	×
40
41	def __str__(self) -> str:	2✔
42	return str(self._parsed)	×
43
44	def __getattr__(self, key: str) -> Any:	2✔
45	return getattr(self._parsed, key)	×
46
47
48	def _pex_lockfile_requirements(	2✔
49	lockfile_data: Mapping[str, Any] \| None, path: str \| None = None
50	) -> LockfilePackages:
51	if not lockfile_data:	×
52	return LockfilePackages({})	×
53
54	try:	×
55	# Setup generators
56	locked_resolves = (	×
57	(
58	(PackageName(r["project_name"]), PythonRequirementVersion.parse(r["version"]))
59	for r in resolve["locked_requirements"]
60	)
61	for resolve in lockfile_data["locked_resolves"]
62	)
63	requirements = dict(itertools.chain.from_iterable(locked_resolves))	×
64	except KeyError as e:	×
65	if path:	×
66	logger.warning(f"{path}: Failed to parse lockfile: {e}")	×
67
68	requirements = {}	×
69
70	return LockfilePackages(requirements)	×
71
72
73	def _uv_lockfile_requirements(	2✔
74	lockfile_data: Mapping[str, Any] \| None, path: str \| None = None
75	) -> LockfilePackages:
76	if not lockfile_data:	×
77	return LockfilePackages({})	×
78
79	requirements = {}	×
80	for pkg in lockfile_data.get("package", []):	×
81	try:	×
82	name = pkg["name"]	×
83	version = pkg.get("version")	×
84	# Skip the synthetic virtual package, it's not interesting in diffs.
85	if version is not None and not name.startswith("pants-lockfile-for-"):	×
86	requirements[PackageName(name)] = PythonRequirementVersion.parse(version)	×
87	except Exception as e:	×
88	if path:	×
89	logger.warning(f"{path}: Failed to parse package entry in lockfile: {e}")	×
90
91	return LockfilePackages(requirements)	×
92
93
94	def _parse_lockfile_packages(	2✔
95	content: bytes, lockfile_format: LockfileFormat, path: str \| None = None
96	) -> LockfilePackages:
97	"""Parse the packages from lockfile content according to its format."""
98	try:	×
99	match lockfile_format:	×
100	case LockfileFormat.PEX:	×
101	# strip_comments_from_pex_json_lockfile is idempotent, so safe to call on
102	# already-stripped content (e.g. when content was loaded via load_lockfile).
103	stripped = strip_comments_from_pex_json_lockfile(content)	×
104	data = FrozenDict.deep_freeze(json.loads(stripped))	×
105	return _pex_lockfile_requirements(data, path)	×
106	case LockfileFormat.UV:	×
107	data = FrozenDict.deep_freeze(tomllib.loads(content.decode()))	×
108	return _uv_lockfile_requirements(data, path)	×
109	case LockfileFormat.CONSTRAINTS_DEPRECATED:	×
110	# These can't meaningfully be diffed.
111	return LockfilePackages({})	×
112	case _:	×
113	raise ValueError(f"Unrecognized lockfile format: {lockfile_format}")	×
114	except Exception as e:	×
115	if path:	×
116	logger.debug(f"{path}: Failed to parse lockfile contents: {e}")	×
117	return LockfilePackages({})	×
118
119
120	async def _generate_lockfile_diff(	2✔
121	digest: Digest, resolve_name: str, path: str, new_format: LockfileFormat
122	) -> LockfileDiff:
123	"""Generate a diff between the newly generated lockfile and the existing one on disk.
124
125	Handles all combinations of old vs. new and pex vs. uv lockfile formats.
126	"""
127	new_digest_contents = await get_digest_contents(digest)	×
128	new_content = next(c for c in new_digest_contents if c.path == path).content	×
129	new_packages = _parse_lockfile_packages(new_content, new_format, path)	×
130
131	old_packages = LockfilePackages({})	×
132	try:	×
133	loaded = await load_lockfile(	×
134	LoadedLockfileRequest(
135	Lockfile(
136	url=path,
137	url_description_of_origin="existing lockfile",
138	resolve_name=resolve_name,
139	)
140	),
141	**implicitly(),
142	)
143	old_content_entries = await get_digest_contents(loaded.lockfile_digest)	×
144	old_content = next(iter(old_content_entries)).content	×
145	old_packages = _parse_lockfile_packages(old_content, loaded.lockfile_format, path)	×
146	except EngineError:	×
147	# May fail if the file doesn't exist, which is expected the first time a new lockfile
148	# is generated.
149	pass	×
150
151	return LockfileDiff.create(	×
152	path=path,
153	resolve_name=resolve_name,
154	old=old_packages,
155	new=new_packages,
156	)

pantsbuild / pants / 26380816428

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