25443604553

Committed 06 May 2026 03:05PM UTC coverage: 92.879% (-0.04%) from 92.915%

Build # 25443604553

Build Type

push

github

Committed by

web-flow

Commit Message

[pants_ng] Scaffolding for a pants_ng mode. (#23319)

In this mode the command line is parsed as an
NG invocation, and dispatched appropriately.

Of course at the moment there are no
implementations to dispatch to. That will follow.

This does expose a new option, `pants_ng` to users. 
There is a big warning not to set it, but we're not trying
to hide that we're working on a new thing, so I am
comfortable with this.

Coverage Stats

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

1294 existing lines in 76 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

100.0

/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	2✔
5
6	import json	2✔
7	import os.path	2✔
8	import zipfile	2✔
9	from collections.abc import Iterable, Iterator, Mapping	2✔
10	from dataclasses import dataclass	2✔
11	from pathlib import PurePath	2✔
12	from typing import Any	2✔
13
14	from pants.backend.python.target_types import MainSpecification, PexLayout	2✔
15	from pants.backend.python.util_rules.interpreter_constraints import InterpreterConstraints	2✔
16	from pants.backend.python.util_rules.pex import (	2✔
17	Pex,
18	PexPlatforms,
19	PexProcess,
20	PexRequest,
21	VenvPex,
22	VenvPexProcess,
23	)
24	from pants.backend.python.util_rules.pex_cli import PexPEX	2✔
25	from pants.backend.python.util_rules.pex_requirements import EntireLockfile, PexRequirements	2✔
26	from pants.engine.fs import Digest	2✔
27	from pants.engine.process import Process, ProcessResult	2✔
28	from pants.testutil.rule_runner import PYTHON_BOOTSTRAP_ENV, QueryRule, RuleRunner	2✔
29	from pants.util.pip_requirement import PipRequirement	2✔
30	from pants.util.strutil import softwrap	2✔
31
32
33	@dataclass(frozen=True)	2✔
34	class ExactRequirement:	2✔
35	name: str	2✔
36	version: str	2✔
37
38	@classmethod	2✔
39	def parse(cls, requirement: str) -> ExactRequirement:	2✔
UNCOV 40	req = PipRequirement.parse(requirement)	1✔
UNCOV 41	assert len(req.specifier_set) == 1, softwrap(	1✔
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))	1✔
UNCOV 48	assert specifier.operator == "==", softwrap(	1✔
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)	1✔
55
56
57	def parse_requirements(requirements: Iterable[str]) -> Iterator[ExactRequirement]:	2✔
UNCOV 58	for requirement in requirements:	1✔
UNCOV 59	yield ExactRequirement.parse(requirement)	1✔
60
61
62	@dataclass(frozen=True)	2✔
63	class PexData:	2✔
64	pex: Pex \| VenvPex	2✔
65	is_zipapp: bool	2✔
66	sandbox_path: PurePath	2✔
67	local_path: PurePath	2✔
68	info: Mapping[str, Any]	2✔
69	files: tuple[str, ...]	2✔
70
71
72	def get_all_data(rule_runner: RuleRunner, pex: Pex \| VenvPex) -> PexData:	2✔
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):	2✔
UNCOV 76	digest = pex.digest	1✔
UNCOV 77	sandbox_path = pex.pex_filename	1✔
78	else:
79	digest = pex.digest	2✔
80	sandbox_path = pex.name	2✔
81
82	rule_runner.scheduler.write_digest(digest)	2✔
83	local_path = PurePath(rule_runner.build_root) / sandbox_path	2✔
84
85	is_zipapp = zipfile.is_zipfile(local_path)	2✔
86	if is_zipapp:	2✔
87	with zipfile.ZipFile(local_path, "r") as zipfp:	2✔
88	files = tuple(zipfp.namelist())	2✔
89	pex_info_content = zipfp.read("PEX-INFO")	2✔
90	else:
UNCOV 91	files = tuple(	1✔
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:	1✔
UNCOV 97	pex_info_content = fp.read()	1✔
98
99	return PexData(	2✔
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(	2✔
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(	2✔
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)	2✔
138
139	pex: Pex \| VenvPex
140	if pex_type == Pex:	2✔
141	pex = rule_runner.request(Pex, [request])	2✔
142	else:
UNCOV 143	pex = rule_runner.request(VenvPex, [request])	1✔
144	return get_all_data(rule_runner, pex)	2✔
145
146
147	def create_pex_and_get_pex_info(	2✔
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(	1✔
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():	2✔
175	return [	2✔
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 / 25443604553

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