Vito Sansevero
ea3c3d4dce
Maintainer ruling (a): Windows never uses the Recycle Bin, because it deletes files larger than the bin's capacity permanently and IFileOperation's PreDeleteItem guard cannot detect it (native probe runs 40951037575, 40953347904). Each file in the Delete review now reads "Moving files to the Recycle Bin isn't available on Windows: files larger than the bin's capacity would be deleted permanently." (code windows-recycle) instead of "This drive has no trash.", and the summary line drops its "yet": "Moving to the trash is not available for these files; each one lists why."
313 lines
14 KiB
Python
313 lines
14 KiB
Python
"""The freedesktop trash backend (#480): rename into a trash or refuse, never unlink.
|
|
|
|
Every test builds a disposable tree: a fake ``XDG_DATA_HOME`` and, through the
|
|
backend's ``mounts`` seam, a fake "drive" mount point, so the real trash of the
|
|
user running the tests is never touched. Moves are real renames on one
|
|
filesystem.
|
|
"""
|
|
|
|
import os
|
|
import shutil
|
|
import stat
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from unittest import mock
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
|
|
from utils import trash # noqa: E402
|
|
|
|
POSIX = os.name == "posix" and sys.platform != "darwin"
|
|
|
|
|
|
@unittest.skipUnless(POSIX, "freedesktop trash is the Linux/BSD backend")
|
|
class FreedesktopTrashTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp = tempfile.mkdtemp(prefix="pm-trash-")
|
|
self.addCleanup(shutil.rmtree, self.tmp, True)
|
|
self.data_home = os.path.join(self.tmp, "xdg")
|
|
self.drive = os.path.join(self.tmp, "drive")
|
|
self.root = os.path.join(self.drive, "output")
|
|
self.home_root = os.path.join(self.tmp, "home-output")
|
|
for folder in (self.data_home, self.root, self.home_root):
|
|
os.makedirs(folder)
|
|
self.mounts = ["/", os.path.realpath(self.drive)]
|
|
self.backend = trash.FreedesktopBackend(
|
|
data_home=self.data_home, mounts=lambda: list(self.mounts)
|
|
)
|
|
|
|
def _file(self, folder, name="a.png", content=b"png-bytes"):
|
|
path = os.path.join(folder, name)
|
|
with open(path, "wb") as handle:
|
|
handle.write(content)
|
|
return path
|
|
|
|
def _info(self, trash_dir, name):
|
|
with open(os.path.join(trash_dir, "info", name + ".trashinfo")) as handle:
|
|
return handle.read()
|
|
|
|
# ── choosing the trash ────────────────────────────────────────────
|
|
|
|
def test_file_on_the_home_mount_goes_to_the_home_trash(self):
|
|
path = self._file(self.home_root)
|
|
label = self.backend.move(path, [self.home_root])
|
|
home_trash = os.path.join(self.data_home, "Trash")
|
|
self.assertEqual(label, trash.HOME_TRASH_LABEL)
|
|
self.assertFalse(os.path.exists(path))
|
|
with open(os.path.join(home_trash, "files", "a.png"), "rb") as handle:
|
|
self.assertEqual(handle.read(), b"png-bytes")
|
|
info = self._info(home_trash, "a.png")
|
|
self.assertIn("[Trash Info]\n", info)
|
|
self.assertIn("Path=" + os.path.realpath(path) + "\n", info)
|
|
self.assertRegex(info, r"DeletionDate=\d{4}-\d\d-\d\dT\d\d:\d\d:\d\d\n")
|
|
|
|
def test_file_on_another_mount_uses_its_trash_dash_uid(self):
|
|
path = self._file(self.root, "b c%.png")
|
|
self.backend.move(path, [self.root])
|
|
top = os.path.join(self.drive, f".Trash-{os.getuid()}")
|
|
info_mode = stat.S_IMODE(os.lstat(top).st_mode)
|
|
self.assertEqual(info_mode, 0o700)
|
|
self.assertTrue(os.path.exists(os.path.join(top, "files", "b c%.png")))
|
|
# A top-directory trash records the path relative to its top folder,
|
|
# percent-encoded.
|
|
self.assertIn("Path=output/b%20c%25.png\n", self._info(top, "b c%.png"))
|
|
|
|
def test_sticky_dot_trash_folder_is_used(self):
|
|
shared = os.path.join(self.drive, ".Trash")
|
|
os.mkdir(shared)
|
|
os.chmod(shared, 0o1777)
|
|
path = self._file(self.root)
|
|
self.backend.move(path, [self.root])
|
|
self.assertTrue(
|
|
os.path.exists(os.path.join(shared, str(os.getuid()), "files", "a.png"))
|
|
)
|
|
|
|
def test_dot_trash_without_sticky_bit_is_skipped(self):
|
|
shared = os.path.join(self.drive, ".Trash")
|
|
os.mkdir(shared, 0o777)
|
|
os.chmod(shared, 0o777)
|
|
path = self._file(self.root)
|
|
self.backend.move(path, [self.root])
|
|
self.assertFalse(os.path.exists(os.path.join(shared, str(os.getuid()))))
|
|
top = os.path.join(self.drive, f".Trash-{os.getuid()}")
|
|
self.assertTrue(os.path.exists(os.path.join(top, "files", "a.png")))
|
|
|
|
def test_dot_trash_that_is_a_link_is_skipped(self):
|
|
elsewhere = os.path.join(self.tmp, "elsewhere")
|
|
os.mkdir(elsewhere)
|
|
os.chmod(elsewhere, 0o1777)
|
|
os.symlink(elsewhere, os.path.join(self.drive, ".Trash"))
|
|
path = self._file(self.root)
|
|
self.backend.move(path, [self.root])
|
|
self.assertEqual(os.listdir(elsewhere), [])
|
|
|
|
def test_trash_dash_uid_that_is_a_link_is_refused(self):
|
|
elsewhere = os.path.join(self.tmp, "elsewhere")
|
|
os.mkdir(elsewhere)
|
|
os.symlink(elsewhere, os.path.join(self.drive, f".Trash-{os.getuid()}"))
|
|
path = self._file(self.root)
|
|
with self.assertRaises(trash.TrashRefused) as caught:
|
|
self.backend.move(path, [self.root])
|
|
self.assertEqual(caught.exception.code, trash.NO_TRASH)
|
|
self.assertTrue(os.path.exists(path))
|
|
self.assertEqual(os.listdir(elsewhere), [])
|
|
|
|
def test_trash_inside_an_output_root_is_refused(self):
|
|
# The drive's top folder is itself an output root: its .Trash-uid
|
|
# would be listed as images.
|
|
path = self._file(self.root)
|
|
with self.assertRaises(trash.TrashRefused) as caught:
|
|
self.backend.move(path, [self.drive])
|
|
self.assertEqual(caught.exception.code, trash.TRASH_IN_ROOT)
|
|
self.assertTrue(os.path.exists(path))
|
|
self.assertFalse(
|
|
os.path.exists(os.path.join(self.drive, f".Trash-{os.getuid()}"))
|
|
)
|
|
|
|
def test_home_trash_inside_an_output_root_is_refused(self):
|
|
path = self._file(self.home_root)
|
|
with self.assertRaises(trash.TrashRefused) as caught:
|
|
self.backend.move(path, [self.home_root, self.tmp])
|
|
self.assertEqual(caught.exception.code, trash.TRASH_IN_ROOT)
|
|
self.assertTrue(os.path.exists(path))
|
|
|
|
def test_read_only_top_folder_is_no_trash(self):
|
|
path = self._file(self.root)
|
|
os.chmod(self.drive, 0o555)
|
|
self.addCleanup(os.chmod, self.drive, 0o755)
|
|
if os.access(self.drive, os.W_OK):
|
|
self.skipTest("running as root: permissions are not enforced")
|
|
self.assertEqual(self.backend.available(path, [self.root]), trash.NO_TRASH)
|
|
with self.assertRaises(trash.TrashRefused) as caught:
|
|
self.backend.move(path, [self.root])
|
|
self.assertEqual(caught.exception.code, trash.NO_TRASH)
|
|
self.assertTrue(os.path.exists(path))
|
|
|
|
def test_available_reports_without_writing(self):
|
|
path = self._file(self.root)
|
|
self.assertIsNone(self.backend.available(path, [self.root]))
|
|
self.assertEqual(
|
|
self.backend.available(path, [self.drive]), trash.TRASH_IN_ROOT
|
|
)
|
|
self.assertEqual(os.listdir(self.drive), ["output"])
|
|
|
|
# ── names and the no-replace rename ───────────────────────────────
|
|
|
|
def test_existing_name_in_the_trash_is_never_replaced(self):
|
|
top = os.path.join(self.drive, f".Trash-{os.getuid()}")
|
|
os.makedirs(os.path.join(top, "files"), mode=0o700)
|
|
os.makedirs(os.path.join(top, "info"), mode=0o700)
|
|
os.chmod(top, 0o700)
|
|
# A stray trashed file with no info file (another tool crashed).
|
|
stray = self._file(os.path.join(top, "files"), "a.png", b"older-file")
|
|
path = self._file(self.root)
|
|
self.backend.move(path, [self.root])
|
|
with open(stray, "rb") as handle:
|
|
self.assertEqual(handle.read(), b"older-file")
|
|
self.assertTrue(os.path.exists(os.path.join(top, "files", "a.1.png")))
|
|
self.assertTrue(os.path.exists(os.path.join(top, "info", "a.1.png.trashinfo")))
|
|
self.assertFalse(os.path.exists(os.path.join(top, "info", "a.png.trashinfo")))
|
|
|
|
def test_existing_info_name_is_skipped(self):
|
|
path = self._file(self.root)
|
|
self.backend.move(path, [self.root])
|
|
second = self._file(self.root)
|
|
self.backend.move(second, [self.root])
|
|
top = os.path.join(self.drive, f".Trash-{os.getuid()}")
|
|
self.assertEqual(
|
|
sorted(os.listdir(os.path.join(top, "files"))), ["a.1.png", "a.png"]
|
|
)
|
|
|
|
def test_missing_no_replace_rename_is_no_trash(self):
|
|
path = self._file(self.root)
|
|
for error in (38, 22): # ENOSYS, EINVAL
|
|
with self.subTest(error=error):
|
|
backend = trash.FreedesktopBackend(
|
|
data_home=self.data_home,
|
|
mounts=lambda: list(self.mounts),
|
|
rename=mock.Mock(side_effect=OSError(error, "no")),
|
|
)
|
|
with self.assertRaises(trash.TrashRefused) as caught:
|
|
backend.move(path, [self.root])
|
|
self.assertEqual(caught.exception.code, trash.NO_TRASH)
|
|
self.assertTrue(os.path.exists(path))
|
|
top = os.path.join(self.drive, f".Trash-{os.getuid()}")
|
|
self.assertEqual(os.listdir(os.path.join(top, "info")), [])
|
|
|
|
def test_cross_device_rename_is_no_trash_and_never_copies(self):
|
|
path = self._file(self.root)
|
|
backend = trash.FreedesktopBackend(
|
|
data_home=self.data_home,
|
|
mounts=lambda: list(self.mounts),
|
|
rename=mock.Mock(side_effect=OSError(18, "EXDEV")),
|
|
)
|
|
with mock.patch("shutil.move") as move, mock.patch("shutil.copy2") as copy:
|
|
with self.assertRaises(trash.TrashRefused) as caught:
|
|
backend.move(path, [self.root])
|
|
self.assertEqual(caught.exception.code, trash.NO_TRASH)
|
|
move.assert_not_called()
|
|
copy.assert_not_called()
|
|
self.assertTrue(os.path.exists(path))
|
|
|
|
def test_other_rename_failure_is_failed_and_removes_only_our_info(self):
|
|
path = self._file(self.root)
|
|
backend = trash.FreedesktopBackend(
|
|
data_home=self.data_home,
|
|
mounts=lambda: list(self.mounts),
|
|
rename=mock.Mock(side_effect=OSError(5, "EIO")),
|
|
)
|
|
with self.assertRaises(trash.TrashRefused) as caught:
|
|
backend.move(path, [self.root])
|
|
self.assertEqual(caught.exception.code, trash.FAILED)
|
|
top = os.path.join(self.drive, f".Trash-{os.getuid()}")
|
|
self.assertEqual(os.listdir(os.path.join(top, "info")), [])
|
|
self.assertTrue(os.path.exists(path))
|
|
|
|
def test_never_removes_or_moves_the_source(self):
|
|
path = self._file(self.root)
|
|
real_unlink = os.unlink
|
|
|
|
def guarded_unlink(target, *args, **kwargs):
|
|
# Only our own .trashinfo may ever be removed.
|
|
self.assertTrue(str(target).endswith(".trashinfo"), target)
|
|
return real_unlink(target, *args, **kwargs)
|
|
|
|
backend = trash.FreedesktopBackend(
|
|
data_home=self.data_home,
|
|
mounts=lambda: list(self.mounts),
|
|
rename=mock.Mock(side_effect=OSError(5, "EIO")),
|
|
)
|
|
with (
|
|
mock.patch("os.unlink", guarded_unlink),
|
|
mock.patch("os.remove", guarded_unlink),
|
|
mock.patch("shutil.move") as move,
|
|
mock.patch("os.rename") as rename,
|
|
):
|
|
with self.assertRaises(trash.TrashRefused):
|
|
backend.move(path, [self.root])
|
|
move.assert_not_called()
|
|
rename.assert_not_called()
|
|
self.assertTrue(os.path.exists(path))
|
|
|
|
def test_mountinfo_parsing_decodes_escapes(self):
|
|
text = (
|
|
"36 35 98:0 / / rw - ext4 /dev/root rw\n"
|
|
"37 36 98:1 / /mnt/my\\040drive rw - ext4 /dev/sdb1 rw\n"
|
|
)
|
|
self.assertEqual(trash.parse_mountinfo(text), ["/", "/mnt/my drive"])
|
|
|
|
def test_mount_of_picks_the_longest_containing_mount(self):
|
|
mounts = ["/", "/mnt", "/mnt/data"]
|
|
self.assertEqual(trash.mount_of("/mnt/data/x/y.png", mounts), "/mnt/data")
|
|
self.assertEqual(trash.mount_of("/mnt/datax/y.png", mounts), "/mnt")
|
|
self.assertEqual(trash.mount_of("/home/u/y.png", mounts), "/")
|
|
|
|
|
|
class UnavailableBackendTests(unittest.TestCase):
|
|
def test_reports_no_trash_and_leaves_the_file(self):
|
|
backend = trash.UnavailableBackend()
|
|
with tempfile.TemporaryDirectory() as folder:
|
|
path = os.path.join(folder, "a.png")
|
|
with open(path, "wb") as handle:
|
|
handle.write(b"x")
|
|
self.assertEqual(backend.available(path, [folder]), trash.NO_TRASH)
|
|
with self.assertRaises(trash.TrashRefused) as caught:
|
|
backend.move(path, [folder])
|
|
self.assertEqual(caught.exception.code, trash.NO_TRASH)
|
|
self.assertTrue(os.path.exists(path))
|
|
|
|
def test_windows_and_macos_have_no_trash_yet(self):
|
|
for platform in ("win32", "darwin"):
|
|
with self.subTest(platform=platform):
|
|
self.assertIsInstance(
|
|
trash.default_backend(platform), trash.UnavailableBackend
|
|
)
|
|
self.assertIsInstance(trash.default_backend("linux"), trash.FreedesktopBackend)
|
|
|
|
def test_windows_says_why_the_recycle_bin_is_not_used(self):
|
|
# #500 ruling (a): the Recycle Bin can delete permanently, so it is never used.
|
|
backend = trash.default_backend("win32")
|
|
self.assertEqual(
|
|
backend.available("C:/x/a.png", ["C:/x"]), trash.WINDOWS_RECYCLE
|
|
)
|
|
with self.assertRaises(trash.TrashRefused) as caught:
|
|
backend.move("C:/x/a.png", ["C:/x"])
|
|
self.assertEqual(caught.exception.code, trash.WINDOWS_RECYCLE)
|
|
self.assertIn(trash.WINDOWS_RECYCLE, trash.CODES)
|
|
self.assertEqual(
|
|
trash.default_backend("darwin").available("/x", ["/"]), trash.NO_TRASH
|
|
)
|
|
|
|
def test_module_functions_use_the_replaceable_backend(self):
|
|
fake = mock.Mock()
|
|
fake.available.return_value = "network"
|
|
fake.move.return_value = "label"
|
|
with mock.patch.object(trash, "BACKEND", fake):
|
|
self.assertEqual(trash.trash_available("/x/a.png", ["/x"]), "network")
|
|
self.assertEqual(trash.move_to_trash("/x/a.png", ["/x"]), "label")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|