20332790708

Committed 18 Dec 2025 09:48AM UTC coverage: 64.992% (-15.3%) from 80.295%

Build # 20332790708

Build Type

Pull #22949

github

Committed by

web-flow

Commit Message

Merge f730a56cd into 407284c67

Pull Request Pull Request #22949: Add experimental uv resolver for Python lockfiles

Run Details

54 of 97 new or added lines in 5 files covered. (55.67%)

8270 existing lines in 295 files now uncovered.

48990 of 75379 relevant lines covered (64.99%)

1.81 hits per line

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

79.41

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

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

from __future__ import annotations

import json
import os.path
import zipfile
from collections.abc import Iterable, Iterator, Mapping
from dataclasses import dataclass
from pathlib import PurePath
from typing import Any

from pants.backend.python.target_types import MainSpecification, PexLayout
from pants.backend.python.util_rules.interpreter_constraints import InterpreterConstraints
from pants.backend.python.util_rules.pex import (
    Pex,
    PexPlatforms,
    PexProcess,
    PexRequest,
    VenvPex,
    VenvPexProcess,
)
from pants.backend.python.util_rules.pex_cli import PexPEX
from pants.backend.python.util_rules.pex_requirements import EntireLockfile, PexRequirements
from pants.engine.fs import Digest
from pants.engine.process import Process, ProcessResult
from pants.testutil.rule_runner import PYTHON_BOOTSTRAP_ENV, QueryRule, RuleRunner
from pants.util.pip_requirement import PipRequirement
from pants.util.strutil import softwrap


@dataclass(frozen=True)
class ExactRequirement:
    name: str
    version: str

    @classmethod
    def parse(cls, requirement: str) -> ExactRequirement:
        req = PipRequirement.parse(requirement)
        assert len(req.specifier_set) == 1, softwrap(
            f"""
            Expected an exact requirement with only 1 specifier, given {requirement} with
            {len(req.specifier_set)} specifiers
            """
        )
        specifier = next(iter(req.specifier_set))
        assert specifier.operator == "==", softwrap(
            f"""
            Expected an exact requirement using only the '==' specifier, given {requirement}
            using the {specifier.operator!r} operator
            """
        )
        return cls(name=req.name, version=specifier.version)


def parse_requirements(requirements: Iterable[str]) -> Iterator[ExactRequirement]:
    for requirement in requirements:
        yield ExactRequirement.parse(requirement)


@dataclass(frozen=True)
class PexData:
    pex: Pex | VenvPex
    is_zipapp: bool
    sandbox_path: PurePath
    local_path: PurePath
    info: Mapping[str, Any]
    files: tuple[str, ...]


def get_all_data(rule_runner: RuleRunner, pex: Pex | VenvPex) -> PexData:
    # We fish PEX-INFO out of the pex manually rather than running PEX_TOOLS, as
    # we don't know if the pex can run on the current system.
    if isinstance(pex, VenvPex):
        digest = pex.digest
        sandbox_path = pex.pex_filename
    else:
        digest = pex.digest
        sandbox_path = pex.name

    rule_runner.scheduler.write_digest(digest)
    local_path = PurePath(rule_runner.build_root) / sandbox_path

    is_zipapp = zipfile.is_zipfile(local_path)
    if is_zipapp:
        with zipfile.ZipFile(local_path, "r") as zipfp:
            files = tuple(zipfp.namelist())
            pex_info_content = zipfp.read("PEX-INFO")
    else:
        files = tuple(
            os.path.normpath(os.path.relpath(os.path.join(root, path), local_path))
            for root, dirs, files in os.walk(local_path)
            for path in dirs + files
        )
        with open(os.path.join(local_path, "PEX-INFO"), "rb") as fp:
            pex_info_content = fp.read()

    return PexData(
        pex=pex,
        is_zipapp=is_zipapp,
        sandbox_path=PurePath(sandbox_path),
        local_path=local_path,
        info=json.loads(pex_info_content.decode()),
        files=files,
    )


def create_pex_and_get_all_data(
    rule_runner: RuleRunner,
    *,
    pex_type: type[Pex | VenvPex] = Pex,
    requirements: PexRequirements | EntireLockfile = PexRequirements(),
    main: MainSpecification | None = None,
    interpreter_constraints: InterpreterConstraints = InterpreterConstraints(),
    platforms: PexPlatforms = PexPlatforms(),
    sources: Digest | None = None,
    additional_inputs: Digest | None = None,
    additional_pants_args: tuple[str, ...] = (),
    additional_pex_args: tuple[str, ...] = (),
    env: Mapping[str, str] | None = None,
    internal_only: bool = True,
    layout: PexLayout | None = None,
) -> PexData:
    request = PexRequest(
        output_filename="test.pex",
        internal_only=internal_only,
        requirements=requirements,
        interpreter_constraints=interpreter_constraints,
        platforms=platforms,
        main=main,
        sources=sources,
        additional_inputs=additional_inputs,
        additional_args=additional_pex_args,
        layout=layout,
    )
    rule_runner.set_options(additional_pants_args, env=env, env_inherit=PYTHON_BOOTSTRAP_ENV)

    pex: Pex | VenvPex
    if pex_type == Pex:
        pex = rule_runner.request(Pex, [request])
    else:
        pex = rule_runner.request(VenvPex, [request])
    return get_all_data(rule_runner, pex)


