19068377358

Committed 04 Nov 2025 12:18PM UTC coverage: 92.46% (+12.2%) from 80.3%

Build # 19068377358

Build Type

Pull #22816

github

Committed by

web-flow

Commit Message

Merge a242f1805 into 89462b7ef

Pull Request Pull Request #22816: Update Pants internal Python to 3.14

Run Details

13 of 14 new or added lines in 12 files covered. (92.86%)

244 existing lines in 13 files now uncovered.

89544 of 96846 relevant lines covered (92.46%)

3.72 hits per line

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

82.63

/src/python/pants/init/plugin_resolver_test.py

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

from __future__ import annotations

import importlib.metadata
import os
import shutil
import sys
import textwrap
from collections.abc import Generator, Iterable, Sequence
from contextlib import contextmanager
from dataclasses import dataclass
from pathlib import Path, PurePath
from textwrap import dedent

import pytest
from packaging.requirements import Requirement
from packaging.utils import canonicalize_name
from packaging.version import Version

from pants.backend.python.util_rules import pex
from pants.backend.python.util_rules.interpreter_constraints import InterpreterConstraints
from pants.backend.python.util_rules.pex import Pex, PexProcess, PexRequest
from pants.backend.python.util_rules.pex_requirements import PexRequirements
from pants.core.util_rules import external_tool
from pants.engine.env_vars import CompleteEnvironmentVars
from pants.engine.fs import CreateDigest, Digest, FileContent, MergeDigests, Snapshot
from pants.engine.internals.scheduler import ExecutionError
from pants.engine.process import ProcessResult
from pants.init.options_initializer import create_bootstrap_scheduler
from pants.init.plugin_resolver import PluginResolver
from pants.option.options_bootstrapper import OptionsBootstrapper
from pants.testutil.python_interpreter_selection import (
    PY_38,
    PY_39,
    skip_unless_python38_and_python39_present,
)
from pants.testutil.rule_runner import EXECUTOR, QueryRule, RuleRunner
from pants.util.contextutil import temporary_dir
from pants.util.dirutil import safe_mkdir, safe_rmtree, touch
from pants.util.strutil import softwrap

DEFAULT_VERSION = "0.0.0"


@pytest.fixture
def rule_runner() -> RuleRunner:
    rule_runner = RuleRunner(
        rules=[
            *pex.rules(),
            *external_tool.rules(),
            QueryRule(Pex, [PexRequest]),
            QueryRule(ProcessResult, [PexProcess]),
        ]
    )
    rule_runner.set_options(
        [
            "--backend-packages=pants.backend.python",
        ],
        env_inherit={"PATH", "PYENV_ROOT", "HOME"},
    )
    return rule_runner


def _create_pex(
    rule_runner: RuleRunner,
    interpreter_constraints: InterpreterConstraints,
) -> Pex:
    request = PexRequest(
        output_filename="setup-py-runner.pex",
        internal_only=True,
        requirements=PexRequirements(["setuptools==44.0.0", "wheel==0.34.2"]),
        interpreter_constraints=interpreter_constraints,
    )
    return rule_runner.request(Pex, [request])


def _run_setup_py(
    rule_runner: RuleRunner,
    plugin: str,
    interpreter_constraints: InterpreterConstraints,
    version: str | None,
    install_requires: Sequence[str] | None,
    setup_py_args: Sequence[str],
    install_dir: str,
) -> None:
    pex_obj = _create_pex(rule_runner, interpreter_constraints)
    install_requires_str = f", install_requires={install_requires!r}" if install_requires else ""
    setup_py_file = FileContent(
        "setup.py",
        dedent(
            f"""
                from setuptools import setup

                setup(name="{plugin}", version="{version or DEFAULT_VERSION}"{install_requires_str})
            """
        ).encode(),
    )
    source_digest = rule_runner.request(
        Digest,
        [CreateDigest([setup_py_file])],
    )
    merged_digest = rule_runner.request(Digest, [MergeDigests([pex_obj.digest, source_digest])])

    process = PexProcess(
        pex=pex_obj,
        argv=("setup.py", *setup_py_args),
        input_digest=merged_digest,
        description="Run setup.py",
        output_directories=("dist/",),
    )
    result = rule_runner.request(ProcessResult, [process])
    result_snapshot = rule_runner.request(Snapshot, [result.output_digest])
    rule_runner.scheduler.write_digest(result.output_digest, path_prefix="output")
    safe_mkdir(install_dir)
    for path in result_snapshot.files:
        shutil.copy(PurePath(rule_runner.build_root, "output", path), install_dir)


