19381742489

Committed 15 Nov 2025 12:52AM UTC coverage: 49.706% (-30.6%) from 80.29%

Build # 19381742489

Build Type

Pull #22890

github

Committed by

web-flow

Commit Message

Merge d961abf79 into 42e1ebd41

Pull Request Pull Request #22890: Updated all python subsystem constraints to 3.14

Run Details

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

14659 existing lines in 485 files now uncovered.

31583 of 63540 relevant lines covered (49.71%)

0.79 hits per line

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

45.76

/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
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.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)


async def _parse_lockfile(lockfile: Lockfile) -> FrozenDict[str, Any] | None:
    try:
        loaded = await load_lockfile(LoadedLockfileRequest(lockfile), **implicitly())
        fc = await get_digest_contents(loaded.lockfile_digest)
        parsed = await _parse_lockfile_content(next(iter(fc)).content, lockfile.url)
        return parsed
    except EngineError:
        # May fail in case the file doesn't exist, which is expected when parsing the "old" lockfile
        # the first time a new lockfile is generated.
        return None


async def _parse_lockfile_content(content: bytes, url: str) -> FrozenDict[str, Any] | None:
    try:
        parsed_lockfile = json.loads(content)
        return FrozenDict.deep_freeze(parsed_lockfile)
    except json.JSONDecodeError as e:
        logger.debug(f"{url}: Failed to parse lockfile contents: {e}")
        return None


async def _generate_python_lockfile_diff(
    digest: Digest, resolve_name: str, path: str
) -> LockfileDiff:
    new_digest_contents = await get_digest_contents(digest)
    new_content = next(c for c in new_digest_contents if c.path == path).content
    new_content = strip_comments_from_pex_json_lockfile(new_content)
    new = await _parse_lockfile_content(new_content, path)
    old = await _parse_lockfile(
        Lockfile(
            url=path,
            url_description_of_origin="existing lockfile",
            resolve_name=resolve_name,
        )
    )
    return LockfileDiff.create(
        path=path,
        resolve_name=resolve_name,
        old=_pex_lockfile_requirements(old),
        new=_pex_lockfile_requirements(new, path),
    )

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	from collections.abc import Mapping	2✔
10	from dataclasses import dataclass	2✔
11	from typing import Any	2✔
12
13	from packaging.version import Version, parse	2✔
14
15	from pants.backend.python.util_rules.pex_requirements import (	2✔
16	LoadedLockfileRequest,
17	Lockfile,
18	load_lockfile,
19	strip_comments_from_pex_json_lockfile,
20	)
21	from pants.base.exceptions import EngineError	2✔
22	from pants.core.goals.generate_lockfiles import LockfileDiff, LockfilePackages, PackageName	2✔
23	from pants.engine.fs import Digest	2✔
24	from pants.engine.intrinsics import get_digest_contents	2✔
25	from pants.engine.rules import implicitly	2✔
26	from pants.util.frozendict import FrozenDict	2✔
27
28	logger = logging.getLogger(__name__)	2✔
29
30
31	@dataclass(frozen=True, order=True)	2✔
32	class PythonRequirementVersion:	2✔
33	_parsed: Version	2✔
34
35	@classmethod	2✔
36	def parse(cls, version: str) -> PythonRequirementVersion:	2✔
37	return cls(parse(version))	×
38
39	def __str__(self) -> str:	2✔
UNCOV 40	return str(self._parsed)	×
41
42	def __getattr__(self, key: str) -> Any:	2✔
43	return getattr(self._parsed, key)	×
44
45
46	def _pex_lockfile_requirements(	2✔
47	lockfile_data: Mapping[str, Any] \| None, path: str \| None = None
48	) -> LockfilePackages:
49	if not lockfile_data:	×
50	return LockfilePackages({})	×
51
52	try:	×
53	# Setup generators
54	locked_resolves = (	×
55	(
56	(PackageName(r["project_name"]), PythonRequirementVersion.parse(r["version"]))
57	for r in resolve["locked_requirements"]
58	)
59	for resolve in lockfile_data["locked_resolves"]
60	)
61	requirements = dict(itertools.chain.from_iterable(locked_resolves))	×
62	except KeyError as e:	×
63	if path:	×
64	logger.warning(f"{path}: Failed to parse lockfile: {e}")	×
65
66	requirements = {}	×
67
68	return LockfilePackages(requirements)	×
69
70
71	async def _parse_lockfile(lockfile: Lockfile) -> FrozenDict[str, Any] \| None:	2✔
72	try:	×
73	loaded = await load_lockfile(LoadedLockfileRequest(lockfile), **implicitly())	×
74	fc = await get_digest_contents(loaded.lockfile_digest)	×
75	parsed = await _parse_lockfile_content(next(iter(fc)).content, lockfile.url)	×
76	return parsed	×
77	except EngineError:	×
78	# May fail in case the file doesn't exist, which is expected when parsing the "old" lockfile
79	# the first time a new lockfile is generated.
80	return None	×
81
82
83	async def _parse_lockfile_content(content: bytes, url: str) -> FrozenDict[str, Any] \| None:	2✔
84	try:	×
85	parsed_lockfile = json.loads(content)	×
86	return FrozenDict.deep_freeze(parsed_lockfile)	×
87	except json.JSONDecodeError as e:	×
88	logger.debug(f"{url}: Failed to parse lockfile contents: {e}")	×
89	return None	×
90
91
92	async def _generate_python_lockfile_diff(	2✔
93	digest: Digest, resolve_name: str, path: str
94	) -> LockfileDiff:
95	new_digest_contents = await get_digest_contents(digest)	×
96	new_content = next(c for c in new_digest_contents if c.path == path).content	×
97	new_content = strip_comments_from_pex_json_lockfile(new_content)	×
98	new = await _parse_lockfile_content(new_content, path)	×
99	old = await _parse_lockfile(	×
100	Lockfile(
101	url=path,
102	url_description_of_origin="existing lockfile",
103	resolve_name=resolve_name,
104	)
105	)
106	return LockfileDiff.create(	×
107	path=path,
108	resolve_name=resolve_name,
109	old=_pex_lockfile_requirements(old),
110	new=_pex_lockfile_requirements(new, path),
111	)

pantsbuild / pants / 19381742489

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