1
0
mirror of https://github.com/arsenetar/dupeguru.git synced 2026-01-22 14:41:39 +00:00

Squashed commit of the following:

commit 8b15fe9a502ebf4841c6529e7098cef03a6a5e6f
Author: Andrew Senetar <arsenetar@gmail.com>
Date:   Sun Mar 27 23:48:15 2022 -0500

    Finish up changes to copy_or_move

commit 21f6a32cf3186a400af8f30e67ad2743dc9a49bd
Author: Andrew Senetar <arsenetar@gmail.com>
Date:   Thu Mar 17 23:56:52 2022 -0500

    Migrate from hscommon.path to pathlib
    - Part one, this gets all hscommon and core tests passing
    - App appears to be able to load directories and complete scans, need further testing
    - app.py copy_or_move needs some additional work
This commit is contained in:
2022-03-27 23:50:03 -05:00
parent 5ed5eddde6
commit da9f8b2b9d
17 changed files with 267 additions and 731 deletions

View File

@@ -14,7 +14,7 @@ import re
import os
import shutil
from .path import Path, pathify
from pathlib import Path
# This matches [123], but not [12] (3 digits being the minimum).
# It also matches [1234] [12345] etc..
@@ -52,16 +52,15 @@ def is_conflicted(name):
return re_conflict.match(name) is not None
@pathify
def _smart_move_or_copy(operation, source_path: Path, dest_path: Path):
"""Use move() or copy() to move and copy file with the conflict management."""
if dest_path.isdir() and not source_path.isdir():
dest_path = dest_path[source_path.name]
if dest_path.is_dir() and not source_path.is_dir():
dest_path = dest_path.joinpath(source_path.name)
if dest_path.exists():
filename = dest_path.name
dest_dir_path = dest_path.parent()
dest_dir_path = dest_path.parent
newname = get_conflicted_name(os.listdir(str(dest_dir_path)), filename)
dest_path = dest_dir_path[newname]
dest_path = dest_dir_path.joinpath(newname)
operation(str(source_path), str(dest_path))

View File