@dataclass
class Plugin:
    name: str
    version: str | None = None
    install_requires: list[str] | None = None


@dataclass
class MockDistribution:
    name: str
    version: Version

    def create(self, site_packages_path: Path) -> None:
        # Create package directory
        pkg_dir = site_packages_path / self.name
        pkg_dir.mkdir(parents=True, exist_ok=True)

        # Create __init__.py
        (pkg_dir / "__init__.py").write_text(
            textwrap.dedent('''\
            f"""Mock package for testing."""
            __version__ = "{self.version}"

            def mock_function():
                return "Mock function called"
            ''')
        )

        # Create a module
        (pkg_dir / "module1.py").write_text(
            textwrap.dedent("""\
            def test_function():
            return "Test function from module1"
        """)
        )

        # Create dist-info directory (for installed package simulation)
        dist_info = site_packages_path / f"{self.name}-{self.version}.dist-info"
        dist_info.mkdir(exist_ok=True)

        # Create METADATA file
        (dist_info / "METADATA").write_text(
            textwrap.dedent(f"""\
            Metadata-Version: 2.1
            Name: {self.name}
            Version: {self.version}
            Summary: A mock package for testing
            Author: Test Author
            """)
        )

        # Create top_level.txt
        (dist_info / "top_level.txt").write_text(f"{self.name}\n")