def create_pex_and_get_pex_info(
    rule_runner: RuleRunner,
    *,
    pex_type: type[Pex | VenvPex] = Pex,
    requirements: PexRequirements | EntireLockfile = PexRequirements(),
    main: MainSpecification | None = None,
    interpreter_constraints: InterpreterConstraints = InterpreterConstraints(),
    platforms: PexPlatforms = PexPlatforms(),
    sources: Digest | None = None,
    additional_pants_args: tuple[str, ...] = (),
    additional_pex_args: tuple[str, ...] = (),
    internal_only: bool = True,
) -> Mapping[str, Any]:
    return create_pex_and_get_all_data(
        rule_runner,
        pex_type=pex_type,
        requirements=requirements,
        main=main,
        interpreter_constraints=interpreter_constraints,
        platforms=platforms,
        sources=sources,
        additional_pants_args=additional_pants_args,
        additional_pex_args=additional_pex_args,
        internal_only=internal_only,
    ).info


def rules():
    return [
        QueryRule(PexPEX, ()),
        QueryRule(Pex, (PexRequest,)),
        QueryRule(VenvPex, (PexRequest,)),
        QueryRule(Process, (PexProcess,)),
        QueryRule(Process, (VenvPexProcess,)),
        QueryRule(ProcessResult, (Process,)),
    ]