@@ -7,208 +7,9 @@
# http://www.gnu.org/licenses/gpl-3.0.html
import logging
import os
import os.path as op
import shutil
import sys
from itertools import takewhile
from functools import wraps
from inspect import signature
class Path(tuple):
"""A handy class to work with paths.
We subclass ``tuple``, each element of the tuple represents an element of the path.
* ``Path('/foo/bar/baz')[1]`` --> ``'bar'``
* ``Path('/foo/bar/baz')[1:2]`` --> ``Path('bar/baz')``
* ``Path('/foo/bar')['baz']`` --> ``Path('/foo/bar/baz')``
* ``str(Path('/foo/bar/baz'))`` --> ``'/foo/bar/baz'``
"""
# Saves a little bit of memory usage
__slots__ = ()
def __new__(cls, value, separator=None):
def unicode_if_needed(s):
if isinstance(s, str):
return s
else:
try:
return str(s, sys.getfilesystemencoding())
except UnicodeDecodeError:
logging.warning("Could not decode %r", s)
raise
if isinstance(value, Path):
return value
if not separator:
separator = os.sep
if isinstance(value, bytes):
value = unicode_if_needed(value)
if isinstance(value, str):
if value:
if (separator not in value) and ("/" in value):
separator = "/"
value = value.split(separator)
else:
value = ()
else:
if any(isinstance(x, bytes) for x in value):
value = [unicode_if_needed(x) for x in value]
# value is a tuple/list
if any(separator in x for x in value):
# We have a component with a separator in it. Let's rejoin it, and generate another path.
return Path(separator.join(value), separator)
if (len(value) > 1) and (not value[-1]):
value = value[
:-1
] # We never want a path to end with a '' (because Path() can be called with a trailing slash ending path)
return tuple.__new__(cls, value)
def __add__(self, other):
other = Path(other)
if other and (not other[0]):
other = other[1:]
return Path(tuple.__add__(self, other))
def __contains__(self, item):
if isinstance(item, Path):
return item[: len(self)] == self
else:
return tuple.__contains__(self, item)
def __eq__(self, other):
return tuple.__eq__(self, Path(other))
def __getitem__(self, key):
if isinstance(key, slice):
if isinstance(key.start, Path):
equal_elems = list(takewhile(lambda pair: pair[0] == pair[1], zip(self, key.start)))
key = slice(len(equal_elems), key.stop, key.step)
if isinstance(key.stop, Path):
equal_elems = list(
takewhile(
lambda pair: pair[0] == pair[1],
zip(reversed(self), reversed(key.stop)),
)
)
stop = -len(equal_elems) if equal_elems else None
key = slice(key.start, stop, key.step)
return Path(tuple.__getitem__(self, key))
elif isinstance(key, (str, Path)):
return self + key
else:
return tuple.__getitem__(self, key)
def __hash__(self):
return tuple.__hash__(self)
def __ne__(self, other):
return not self.__eq__(other)
def __radd__(self, other):
return Path(other) + self
def __str__(self):
if len(self) == 1:
first = self[0]
if (len(first) == 2) and (first[1] == ":"): # Windows drive letter
return first + "\\"
elif not len(first): # root directory
return "/"
return os.sep.join(self)
def has_drive_letter(self):
if not self:
return False
first = self[0]
return (len(first) == 2) and (first[1] == ":")
def is_parent_of(self, other):
"""Whether ``other`` is a subpath of ``self``.
Almost the same as ``other in self``, but it's a bit more self-explicative and when
``other == self``, returns False.
"""
if other == self:
return False
else:
return other in self
def remove_drive_letter(self):
if self.has_drive_letter():
return self[1:]
else:
return self
def tobytes(self):
return str(self).encode(sys.getfilesystemencoding())
def parent(self):
"""Returns the parent path.
``Path('/foo/bar/baz').parent()`` --> ``Path('/foo/bar')``
"""
return self[:-1]
@property
def name(self):
"""Last element of the path (filename), with extension.
``Path('/foo/bar/baz').name`` --> ``'baz'``
"""
return self[-1]
# OS method wrappers
def exists(self):
return op.exists(str(self))
def copy(self, dest_path):
return shutil.copy(str(self), str(dest_path))
def copytree(self, dest_path, *args, **kwargs):
return shutil.copytree(str(self), str(dest_path), *args, **kwargs)
def isdir(self):
return op.isdir(str(self))
def isfile(self):
return op.isfile(str(self))
def islink(self):
return op.islink(str(self))
def listdir(self):
return [self[name] for name in os.listdir(str(self))]
def mkdir(self, *args, **kwargs):
return os.mkdir(str(self), *args, **kwargs)
def makedirs(self, *args, **kwargs):
return os.makedirs(str(self), *args, **kwargs)
def move(self, dest_path):
return shutil.move(str(self), str(dest_path))
def open(self, *args, **kwargs):
return open(str(self), *args, **kwargs)
def remove(self):
return os.remove(str(self))
def rename(self, dest_path):
return os.rename(str(self), str(dest_path))
def rmdir(self):
return os.rmdir(str(self))
def rmtree(self):
return shutil.rmtree(str(self))
def stat(self):
return os.stat(str(self))
from pathlib import Path
def pathify(f):

View File