@contextmanager
def plugin_resolution(
    rule_runner: RuleRunner,
    *,
    python_version: str | None = None,
    chroot: str | None = None,
    plugins: Sequence[Plugin] = (),
    requirements: Iterable[str] = (),
    sdist: bool = True,
    existing_distributions: Sequence[MockDistribution] = (),
    use_pypi: bool = False,
):
    @contextmanager
    def provide_chroot(existing: str | None) -> Generator[tuple[str, bool], None, None]:
        if existing:
            yield existing, False
        else:
            with temporary_dir(cleanup=False) as new_chroot:
                yield new_chroot, True

    @contextmanager
    def save_sys_path() -> Generator[list[str], None, None]:
        """Restores the previous `sys.path` once context ends."""
        orig_sys_path = sys.path
        sys.path = sys.path[:]
        try:
            yield orig_sys_path
        finally:
            sys.path = orig_sys_path

    # Default to resolving with whatever we're currently running with.
    interpreter_constraints = (
        InterpreterConstraints([f"=={python_version}.*"]) if python_version else None
    )
    artifact_interpreter_constraints = interpreter_constraints or InterpreterConstraints(
        [f"=={'.'.join(map(str, sys.version_info[:3]))}"]
    )

    with provide_chroot(chroot) as (root_dir, create_artifacts), save_sys_path() as saved_sys_path:
        env: dict[str, str] = {}
        repo_dir = os.path.join(root_dir, "repo")

        def _create_artifact(
            name: str, version: str | None, install_requires: Sequence[str] | None
        ) -> None:
            if create_artifacts:
                setup_py_args = ["sdist" if sdist else "bdist_wheel", "--dist-dir", "dist/"]
                _run_setup_py(
                    rule_runner,
                    name,
                    artifact_interpreter_constraints,
                    version,
                    install_requires,
                    setup_py_args,
                    repo_dir,
                )

        env.update(
            PANTS_PYTHON_REPOS_FIND_LINKS=f"['file://{repo_dir}/']",
            PANTS_PYTHON_RESOLVER_CACHE_TTL="1",
        )
        if not use_pypi:
            env.update(PANTS_PYTHON_REPOS_INDEXES="[]")

        if plugins:
            plugin_list = []
            for plugin in plugins:
                version = plugin.version
                plugin_list.append(f"{plugin.name}=={version}" if version else plugin.name)
                _create_artifact(plugin.name, version, plugin.install_requires)
            env["PANTS_PLUGINS"] = f"[{','.join(map(repr, plugin_list))}]"

            for requirement in tuple(requirements):
                r = Requirement(requirement)
                assert len(r.specifier) == 1, (
                    f"Expected requirement {requirement} to only have one comparison."
                )
                specs = next(iter(r.specifier))
                _create_artifact(canonicalize_name(r.name), specs.version, [])

        configpath = os.path.join(root_dir, "pants.toml")
        if create_artifacts:
            touch(configpath)

        args = [
            "pants",
            f"--pants-config-files=['{configpath}']",
        ]

        options_bootstrapper = OptionsBootstrapper.create(env=env, args=args, allow_pantsrc=False)
        complete_env = CompleteEnvironmentVars(
            {**{k: os.environ[k] for k in ["PATH", "HOME", "PYENV_ROOT"] if k in os.environ}, **env}
        )
        bootstrap_scheduler = create_bootstrap_scheduler(options_bootstrapper, EXECUTOR)
        cache_dir = options_bootstrapper.bootstrap_options.for_global_scope().named_caches_dir

        site_packages_path = Path(root_dir, "site-packages")
        expected_distribution_names: set[str] = set()
        for dist in existing_distributions:
            dist.create(site_packages_path)
            expected_distribution_names.add(dist.name)

        plugin_resolver = PluginResolver(
            bootstrap_scheduler, interpreter_constraints, inherit_existing_constraints=False
        )
        plugin_paths = plugin_resolver.resolve(options_bootstrapper, complete_env, requirements)

        for found_dist in importlib.metadata.distributions():
            if found_dist.name in expected_distribution_names:
                assert (
                    Path(os.path.realpath(cache_dir))
                    in Path(os.path.realpath(str(found_dist.locate_file("")))).parents
                )

        yield plugin_paths, root_dir, repo_dir, saved_sys_path


def test_no_plugins(rule_runner: RuleRunner) -> None:
    with plugin_resolution(rule_runner) as (plugin_paths, _, _, saved_sys_path):
        assert len(plugin_paths) == 0
        assert saved_sys_path == sys.path


@pytest.mark.parametrize("sdist", (False, True), ids=("bdist", "sdist"))
def test_plugins(rule_runner: RuleRunner, sdist: bool) -> None:
    with plugin_resolution(
        rule_runner,
        plugins=[Plugin("jake", "1.2.3"), Plugin("jane")],
        sdist=sdist,
        requirements=["lib==4.5.6"],
    ) as (
        _,
        _,
        _,
        _,
    ):

        def assert_dist_version(name: str, expected_version: str) -> None:
            dist = importlib.metadata.distribution(name)
            assert dist.version == expected_version, (
                f"Expected distribution {name} to have version {expected_version}, got {dist.version}"
            )

        assert_dist_version(name="jake", expected_version="1.2.3")
        assert_dist_version(name="jane", expected_version=DEFAULT_VERSION)


@pytest.mark.parametrize("sdist", (False, True), ids=("bdist", "sdist"))
def test_exact_requirements(rule_runner: RuleRunner, sdist: bool) -> None:
    with plugin_resolution(
        rule_runner, plugins=[Plugin("jake", "1.2.3"), Plugin("jane", "3.4.5")], sdist=sdist
    ) as results:
        plugin_paths1, chroot, repo_dir, saved_sys_path = results
        assert len(plugin_paths1) > 0

        # Kill the repo source dir and re-resolve. If the PluginResolver truly detects exact
        # requirements it should skip any resolves and load directly from the still intact
        # cache.
        safe_rmtree(repo_dir)

        with plugin_resolution(
            rule_runner, chroot=chroot, plugins=[Plugin("jake", "1.2.3"), Plugin("jane", "3.4.5")]
        ) as results2:
            plugin_paths2, _, _, _ = results2
            assert plugin_paths1 == plugin_paths2