1	# Copyright 2019 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 json	1✔
7	import os.path	1✔
8	import zipfile	1✔
9	from collections.abc import Iterable, Iterator, Mapping	1✔
10	from dataclasses import dataclass	1✔
11	from pathlib import PurePath	1✔
12	from typing import Any	1✔
13
14	from pants.backend.python.target_types import MainSpecification, PexLayout	1✔
15	from pants.backend.python.util_rules.interpreter_constraints import InterpreterConstraints	1✔
16	from pants.backend.python.util_rules.pex import (	1✔
17	Pex,
18	PexPlatforms,
19	PexProcess,
20	PexRequest,
21	VenvPex,
22	VenvPexProcess,
23	)
24	from pants.backend.python.util_rules.pex_cli import PexPEX	1✔
25	from pants.backend.python.util_rules.pex_requirements import EntireLockfile, PexRequirements	1✔
26	from pants.engine.fs import Digest	1✔
27	from pants.engine.process import Process, ProcessResult	1✔
28	from pants.testutil.rule_runner import PYTHON_BOOTSTRAP_ENV, QueryRule, RuleRunner	1✔
29	from pants.util.pip_requirement import PipRequirement	1✔
30	from pants.util.strutil import softwrap	1✔
31
32
33	@dataclass(frozen=True)	1✔
34	class ExactRequirement:	1✔
35	name: str	1✔
36	version: str	1✔
37
38	@classmethod	1✔
39	def parse(cls, requirement: str) -> ExactRequirement:	1✔
UNCOV 40	req = PipRequirement.parse(requirement)	×
UNCOV 41	assert len(req.specifier_set) == 1, softwrap(	×
42	f"""
43	Expected an exact requirement with only 1 specifier, given {requirement} with
44	{len(req.specifier_set)} specifiers
45	"""
46	)
UNCOV 47	specifier = next(iter(req.specifier_set))	×
UNCOV 48	assert specifier.operator == "==", softwrap(	×
49	f"""
50	Expected an exact requirement using only the '==' specifier, given {requirement}
51	using the {specifier.operator!r} operator
52	"""
53	)
UNCOV 54	return cls(name=req.name, version=specifier.version)	×
55
56
57	def parse_requirements(requirements: Iterable[str]) -> Iterator[ExactRequirement]:	1✔
UNCOV 58	for requirement in requirements:	×
UNCOV 59	yield ExactRequirement.parse(requirement)	×
60
61
62	@dataclass(frozen=True)	1✔
63	class PexData:	1✔
64	pex: Pex \| VenvPex	1✔
65	is_zipapp: bool	1✔
66	sandbox_path: PurePath	1✔
67	local_path: PurePath	1✔
68	info: Mapping[str, Any]	1✔
69	files: tuple[str, ...]	1✔
70
71
72	def get_all_data(rule_runner: RuleRunner, pex: Pex \| VenvPex) -> PexData:	1✔
73	# We fish PEX-INFO out of the pex manually rather than running PEX_TOOLS, as
74	# we don't know if the pex can run on the current system.
75	if isinstance(pex, VenvPex):	1✔
UNCOV 76	digest = pex.digest	×
UNCOV 77	sandbox_path = pex.pex_filename	×
78	else:
79	digest = pex.digest	1✔
80	sandbox_path = pex.name	1✔
81
82	rule_runner.scheduler.write_digest(digest)	1✔
83	local_path = PurePath(rule_runner.build_root) / sandbox_path	1✔
84
85	is_zipapp = zipfile.is_zipfile(local_path)	1✔
86	if is_zipapp:	1✔
87	with zipfile.ZipFile(local_path, "r") as zipfp:	1✔
88	files = tuple(zipfp.namelist())	1✔
89	pex_info_content = zipfp.read("PEX-INFO")	1✔
90	else:
UNCOV 91	files = tuple(	×
92	os.path.normpath(os.path.relpath(os.path.join(root, path), local_path))
93	for root, dirs, files in os.walk(local_path)
94	for path in dirs + files
95	)
UNCOV 96	with open(os.path.join(local_path, "PEX-INFO"), "rb") as fp:	×
UNCOV 97	pex_info_content = fp.read()	×
98
99	return PexData(	1✔
100	pex=pex,
101	is_zipapp=is_zipapp,
102	sandbox_path=PurePath(sandbox_path),
103	local_path=local_path,
104	info=json.loads(pex_info_content.decode()),
105	files=files,
106	)
107
108
109	def create_pex_and_get_all_data(	1✔
110	rule_runner: RuleRunner,
111	*,
112	pex_type: type[Pex \| VenvPex] = Pex,
113	requirements: PexRequirements \| EntireLockfile = PexRequirements(),
114	main: MainSpecification \| None = None,
115	interpreter_constraints: InterpreterConstraints = InterpreterConstraints(),
116	platforms: PexPlatforms = PexPlatforms(),
117	sources: Digest \| None = None,
118	additional_inputs: Digest \| None = None,
119	additional_pants_args: tuple[str, ...] = (),
120	additional_pex_args: tuple[str, ...] = (),
121	env: Mapping[str, str] \| None = None,
122	internal_only: bool = True,
123	layout: PexLayout \| None = None,
124	) -> PexData:
125	request = PexRequest(	1✔
126	output_filename="test.pex",
127	internal_only=internal_only,
128	requirements=requirements,
129	interpreter_constraints=interpreter_constraints,
130	platforms=platforms,
131	main=main,
132	sources=sources,
133	additional_inputs=additional_inputs,
134	additional_args=additional_pex_args,
135	layout=layout,
136	)
137	rule_runner.set_options(additional_pants_args, env=env, env_inherit=PYTHON_BOOTSTRAP_ENV)	1✔
138
139	pex: Pex \| VenvPex
140	if pex_type == Pex:	1✔
141	pex = rule_runner.request(Pex, [request])	1✔
142	else:
UNCOV 143	pex = rule_runner.request(VenvPex, [request])	×
144	return get_all_data(rule_runner, pex)	1✔
145
146
147	def create_pex_and_get_pex_info(	1✔
148	rule_runner: RuleRunner,
149	*,
150	pex_type: type[Pex \| VenvPex] = Pex,
151	requirements: PexRequirements \| EntireLockfile = PexRequirements(),
152	main: MainSpecification \| None = None,
153	interpreter_constraints: InterpreterConstraints = InterpreterConstraints(),
154	platforms: PexPlatforms = PexPlatforms(),
155	sources: Digest \| None = None,
156	additional_pants_args: tuple[str, ...] = (),
157	additional_pex_args: tuple[str, ...] = (),
158	internal_only: bool = True,
159	) -> Mapping[str, Any]:
UNCOV 160	return create_pex_and_get_all_data(	×
161	rule_runner,
162	pex_type=pex_type,
163	requirements=requirements,
164	main=main,
165	interpreter_constraints=interpreter_constraints,
166	platforms=platforms,
167	sources=sources,
168	additional_pants_args=additional_pants_args,
169	additional_pex_args=additional_pex_args,
170	internal_only=internal_only,
171	).info
172
173
174	def rules():	1✔
175	return [	1✔
176	QueryRule(PexPEX, ()),
177	QueryRule(Pex, (PexRequest,)),
178	QueryRule(VenvPex, (PexRequest,)),
179	QueryRule(Process, (PexProcess,)),
180	QueryRule(Process, (VenvPexProcess,)),
181	QueryRule(ProcessResult, (Process,)),
182	]

pantsbuild / pants / 20332790708

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