@@ -15,7 +15,7 @@ from ..conflict import (
smart_copy,
smart_move,
)
from ..path import Path
from pathlib import Path
from ..testutil import eq_
@@ -71,43 +71,43 @@ class TestCaseMoveCopy:
def do_setup(self, request):
tmpdir = request.getfixturevalue("tmpdir")
self.path = Path(str(tmpdir))
self.path["foo"].open("w").close()
self.path["bar"].open("w").close()
self.path["dir"].mkdir()
self.path.joinpath("foo").touch()
self.path.joinpath("bar").touch()
self.path.joinpath("dir").mkdir()
def test_move_no_conflict(self, do_setup):
smart_move(self.path + "foo", self.path + "baz")
assert self.path["baz"].exists()
assert not self.path["foo"].exists()
smart_move(self.path.joinpath("foo"), self.path.joinpath("baz"))
assert self.path.joinpath("baz").exists()
assert not self.path.joinpath("foo").exists()
def test_copy_no_conflict(self, do_setup): # No need to duplicate the rest of the tests... Let's just test on move
smart_copy(self.path + "foo", self.path + "baz")
assert self.path["baz"].exists()
assert self.path["foo"].exists()
smart_copy(self.path.joinpath("foo"), self.path.joinpath("baz"))
assert self.path.joinpath("baz").exists()
assert self.path.joinpath("foo").exists()
def test_move_no_conflict_dest_is_dir(self, do_setup):
smart_move(self.path + "foo", self.path + "dir")
assert self.path["dir"]["foo"].exists()
assert not self.path["foo"].exists()
smart_move(self.path.joinpath("foo"), self.path.joinpath("dir"))
assert self.path.joinpath("dir", "foo").exists()
assert not self.path.joinpath("foo").exists()
def test_move_conflict(self, do_setup):
smart_move(self.path + "foo", self.path + "bar")
assert self.path["[000] bar"].exists()
assert not self.path["foo"].exists()
smart_move(self.path.joinpath("foo"), self.path.joinpath("bar"))
assert self.path.joinpath("[000] bar").exists()
assert not self.path.joinpath("foo").exists()
def test_move_conflict_dest_is_dir(self, do_setup):
smart_move(self.path["foo"], self.path["dir"])
smart_move(self.path["bar"], self.path["foo"])
smart_move(self.path["foo"], self.path["dir"])
assert self.path["dir"]["foo"].exists()
assert self.path["dir"]["[000] foo"].exists()
assert not self.path["foo"].exists()
assert not self.path["bar"].exists()
smart_move(self.path.joinpath("foo"), self.path.joinpath("dir"))
smart_move(self.path.joinpath("bar"), self.path.joinpath("foo"))
smart_move(self.path.joinpath("foo"), self.path.joinpath("dir"))
assert self.path.joinpath("dir", "foo").exists()
assert self.path.joinpath("dir", "[000] foo").exists()
assert not self.path.joinpath("foo").exists()
assert not self.path.joinpath("bar").exists()
def test_copy_folder(self, tmpdir):
# smart_copy also works on folders
path = Path(str(tmpdir))
path["foo"].mkdir()
path["bar"].mkdir()
smart_copy(path["foo"], path["bar"]) # no crash
assert path["[000] bar"].exists()
path.joinpath("foo").mkdir()
path.joinpath("bar").mkdir()
smart_copy(path.joinpath("foo"), path.joinpath("bar")) # no crash
assert path.joinpath("[000] bar").exists()

View File