def test_range_deps(rule_runner: RuleRunner) -> None:
    # Test that when a plugin has a range dependency, specifying a working set constrains
    # to a particular version, where otherwise we would get the highest released (2.27.1 in
    # this case).
    with plugin_resolution(
        rule_runner,
        plugins=[Plugin("jane", "3.4.5", ["requests>=2.25.1,<2.28.0"])],
        existing_distributions=[MockDistribution(name="requests", version=Version("2.26.0"))],
        # Because we're resolving real distributions, we enable access to pypi.
        use_pypi=True,
    ) as (
        _,
        _,
        _,
        _,
    ):
        dist = importlib.metadata.distribution("requests")
        assert "2.27.1" == dist.version


@skip_unless_python38_and_python39_present
@pytest.mark.parametrize("sdist", (False, True), ids=("bdist", "sdist"))
def test_exact_requirements_interpreter_change(rule_runner: RuleRunner, sdist: bool) -> None:
    with plugin_resolution(
        rule_runner,
        python_version=PY_38,
        plugins=[Plugin("jake", "1.2.3"), Plugin("jane", "3.4.5")],
        sdist=sdist,
    ) as results:
        plugin_paths_1, chroot, repo_dir, saved_sys_path = results

        safe_rmtree(repo_dir)
        with pytest.raises(ExecutionError):
            with plugin_resolution(
                rule_runner,
                python_version=PY_39,
                chroot=chroot,
                plugins=[Plugin("jake", "1.2.3"), Plugin("jane", "3.4.5")],
            ):
                pytest.fail(
                    softwrap(
                        """
                            Plugin re-resolution is expected for an incompatible interpreter and it
                            is expected to fail since we removed the dist `repo_dir` above.
                        """
                    )
                )

        # But for a compatible interpreter the exact resolve results should be re-used and load
        # directly from the still in-tact cache.
        with plugin_resolution(
            rule_runner,
            python_version=PY_38,
            chroot=chroot,
            plugins=[Plugin("jake", "1.2.3"), Plugin("jane", "3.4.5")],
        ) as results2:
            plugin_paths_2, _, _, _ = results2
            assert plugin_paths_1 == plugin_paths_2

