25405422172

Committed 05 May 2026 10:18PM UTC coverage: 92.879% (-0.07%) from 92.944%

Build # 25405422172

Build Type

Pull #23319

github

Committed by

web-flow

Commit Message

Merge c82d0f333 into e8b784f89

Pull Request Pull Request #23319: [pants_ng] Scaffolding for a pants_ng mode.

Coverage Stats

25 of 76 new or added lines in 9 files covered. (32.89%)

209 existing lines in 15 files now uncovered.

92234 of 99306 relevant lines covered (92.88%)

4.05 hits per line

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

61.9

/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	12✔
5
6	import itertools	12✔
7	import json	12✔
8	import logging	12✔
9	import tomllib	12✔
10	from collections.abc import Mapping	12✔
11	from dataclasses import dataclass	12✔
12	from typing import Any	12✔
13
14	from packaging.version import Version, parse	12✔
15
16	from pants.backend.python.util_rules.lockfile_metadata import LockfileFormat	12✔
17	from pants.backend.python.util_rules.pex_requirements import (	12✔
18	LoadedLockfileRequest,
19	Lockfile,
20	load_lockfile,
21	strip_comments_from_pex_json_lockfile,
22	)
23	from pants.base.exceptions import EngineError	12✔
24	from pants.core.goals.generate_lockfiles import LockfileDiff, LockfilePackages, PackageName	12✔
25	from pants.engine.fs import Digest	12✔
26	from pants.engine.intrinsics import get_digest_contents	12✔
27	from pants.engine.rules import implicitly	12✔
28	from pants.util.frozendict import FrozenDict	12✔
29
30	logger = logging.getLogger(__name__)	12✔
31
32
33	@dataclass(frozen=True, order=True)	12✔
34	class PythonRequirementVersion:	12✔
35	_parsed: Version	12✔
36
37	@classmethod	12✔
38	def parse(cls, version: str) -> PythonRequirementVersion:	12✔
39	return cls(parse(version))	1✔
40
41	def __str__(self) -> str:	12✔
42	return str(self._parsed)	1✔
43
44	def __getattr__(self, key: str) -> Any:	12✔
UNCOV 45	return getattr(self._parsed, key)	×
46
47
48	def _pex_lockfile_requirements(	12✔
49	lockfile_data: Mapping[str, Any] \| None, path: str \| None = None
50	) -> LockfilePackages:
51	if not lockfile_data:	1✔
UNCOV 52	return LockfilePackages({})	×
53
54	try:	1✔
55	# Setup generators
56	locked_resolves = (	1✔
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))	1✔
64	except KeyError as e:	×
UNCOV 65	if path:	×
66	logger.warning(f"{path}: Failed to parse lockfile: {e}")	×
67
UNCOV 68	requirements = {}	×
69
70	return LockfilePackages(requirements)	1✔
71
72
73	def _uv_lockfile_requirements(	12✔
74	lockfile_data: Mapping[str, Any] \| None, path: str \| None = None
75	) -> LockfilePackages:
UNCOV 76	if not lockfile_data:	×
77	return LockfilePackages({})	×
78
UNCOV 79	requirements = {}	×
80	for pkg in lockfile_data.get("package", []):	×
UNCOV 81	try:	×
UNCOV 82	name = pkg["name"]	×
UNCOV 83	version = pkg.get("version")	×
84	# Skip the synthetic virtual package, it's not interesting in diffs.
UNCOV 85	if version is not None and not name.startswith("pants-lockfile-for-"):	×
UNCOV 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
UNCOV 91	return LockfilePackages(requirements)	×
92
93
94	def _parse_lockfile_packages(	12✔
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:	1✔
99	match lockfile_format:	1✔
100	case LockfileFormat.PEX:	1✔
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)	1✔
104	data = FrozenDict.deep_freeze(json.loads(stripped))	1✔
105	return _pex_lockfile_requirements(data, path)	1✔
UNCOV 106	case LockfileFormat.UV:	×
UNCOV 107	data = FrozenDict.deep_freeze(tomllib.loads(content.decode()))	×
UNCOV 108	return _uv_lockfile_requirements(data, path)	×
UNCOV 109	case LockfileFormat.CONSTRAINTS_DEPRECATED:	×
110	# These can't meaningfully be diffed.
UNCOV 111	return LockfilePackages({})	×
UNCOV 112	case _:	×
UNCOV 113	raise ValueError(f"Unrecognized lockfile format: {lockfile_format}")	×
UNCOV 114	except Exception as e:	×
UNCOV 115	if path:	×
UNCOV 116	logger.debug(f"{path}: Failed to parse lockfile contents: {e}")	×
UNCOV 117	return LockfilePackages({})	×
118
119
120	async def _generate_lockfile_diff(	12✔
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)	1✔
128	new_content = next(c for c in new_digest_contents if c.path == path).content	1✔
129	new_packages = _parse_lockfile_packages(new_content, new_format, path)	1✔
130
131	old_packages = LockfilePackages({})	1✔
132	try:	1✔
133	loaded = await load_lockfile(	1✔
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)	1✔
144	old_content = next(iter(old_content_entries)).content	1✔
145	old_packages = _parse_lockfile_packages(old_content, loaded.lockfile_format, path)	1✔
UNCOV 146	except EngineError:	×
147	# May fail if the file doesn't exist, which is expected the first time a new lockfile
148	# is generated.
UNCOV 149	pass	×
150
151	return LockfileDiff.create(	1✔
152	path=path,
153	resolve_name=resolve_name,
154	old=old_packages,
155	new=new_packages,
156	)

pantsbuild / pants / 25405422172

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