@@ -6,261 +6,8 @@
# which should be included with this package. The terms are also available at
# http://www.gnu.org/licenses/gpl-3.0.html
import sys
import os
import pytest
from ..path import Path, pathify
from ..testutil import eq_
@pytest.fixture
def force_ossep(request):
monkeypatch = request.getfixturevalue("monkeypatch")
monkeypatch.setattr(os, "sep", "/")
def test_empty(force_ossep):
path = Path("")
eq_("", str(path))
eq_(0, len(path))
path = Path(())
eq_("", str(path))
eq_(0, len(path))
def test_single(force_ossep):
path = Path("foobar")
eq_("foobar", path)
eq_(1, len(path))
def test_multiple(force_ossep):
path = Path("foo/bar")
eq_("foo/bar", path)
eq_(2, len(path))
def test_init_with_tuple_and_list(force_ossep):
path = Path(("foo", "bar"))
eq_("foo/bar", path)
path = Path(["foo", "bar"])
eq_("foo/bar", path)
def test_init_with_invalid_value(force_ossep):
try:
Path(42)
assert False
except TypeError:
pass
def test_access(force_ossep):
path = Path("foo/bar/bleh")
eq_("foo", path[0])
eq_("foo", path[-3])
eq_("bar", path[1])
eq_("bar", path[-2])
eq_("bleh", path[2])
eq_("bleh", path[-1])
def test_slicing(force_ossep):
path = Path("foo/bar/bleh")
subpath = path[:2]
eq_("foo/bar", subpath)
assert isinstance(subpath, Path)
def test_parent(force_ossep):
path = Path("foo/bar/bleh")
subpath = path.parent()
eq_("foo/bar", subpath)
assert isinstance(subpath, Path)
def test_filename(force_ossep):
path = Path("foo/bar/bleh.ext")
eq_(path.name, "bleh.ext")
def test_deal_with_empty_components(force_ossep):
"""Keep ONLY a leading space, which means we want a leading slash."""
eq_("foo//bar", str(Path(("foo", "", "bar"))))
eq_("/foo/bar", str(Path(("", "foo", "bar"))))
eq_("foo/bar", str(Path("foo/bar/")))
def test_old_compare_paths(force_ossep):
eq_(Path("foobar"), Path("foobar"))
eq_(Path("foobar/"), Path("foobar\\", "\\"))
eq_(Path("/foobar/"), Path("\\foobar\\", "\\"))
eq_(Path("/foo/bar"), Path("\\foo\\bar", "\\"))
eq_(Path("/foo/bar"), Path("\\foo\\bar\\", "\\"))
assert Path("/foo/bar") != Path("\\foo\\foo", "\\")
# We also have to test __ne__
assert not (Path("foobar") != Path("foobar"))
assert Path("/a/b/c.x") != Path("/a/b/c.y")
def test_old_split_path(force_ossep):
eq_(Path("foobar"), ("foobar",))
eq_(Path("foo/bar"), ("foo", "bar"))
eq_(Path("/foo/bar/"), ("", "foo", "bar"))
eq_(Path("\\foo\\bar", "\\"), ("", "foo", "bar"))
def test_representation(force_ossep):
eq_("('foo', 'bar')", repr(Path(("foo", "bar"))))
def test_add(force_ossep):
eq_("foo/bar/bar/foo", Path(("foo", "bar")) + Path("bar/foo"))
eq_("foo/bar/bar/foo", Path("foo/bar") + "bar/foo")
eq_("foo/bar/bar/foo", Path("foo/bar") + ("bar", "foo"))
eq_("foo/bar/bar/foo", ("foo", "bar") + Path("bar/foo"))
eq_("foo/bar/bar/foo", "foo/bar" + Path("bar/foo"))
# Invalid concatenation
try:
Path(("foo", "bar")) + 1
assert False
except TypeError:
pass
def test_path_slice(force_ossep):
foo = Path("foo")
bar = Path("bar")
foobar = Path("foo/bar")
eq_("bar", foobar[foo:])
eq_("foo", foobar[:bar])
eq_("foo/bar", foobar[bar:])
eq_("foo/bar", foobar[:foo])
eq_((), foobar[foobar:])
eq_((), foobar[:foobar])
abcd = Path("a/b/c/d")
a = Path("a")
d = Path("d")
z = Path("z")
eq_("b/c", abcd[a:d])
eq_("b/c/d", abcd[a : d + z])
eq_("b/c", abcd[a : z + d])
eq_("a/b/c/d", abcd[:z])
def test_add_with_root_path(force_ossep):
"""if I perform /a/b/c + /d/e/f, I want /a/b/c/d/e/f, not /a/b/c//d/e/f"""
eq_("/foo/bar", str(Path("/foo") + Path("/bar")))
def test_create_with_tuple_that_have_slash_inside(force_ossep, monkeypatch):
eq_(("", "foo", "bar"), Path(("/foo", "bar")))
monkeypatch.setattr(os, "sep", "\\")
eq_(("", "foo", "bar"), Path(("\\foo", "bar")))
def test_auto_decode_os_sep(force_ossep, monkeypatch):
"""Path should decode any either / or os.sep, but always encode in os.sep."""
eq_(("foo\\bar", "bleh"), Path("foo\\bar/bleh"))
monkeypatch.setattr(os, "sep", "\\")
eq_(("foo", "bar/bleh"), Path("foo\\bar/bleh"))
path = Path("foo/bar")
eq_(("foo", "bar"), path)
eq_("foo\\bar", str(path))
def test_contains(force_ossep):
p = Path(("foo", "bar"))
assert Path(("foo", "bar", "bleh")) in p
assert Path(("foo", "bar")) in p
assert "foo" in p
assert "bleh" not in p
assert Path("foo") not in p
def test_is_parent_of(force_ossep):
assert Path(("foo", "bar")).is_parent_of(Path(("foo", "bar", "bleh")))
assert not Path(("foo", "bar")).is_parent_of(Path(("foo", "baz")))
assert not Path(("foo", "bar")).is_parent_of(Path(("foo", "bar")))
def test_windows_drive_letter(force_ossep):
p = Path(("c:",))
eq_("c:\\", str(p))
def test_root_path(force_ossep):
p = Path("/")
eq_("/", str(p))
def test_str_encodes_unicode_to_getfilesystemencoding(force_ossep):
p = Path(("foo", "bar\u00e9"))
eq_("foo/bar\u00e9".encode(sys.getfilesystemencoding()), p.tobytes())
def test_unicode(force_ossep):
p = Path(("foo", "bar\u00e9"))
eq_("foo/bar\u00e9", str(p))
def test_str_repr_of_mix_between_non_ascii_str_and_unicode(force_ossep):
u = "foo\u00e9"
encoded = u.encode(sys.getfilesystemencoding())
p = Path((encoded, "bar"))
print(repr(tuple(p)))
eq_("foo\u00e9/bar".encode(sys.getfilesystemencoding()), p.tobytes())
def test_path_of_a_path_returns_self(force_ossep):
# if Path() is called with a path as value, just return value.
p = Path("foo/bar")
assert Path(p) is p
def test_getitem_str(force_ossep):
# path['something'] returns the child path corresponding to the name
p = Path("/foo/bar")
eq_(p["baz"], Path("/foo/bar/baz"))
def test_getitem_path(force_ossep):
# path[Path('something')] returns the child path corresponding to the name (or subpath)
p = Path("/foo/bar")
eq_(p[Path("baz/bleh")], Path("/foo/bar/baz/bleh"))
@pytest.mark.xfail(reason="pytest's capture mechanism is flaky, I have to investigate")
def test_log_unicode_errors(force_ossep, monkeypatch, capsys):
# When an there's a UnicodeDecodeError on path creation, log it so it can be possible
# to debug the cause of it.
monkeypatch.setattr(sys, "getfilesystemencoding", lambda: "ascii")
with pytest.raises(UnicodeDecodeError):
Path(["", b"foo\xe9"])
out, err = capsys.readouterr()
assert repr(b"foo\xe9") in err
def test_has_drive_letter(monkeypatch):
monkeypatch.setattr(os, "sep", "\\")
p = Path("foo\\bar")
assert not p.has_drive_letter()
p = Path("C:\\")
assert p.has_drive_letter()
p = Path("z:\\foo")
assert p.has_drive_letter()
def test_remove_drive_letter(monkeypatch):
monkeypatch.setattr(os, "sep", "\\")
p = Path("foo\\bar")
eq_(p.remove_drive_letter(), Path("foo\\bar"))
p = Path("C:\\")
eq_(p.remove_drive_letter(), Path(""))
p = Path("z:\\foo")
eq_(p.remove_drive_letter(), Path("foo"))
from ..path import pathify
from pathlib import Path
def test_pathify():