1	# Copyright 2015 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 importlib.metadata	1✔
7	import os	1✔
8	import shutil	1✔
9	import sys	1✔
10	import textwrap	1✔
11	from collections.abc import Generator, Iterable, Sequence	1✔
12	from contextlib import contextmanager	1✔
13	from dataclasses import dataclass	1✔
14	from pathlib import Path, PurePath	1✔
15	from textwrap import dedent	1✔
16
17	import pytest	1✔
18	from packaging.requirements import Requirement	1✔
19	from packaging.utils import canonicalize_name	1✔
20	from packaging.version import Version	1✔
21
22	from pants.backend.python.util_rules import pex	1✔
23	from pants.backend.python.util_rules.interpreter_constraints import InterpreterConstraints	1✔
24	from pants.backend.python.util_rules.pex import Pex, PexProcess, PexRequest	1✔
25	from pants.backend.python.util_rules.pex_requirements import PexRequirements	1✔
26	from pants.core.util_rules import external_tool	1✔
27	from pants.engine.env_vars import CompleteEnvironmentVars	1✔
28	from pants.engine.fs import CreateDigest, Digest, FileContent, MergeDigests, Snapshot	1✔
29	from pants.engine.internals.scheduler import ExecutionError	1✔
30	from pants.engine.process import ProcessResult	1✔
31	from pants.init.options_initializer import create_bootstrap_scheduler	1✔
32	from pants.init.plugin_resolver import PluginResolver	1✔
33	from pants.option.options_bootstrapper import OptionsBootstrapper	1✔
34	from pants.testutil.python_interpreter_selection import (	1✔
35	PY_38,
36	PY_39,
37	skip_unless_python38_and_python39_present,
38	)
39	from pants.testutil.rule_runner import EXECUTOR, QueryRule, RuleRunner	1✔
40	from pants.util.contextutil import temporary_dir	1✔
41	from pants.util.dirutil import safe_mkdir, safe_rmtree, touch	1✔
42	from pants.util.strutil import softwrap	1✔
43
44	DEFAULT_VERSION = "0.0.0"	1✔
45
46
47	@pytest.fixture	1✔
48	def rule_runner() -> RuleRunner:	1✔
49	rule_runner = RuleRunner(	1✔
50	rules=[
51	*pex.rules(),
52	*external_tool.rules(),
53	QueryRule(Pex, [PexRequest]),
54	QueryRule(ProcessResult, [PexProcess]),
55	]
56	)
57	rule_runner.set_options(	1✔
58	[
59	"--backend-packages=pants.backend.python",
60	],
61	env_inherit={"PATH", "PYENV_ROOT", "HOME"},
62	)
63	return rule_runner	1✔
64
65
66	def _create_pex(	1✔
67	rule_runner: RuleRunner,
68	interpreter_constraints: InterpreterConstraints,
69	) -> Pex:
70	request = PexRequest(	1✔
71	output_filename="setup-py-runner.pex",
72	internal_only=True,
73	requirements=PexRequirements(["setuptools==44.0.0", "wheel==0.34.2"]),
74	interpreter_constraints=interpreter_constraints,
75	)
76	return rule_runner.request(Pex, [request])	1✔
77
78
79	def _run_setup_py(	1✔
80	rule_runner: RuleRunner,
81	plugin: str,
82	interpreter_constraints: InterpreterConstraints,
83	version: str \| None,
84	install_requires: Sequence[str] \| None,
85	setup_py_args: Sequence[str],
86	install_dir: str,
87	) -> None:
88	pex_obj = _create_pex(rule_runner, interpreter_constraints)	1✔
89	install_requires_str = f", install_requires={install_requires!r}" if install_requires else ""	1✔
90	setup_py_file = FileContent(	1✔
91	"setup.py",
92	dedent(
93	f"""
94	from setuptools import setup
95
96	setup(name="{plugin}", version="{version or DEFAULT_VERSION}"{install_requires_str})
97	"""
98	).encode(),
99	)
100	source_digest = rule_runner.request(	1✔
101	Digest,
102	[CreateDigest([setup_py_file])],
103	)
104	merged_digest = rule_runner.request(Digest, [MergeDigests([pex_obj.digest, source_digest])])	1✔
105
106	process = PexProcess(	1✔
107	pex=pex_obj,
108	argv=("setup.py", *setup_py_args),
109	input_digest=merged_digest,
110	description="Run setup.py",
111	output_directories=("dist/",),
112	)
113	result = rule_runner.request(ProcessResult, [process])	1✔
114	result_snapshot = rule_runner.request(Snapshot, [result.output_digest])	1✔
115	rule_runner.scheduler.write_digest(result.output_digest, path_prefix="output")	1✔
116	safe_mkdir(install_dir)	1✔
117	for path in result_snapshot.files:	1✔
118	shutil.copy(PurePath(rule_runner.build_root, "output", path), install_dir)	1✔
119
120
121	@dataclass	1✔
122	class Plugin:	1✔
123	name: str	1✔
124	version: str \| None = None	1✔
125	install_requires: list[str] \| None = None	1✔
126
127
128	@dataclass	1✔
129	class MockDistribution:	1✔
130	name: str	1✔
131	version: Version	1✔
132
133	def create(self, site_packages_path: Path) -> None:	1✔
134	# Create package directory
UNCOV 135	pkg_dir = site_packages_path / self.name	×
UNCOV 136	pkg_dir.mkdir(parents=True, exist_ok=True)	×
137
138	# Create __init__.py
UNCOV 139	(pkg_dir / "__init__.py").write_text(	×
140	textwrap.dedent('''\
141	f"""Mock package for testing."""
142	__version__ = "{self.version}"
143
144	def mock_function():
145	return "Mock function called"
146	''')
147	)
148
149	# Create a module
UNCOV 150	(pkg_dir / "module1.py").write_text(	×
151	textwrap.dedent("""\
152	def test_function():
153	return "Test function from module1"
154	""")
155	)
156
157	# Create dist-info directory (for installed package simulation)
UNCOV 158	dist_info = site_packages_path / f"{self.name}-{self.version}.dist-info"	×
UNCOV 159	dist_info.mkdir(exist_ok=True)	×
160
161	# Create METADATA file
UNCOV 162	(dist_info / "METADATA").write_text(	×
163	textwrap.dedent(f"""\
164	Metadata-Version: 2.1
165	Name: {self.name}
166	Version: {self.version}
167	Summary: A mock package for testing
168	Author: Test Author
169	""")
170	)
171
172	# Create top_level.txt
UNCOV 173	(dist_info / "top_level.txt").write_text(f"{self.name}\n")	×
174
175
176	@contextmanager	1✔
177	def plugin_resolution(	1✔
178	rule_runner: RuleRunner,
179	*,
180	python_version: str \| None = None,
181	chroot: str \| None = None,
182	plugins: Sequence[Plugin] = (),
183	requirements: Iterable[str] = (),
184	sdist: bool = True,
185	existing_distributions: Sequence[MockDistribution] = (),
186	use_pypi: bool = False,
187	):
188	@contextmanager	1✔
189	def provide_chroot(existing: str \| None) -> Generator[tuple[str, bool], None, None]:	1✔
190	if existing:	1✔
191	yield existing, False	1✔
192	else:
193	with temporary_dir(cleanup=False) as new_chroot:	1✔
194	yield new_chroot, True	1✔
195
196	@contextmanager	1✔
197	def save_sys_path() -> Generator[list[str], None, None]:	1✔
198	"""Restores the previous `sys.path` once context ends."""
199	orig_sys_path = sys.path	1✔
200	sys.path = sys.path[:]	1✔
201	try:	1✔
202	yield orig_sys_path	1✔
203	finally:
204	sys.path = orig_sys_path	1✔
205
206	# Default to resolving with whatever we're currently running with.
207	interpreter_constraints = (	1✔
208	InterpreterConstraints([f"=={python_version}.*"]) if python_version else None
209	)
210	artifact_interpreter_constraints = interpreter_constraints or InterpreterConstraints(	1✔
211	[f"=={'.'.join(map(str, sys.version_info[:3]))}"]
212	)
213
214	with provide_chroot(chroot) as (root_dir, create_artifacts), save_sys_path() as saved_sys_path:	1✔
215	env: dict[str, str] = {}	1✔
216	repo_dir = os.path.join(root_dir, "repo")	1✔
217
218	def _create_artifact(	1✔
219	name: str, version: str \| None, install_requires: Sequence[str] \| None
220	) -> None:
221	if create_artifacts:	1✔
222	setup_py_args = ["sdist" if sdist else "bdist_wheel", "--dist-dir", "dist/"]	1✔
223	_run_setup_py(	1✔
224	rule_runner,
225	name,
226	artifact_interpreter_constraints,
227	version,
228	install_requires,
229	setup_py_args,
230	repo_dir,
231	)
232
233	env.update(	1✔
234	PANTS_PYTHON_REPOS_FIND_LINKS=f"['file://{repo_dir}/']",
235	PANTS_PYTHON_RESOLVER_CACHE_TTL="1",
236	)
237	if not use_pypi:	1✔
238	env.update(PANTS_PYTHON_REPOS_INDEXES="[]")	1✔
239
240	if plugins:	1✔
241	plugin_list = []	1✔
242	for plugin in plugins:	1✔
243	version = plugin.version	1✔
244	plugin_list.append(f"{plugin.name}=={version}" if version else plugin.name)	1✔
245	_create_artifact(plugin.name, version, plugin.install_requires)	1✔
246	env["PANTS_PLUGINS"] = f"[{','.join(map(repr, plugin_list))}]"	1✔
247
248	for requirement in tuple(requirements):	1✔
UNCOV 249	r = Requirement(requirement)	×
UNCOV 250	assert len(r.specifier) == 1, (	×
251	f"Expected requirement {requirement} to only have one comparison."
252	)
UNCOV 253	specs = next(iter(r.specifier))	×
UNCOV 254	_create_artifact(canonicalize_name(r.name), specs.version, [])	×
255
256	configpath = os.path.join(root_dir, "pants.toml")	1✔
257	if create_artifacts:	1✔
258	touch(configpath)	1✔
259
260	args = [	1✔
261	"pants",
262	f"--pants-config-files=['{configpath}']",
263	]
264
265	options_bootstrapper = OptionsBootstrapper.create(env=env, args=args, allow_pantsrc=False)	1✔
266	complete_env = CompleteEnvironmentVars(	1✔
267	{{k: os.environ[k] for k in ["PATH", "HOME", "PYENV_ROOT"] if k in os.environ}, env}
268	)
269	bootstrap_scheduler = create_bootstrap_scheduler(options_bootstrapper, EXECUTOR)	1✔
270	cache_dir = options_bootstrapper.bootstrap_options.for_global_scope().named_caches_dir	1✔
271
272	site_packages_path = Path(root_dir, "site-packages")	1✔
273	expected_distribution_names: set[str] = set()	1✔
274	for dist in existing_distributions:	1✔
UNCOV 275	dist.create(site_packages_path)	×
UNCOV 276	expected_distribution_names.add(dist.name)	×
277
278	plugin_resolver = PluginResolver(	1✔
279	bootstrap_scheduler, interpreter_constraints, inherit_existing_constraints=False
280	)
281	plugin_paths = plugin_resolver.resolve(options_bootstrapper, complete_env, requirements)	1✔
282
283	for found_dist in importlib.metadata.distributions():	1✔
284	if found_dist.name in expected_distribution_names:	1✔
UNCOV 285	assert (	×
286	Path(os.path.realpath(cache_dir))
287	in Path(os.path.realpath(str(found_dist.locate_file("")))).parents
288	)
289
290	yield plugin_paths, root_dir, repo_dir, saved_sys_path	1✔
291
292
293	def test_no_plugins(rule_runner: RuleRunner) -> None:	1✔
294	with plugin_resolution(rule_runner) as (plugin_paths, _, _, saved_sys_path):	1✔
295	assert len(plugin_paths) == 0	1✔
296	assert saved_sys_path == sys.path	1✔
297
298
299	@pytest.mark.parametrize("sdist", (False, True), ids=("bdist", "sdist"))	1✔
300	def test_plugins(rule_runner: RuleRunner, sdist: bool) -> None:	1✔
301	with plugin_resolution(	1✔
302	rule_runner,
303	plugins=[Plugin("jake", "1.2.3"), Plugin("jane")],
304	sdist=sdist,
305	requirements=["lib==4.5.6"],
306	) as (
307	_,
308	_,
309	_,
310	_,
311	):
312
UNCOV 313	def assert_dist_version(name: str, expected_version: str) -> None:	×
UNCOV 314	dist = importlib.metadata.distribution(name)	×
UNCOV 315	assert dist.version == expected_version, (	×
316	f"Expected distribution {name} to have version {expected_version}, got {dist.version}"
317	)
318
UNCOV 319	assert_dist_version(name="jake", expected_version="1.2.3")	×
UNCOV 320	assert_dist_version(name="jane", expected_version=DEFAULT_VERSION)	×
321
322
323	@pytest.mark.parametrize("sdist", (False, True), ids=("bdist", "sdist"))	1✔
324	def test_exact_requirements(rule_runner: RuleRunner, sdist: bool) -> None:	1✔
325	with plugin_resolution(	1✔
326	rule_runner, plugins=[Plugin("jake", "1.2.3"), Plugin("jane", "3.4.5")], sdist=sdist
327	) as results:
UNCOV 328	plugin_paths1, chroot, repo_dir, saved_sys_path = results	×
UNCOV 329	assert len(plugin_paths1) > 0	×
330
331	# Kill the repo source dir and re-resolve. If the PluginResolver truly detects exact
332	# requirements it should skip any resolves and load directly from the still intact
333	# cache.
UNCOV 334	safe_rmtree(repo_dir)	×
335
UNCOV 336	with plugin_resolution(	×
337	rule_runner, chroot=chroot, plugins=[Plugin("jake", "1.2.3"), Plugin("jane", "3.4.5")]
338	) as results2:
UNCOV 339	plugin_paths2, _, _, _ = results2	×
UNCOV 340	assert plugin_paths1 == plugin_paths2	×
341
342
343	def test_range_deps(rule_runner: RuleRunner) -> None:	1✔
344	# Test that when a plugin has a range dependency, specifying a working set constrains
345	# to a particular version, where otherwise we would get the highest released (2.27.1 in
346	# this case).
347	with plugin_resolution(	1✔
348	rule_runner,
349	plugins=[Plugin("jane", "3.4.5", ["requests>=2.25.1,<2.28.0"])],
350	existing_distributions=[MockDistribution(name="requests", version=Version("2.26.0"))],
351	# Because we're resolving real distributions, we enable access to pypi.
352	use_pypi=True,
353	) as (
354	_,
355	_,
356	_,
357	_,
358	):
UNCOV 359	dist = importlib.metadata.distribution("requests")	×
UNCOV 360	assert "2.27.1" == dist.version	×
361
362
363	@skip_unless_python38_and_python39_present	1✔
364	@pytest.mark.parametrize("sdist", (False, True), ids=("bdist", "sdist"))	1✔
365	def test_exact_requirements_interpreter_change(rule_runner: RuleRunner, sdist: bool) -> None:	1✔
366	with plugin_resolution(	1✔
367	rule_runner,
368	python_version=PY_38,
369	plugins=[Plugin("jake", "1.2.3"), Plugin("jane", "3.4.5")],
370	sdist=sdist,
371	) as results:
372	plugin_paths_1, chroot, repo_dir, saved_sys_path = results	1✔
373
374	safe_rmtree(repo_dir)	1✔
375	with pytest.raises(ExecutionError):	1✔
376	with plugin_resolution(	1✔
377	rule_runner,
378	python_version=PY_39,
379	chroot=chroot,
380	plugins=[Plugin("jake", "1.2.3"), Plugin("jane", "3.4.5")],
381	):
382	pytest.fail(	×
383	softwrap(
384	"""
385	Plugin re-resolution is expected for an incompatible interpreter and it
386	is expected to fail since we removed the dist `repo_dir` above.
387	"""
388	)
389	)
390
391	# But for a compatible interpreter the exact resolve results should be re-used and load
392	# directly from the still in-tact cache.
393	with plugin_resolution(	1✔
394	rule_runner,
395	python_version=PY_38,
396	chroot=chroot,
397	plugins=[Plugin("jake", "1.2.3"), Plugin("jane", "3.4.5")],
398	) as results2:
399	plugin_paths_2, _, _, _ = results2	1✔
400	assert plugin_paths_1 == plugin_paths_2	1✔

pantsbuild / pants / 19068377358

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