mirror of
https://github.com/pypa/pip
synced 2023-12-13 21:30:23 +01:00
42359a9605
The pip-specific Path implementation has been removed, and all its usages replaced by pathlib.Path. The tmpdir and tmpdir_factory fixtures are also removed, and all usages are replaced by tmp_path and tmp_path_factory, which use pathlib.Path. The pip() function now also accepts pathlib.Path so we don't need to put str() everywhere. Path arguments are coerced with os.fspath() into str.
404 lines
13 KiB
Python
404 lines
13 KiB
Python
import os
|
|
import sys
|
|
from pathlib import Path
|
|
from typing import Iterator, List, Optional, Tuple
|
|
from unittest.mock import Mock
|
|
|
|
import pytest
|
|
|
|
import pip._internal.req.req_uninstall
|
|
from pip._internal.req.req_uninstall import (
|
|
StashedUninstallPathSet,
|
|
UninstallPathSet,
|
|
UninstallPthEntries,
|
|
compact,
|
|
compress_for_output_listing,
|
|
compress_for_rename,
|
|
uninstallation_paths,
|
|
)
|
|
from tests.lib import create_file
|
|
|
|
|
|
# Pretend all files are local, so UninstallPathSet accepts files in the tmpdir,
|
|
# outside the virtualenv
|
|
def mock_is_local(path: str) -> bool:
|
|
return True
|
|
|
|
|
|
def test_uninstallation_paths() -> None:
|
|
class dist:
|
|
def iter_declared_entries(self) -> Optional[Iterator[str]]:
|
|
yield "file.py"
|
|
yield "file.pyc"
|
|
yield "file.so"
|
|
yield "nopyc.py"
|
|
|
|
location = ""
|
|
|
|
d = dist()
|
|
|
|
paths = list(uninstallation_paths(d))
|
|
|
|
expected = [
|
|
"file.py",
|
|
"file.pyc",
|
|
"file.pyo",
|
|
"file.so",
|
|
"nopyc.py",
|
|
"nopyc.pyc",
|
|
"nopyc.pyo",
|
|
]
|
|
|
|
assert paths == expected
|
|
|
|
# Avoid an easy 'unique generator' bug
|
|
paths2 = list(uninstallation_paths(d))
|
|
|
|
assert paths2 == paths
|
|
|
|
|
|
def test_compressed_listing(tmpdir: Path) -> None:
|
|
def in_tmpdir(paths: List[str]) -> List[str]:
|
|
li = []
|
|
for path in paths:
|
|
li.append(str(os.path.join(tmpdir, path.replace("/", os.path.sep))))
|
|
return li
|
|
|
|
sample = in_tmpdir(
|
|
[
|
|
"lib/mypkg.dist-info/METADATA",
|
|
"lib/mypkg.dist-info/PKG-INFO",
|
|
"lib/mypkg/would_be_removed.txt",
|
|
"lib/mypkg/would_be_skipped.skip.txt",
|
|
"lib/mypkg/__init__.py",
|
|
"lib/mypkg/my_awesome_code.py",
|
|
"lib/mypkg/__pycache__/my_awesome_code-magic.pyc",
|
|
"lib/mypkg/support/support_file.py",
|
|
"lib/mypkg/support/more_support.py",
|
|
"lib/mypkg/support/would_be_skipped.skip.py",
|
|
"lib/mypkg/support/__pycache__/support_file-magic.pyc",
|
|
"lib/random_other_place/file_without_a_dot_pyc",
|
|
"bin/mybin",
|
|
]
|
|
)
|
|
|
|
# Create the required files
|
|
for fname in sample:
|
|
create_file(fname, "random blub")
|
|
|
|
# Remove the files to be skipped from the paths
|
|
sample = [path for path in sample if ".skip." not in path]
|
|
|
|
expected_remove = in_tmpdir(
|
|
[
|
|
"bin/mybin",
|
|
"lib/mypkg.dist-info/*",
|
|
"lib/mypkg/*",
|
|
"lib/random_other_place/file_without_a_dot_pyc",
|
|
]
|
|
)
|
|
|
|
expected_skip = in_tmpdir(
|
|
[
|
|
"lib/mypkg/would_be_skipped.skip.txt",
|
|
"lib/mypkg/support/would_be_skipped.skip.py",
|
|
]
|
|
)
|
|
|
|
expected_rename = in_tmpdir(
|
|
[
|
|
"bin/",
|
|
"lib/mypkg.dist-info/",
|
|
"lib/mypkg/would_be_removed.txt",
|
|
"lib/mypkg/__init__.py",
|
|
"lib/mypkg/my_awesome_code.py",
|
|
"lib/mypkg/__pycache__/",
|
|
"lib/mypkg/support/support_file.py",
|
|
"lib/mypkg/support/more_support.py",
|
|
"lib/mypkg/support/__pycache__/",
|
|
"lib/random_other_place/",
|
|
]
|
|
)
|
|
|
|
will_remove, will_skip = compress_for_output_listing(sample)
|
|
will_rename = compress_for_rename(sample)
|
|
assert sorted(expected_skip) == sorted(compact(will_skip))
|
|
assert sorted(expected_remove) == sorted(compact(will_remove))
|
|
assert sorted(expected_rename) == sorted(compact(will_rename))
|
|
|
|
|
|
class TestUninstallPathSet:
|
|
def test_add(self, tmpdir: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setattr(pip._internal.req.req_uninstall, "is_local", mock_is_local)
|
|
# Fix case for windows tests
|
|
file_extant = os.path.normcase(os.path.join(tmpdir, "foo"))
|
|
file_nonexistent = os.path.normcase(os.path.join(tmpdir, "nonexistent"))
|
|
with open(file_extant, "w"):
|
|
pass
|
|
|
|
ups = UninstallPathSet(dist=Mock())
|
|
assert ups._paths == set()
|
|
ups.add(file_extant)
|
|
assert ups._paths == {file_extant}
|
|
|
|
ups.add(file_nonexistent)
|
|
assert ups._paths == {file_extant}
|
|
|
|
def test_add_pth(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setattr(pip._internal.req.req_uninstall, "is_local", mock_is_local)
|
|
# Fix case for windows tests
|
|
tmpdir = os.path.normcase(tmp_path)
|
|
on_windows = sys.platform == "win32"
|
|
pth_file = os.path.join(tmpdir, "foo.pth")
|
|
relative = "../../example"
|
|
if on_windows:
|
|
share = "\\\\example\\share\\"
|
|
share_com = "\\\\example.com\\share\\"
|
|
# Create a .pth file for testing
|
|
with open(pth_file, "w") as f:
|
|
f.writelines([tmpdir, "\n", relative, "\n"])
|
|
if on_windows:
|
|
f.writelines([share, "\n", share_com, "\n"])
|
|
# Add paths to be removed
|
|
pth = UninstallPthEntries(pth_file)
|
|
pth.add(tmpdir)
|
|
pth.add(relative)
|
|
if on_windows:
|
|
pth.add(share)
|
|
pth.add(share_com)
|
|
# Check that the paths were added to entries
|
|
if on_windows:
|
|
check = {tmpdir, relative, share, share_com}
|
|
else:
|
|
check = {tmpdir, relative}
|
|
assert pth.entries == check
|
|
|
|
@pytest.mark.skipif("sys.platform == 'win32'")
|
|
def test_add_symlink(self, tmpdir: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setattr(pip._internal.req.req_uninstall, "is_local", mock_is_local)
|
|
f = os.path.join(tmpdir, "foo")
|
|
with open(f, "w"):
|
|
pass
|
|
foo_link = os.path.join(tmpdir, "foo_link")
|
|
os.symlink(f, foo_link)
|
|
|
|
ups = UninstallPathSet(dist=Mock())
|
|
ups.add(foo_link)
|
|
assert ups._paths == {foo_link}
|
|
|
|
def test_compact_shorter_path(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setattr(pip._internal.req.req_uninstall, "is_local", mock_is_local)
|
|
monkeypatch.setattr("os.path.exists", lambda p: True)
|
|
# This deals with nt/posix path differences
|
|
short_path = os.path.normcase(
|
|
os.path.abspath(os.path.join(os.path.sep, "path"))
|
|
)
|
|
ups = UninstallPathSet(dist=Mock())
|
|
ups.add(short_path)
|
|
ups.add(os.path.join(short_path, "longer"))
|
|
assert compact(ups._paths) == {short_path}
|
|
|
|
@pytest.mark.skipif("sys.platform == 'win32'")
|
|
def test_detect_symlink_dirs(
|
|
self, monkeypatch: pytest.MonkeyPatch, tmpdir: Path
|
|
) -> None:
|
|
monkeypatch.setattr(pip._internal.req.req_uninstall, "is_local", mock_is_local)
|
|
|
|
# construct 2 paths:
|
|
# tmpdir/dir/file
|
|
# tmpdir/dirlink/file (where dirlink is a link to dir)
|
|
d = tmpdir.joinpath("dir")
|
|
d.mkdir()
|
|
dlink = tmpdir.joinpath("dirlink")
|
|
os.symlink(d, dlink)
|
|
d.joinpath("file").touch()
|
|
path1 = str(d.joinpath("file"))
|
|
path2 = str(dlink.joinpath("file"))
|
|
|
|
ups = UninstallPathSet(dist=Mock())
|
|
ups.add(path1)
|
|
ups.add(path2)
|
|
assert ups._paths == {path1}
|
|
|
|
|
|
class TestStashedUninstallPathSet:
|
|
WALK_RESULT: List[Tuple[str, List[str], List[str]]] = [
|
|
("A", ["B", "C"], ["a.py"]),
|
|
("A/B", ["D"], ["b.py"]),
|
|
("A/B/D", [], ["c.py"]),
|
|
("A/C", [], ["d.py", "e.py"]),
|
|
("A/E", ["F"], ["f.py"]),
|
|
("A/E/F", [], []),
|
|
("A/G", ["H"], ["g.py"]),
|
|
("A/G/H", [], ["h.py"]),
|
|
]
|
|
|
|
@classmethod
|
|
def mock_walk(cls, root: str) -> Iterator[Tuple[str, List[str], List[str]]]:
|
|
for dirname, subdirs, files in cls.WALK_RESULT:
|
|
dirname = os.path.sep.join(dirname.split("/"))
|
|
if dirname.startswith(root):
|
|
yield dirname[len(root) + 1 :], subdirs, files
|
|
|
|
def test_compress_for_rename(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
paths = [
|
|
os.path.sep.join(p.split("/"))
|
|
for p in [
|
|
"A/B/b.py",
|
|
"A/B/D/c.py",
|
|
"A/C/d.py",
|
|
"A/E/f.py",
|
|
"A/G/g.py",
|
|
]
|
|
]
|
|
|
|
expected_paths = [
|
|
os.path.sep.join(p.split("/"))
|
|
for p in [
|
|
"A/B/", # selected everything below A/B
|
|
"A/C/d.py", # did not select everything below A/C
|
|
"A/E/", # only empty folders remain under A/E
|
|
"A/G/g.py", # non-empty folder remains under A/G
|
|
]
|
|
]
|
|
|
|
monkeypatch.setattr("os.walk", self.mock_walk)
|
|
|
|
actual_paths = compress_for_rename(paths)
|
|
assert set(expected_paths) == set(actual_paths)
|
|
|
|
@classmethod
|
|
def make_stash(
|
|
cls, tmpdir: Path, paths: List[str]
|
|
) -> Tuple[StashedUninstallPathSet, List[Tuple[str, str]]]:
|
|
for dirname, subdirs, files in cls.WALK_RESULT:
|
|
root = os.path.join(tmpdir, *dirname.split("/"))
|
|
if not os.path.exists(root):
|
|
os.mkdir(root)
|
|
for d in subdirs:
|
|
os.mkdir(os.path.join(root, d))
|
|
for f in files:
|
|
with open(os.path.join(root, f), "wb"):
|
|
pass
|
|
|
|
pathset = StashedUninstallPathSet()
|
|
|
|
paths = [os.path.join(tmpdir, *p.split("/")) for p in paths]
|
|
stashed_paths = [(p, pathset.stash(p)) for p in paths]
|
|
|
|
return pathset, stashed_paths
|
|
|
|
def test_stash(self, tmpdir: Path) -> None:
|
|
pathset, stashed_paths = self.make_stash(
|
|
tmpdir,
|
|
[
|
|
"A/B/",
|
|
"A/C/d.py",
|
|
"A/E/",
|
|
"A/G/g.py",
|
|
],
|
|
)
|
|
|
|
for old_path, new_path in stashed_paths:
|
|
assert not os.path.exists(old_path)
|
|
assert os.path.exists(new_path)
|
|
|
|
assert stashed_paths == pathset._moves
|
|
|
|
def test_commit(self, tmpdir: Path) -> None:
|
|
pathset, stashed_paths = self.make_stash(
|
|
tmpdir,
|
|
[
|
|
"A/B/",
|
|
"A/C/d.py",
|
|
"A/E/",
|
|
"A/G/g.py",
|
|
],
|
|
)
|
|
|
|
pathset.commit()
|
|
|
|
for old_path, new_path in stashed_paths:
|
|
assert not os.path.exists(old_path)
|
|
assert not os.path.exists(new_path)
|
|
|
|
def test_rollback(self, tmpdir: Path) -> None:
|
|
pathset, stashed_paths = self.make_stash(
|
|
tmpdir,
|
|
[
|
|
"A/B/",
|
|
"A/C/d.py",
|
|
"A/E/",
|
|
"A/G/g.py",
|
|
],
|
|
)
|
|
|
|
pathset.rollback()
|
|
|
|
for old_path, new_path in stashed_paths:
|
|
assert os.path.exists(old_path)
|
|
assert not os.path.exists(new_path)
|
|
|
|
@pytest.mark.skipif("sys.platform == 'win32'")
|
|
def test_commit_symlinks(self, tmpdir: Path) -> None:
|
|
adir = tmpdir / "dir"
|
|
adir.mkdir()
|
|
dirlink = tmpdir / "dirlink"
|
|
dirlink.symlink_to(adir)
|
|
afile = tmpdir / "file"
|
|
afile.write_text("...")
|
|
filelink = tmpdir / "filelink"
|
|
filelink.symlink_to(afile)
|
|
|
|
pathset = StashedUninstallPathSet()
|
|
stashed_paths = []
|
|
stashed_paths.append(pathset.stash(os.fspath(dirlink)))
|
|
stashed_paths.append(pathset.stash(os.fspath(filelink)))
|
|
for stashed_path in stashed_paths:
|
|
assert os.path.lexists(stashed_path)
|
|
assert not os.path.exists(dirlink)
|
|
assert not os.path.exists(filelink)
|
|
|
|
pathset.commit()
|
|
|
|
# stash removed, links removed
|
|
for stashed_path in stashed_paths:
|
|
assert not os.path.lexists(stashed_path)
|
|
assert not os.path.lexists(dirlink) and not os.path.isdir(dirlink)
|
|
assert not os.path.lexists(filelink) and not os.path.isfile(filelink)
|
|
|
|
# link targets untouched
|
|
assert os.path.isdir(adir)
|
|
assert os.path.isfile(afile)
|
|
|
|
@pytest.mark.skipif("sys.platform == 'win32'")
|
|
def test_rollback_symlinks(self, tmpdir: Path) -> None:
|
|
adir = tmpdir / "dir"
|
|
adir.mkdir()
|
|
dirlink = tmpdir / "dirlink"
|
|
dirlink.symlink_to(adir)
|
|
afile = tmpdir / "file"
|
|
afile.write_text("...")
|
|
filelink = tmpdir / "filelink"
|
|
filelink.symlink_to(afile)
|
|
|
|
pathset = StashedUninstallPathSet()
|
|
stashed_paths = []
|
|
stashed_paths.append(pathset.stash(os.fspath(dirlink)))
|
|
stashed_paths.append(pathset.stash(os.fspath(filelink)))
|
|
for stashed_path in stashed_paths:
|
|
assert os.path.lexists(stashed_path)
|
|
assert not os.path.lexists(dirlink)
|
|
assert not os.path.lexists(filelink)
|
|
|
|
pathset.rollback()
|
|
|
|
# stash removed, links restored
|
|
for stashed_path in stashed_paths:
|
|
assert not os.path.lexists(stashed_path)
|
|
assert os.path.lexists(dirlink) and os.path.isdir(dirlink)
|
|
assert os.path.lexists(filelink) and os.path.isfile(filelink)
|
|
|
|
# link targets untouched
|
|
assert os.path.isdir(adir)
|
|
assert os.path.isfile(afile)
|