View File

@@ -11,7 +11,7 @@ from io import StringIO
from pytest import raises
from ..testutil import eq_
from ..path import Path
from pathlib import Path
from ..util import (
nonone,
tryint,
@@ -245,30 +245,30 @@ class TestCaseDeleteIfEmpty:
def test_not_empty(self, tmpdir):
testpath = Path(str(tmpdir))
testpath["foo"].mkdir()
testpath.joinpath("foo").mkdir()
assert not delete_if_empty(testpath)
assert testpath.exists()
def test_with_files_to_delete(self, tmpdir):
testpath = Path(str(tmpdir))
testpath["foo"].open("w")
testpath["bar"].open("w")
testpath.joinpath("foo").touch()
testpath.joinpath("bar").touch()
assert delete_if_empty(testpath, ["foo", "bar"])
assert not testpath.exists()
def test_directory_in_files_to_delete(self, tmpdir):
testpath = Path(str(tmpdir))
testpath["foo"].mkdir()
testpath.joinpath("foo").mkdir()
assert not delete_if_empty(testpath, ["foo"])
assert testpath.exists()
def test_delete_files_to_delete_only_if_dir_is_empty(self, tmpdir):
testpath = Path(str(tmpdir))
testpath["foo"].open("w")
testpath["bar"].open("w")
testpath.joinpath("foo").touch()
testpath.joinpath("bar").touch()
assert not delete_if_empty(testpath, ["foo"])
assert testpath.exists()
assert testpath["foo"].exists()
assert testpath.joinpath("foo").exists()
def test_doesnt_exist(self):
# When the 'path' doesn't exist, just do nothing.
@@ -276,8 +276,8 @@ class TestCaseDeleteIfEmpty:
def test_is_file(self, tmpdir):
# When 'path' is a file, do nothing.
p = Path(str(tmpdir)) + "filename"
p.open("w").close()
p = Path(str(tmpdir)).joinpath("filename")
p.touch()
delete_if_empty(p) # no crash
def test_ioerror(self, tmpdir, monkeypatch):

View File

@@ -15,7 +15,8 @@ import glob
import shutil
from datetime import timedelta
from .path import Path, pathify, log_io_error
from pathlib import Path
from .path import pathify, log_io_error
def nonone(value, replace_value):
@@ -354,13 +355,13 @@ def find_in_path(name, paths=None):
@pathify
def delete_if_empty(path: Path, files_to_delete=[]):
"""Deletes the directory at 'path' if it is empty or if it only contains files_to_delete."""
if not path.exists() or not path.isdir():
if not path.exists() or not path.is_dir():
return
contents = path.listdir()
if any(p for p in contents if (p.name not in files_to_delete) or p.isdir()):
contents = list(path.glob("*"))
if any(p for p in contents if (p.name not in files_to_delete) or p.is_dir()):
return False
for p in contents:
p.remove()
p.unlink()
path.rmdir()
return True