Vito Sansevero
e59206b099
Trash mode on macOS now calls -[NSFileManager trashItemAtURL:resultingItemURL:error:] through ctypes (no dependency), in utils/trash_macos.py. Before any move it refuses what it cannot show to be recoverable, already in the review dialog: network volumes, read-only volumes, iCloud Drive and File Provider folders (cloud-sync), a trash folder inside an output root (by file identity), and files macOS would refuse for permission (no-permission). A refused move keeps the file and logs the NSError; a move macOS reports done but whose place cannot be shown goes under errors as trash-unconfirmed, keeps its records and stops the job. The native proof runs in portability.yml's new native-trash job on macos-latest and macos-15-intel (PM_NATIVE_TRASH=macos). Windows still reports no-trash (#500).
192 lines
7.6 KiB
Python
192 lines
7.6 KiB
Python
"""Native proof of the macOS trash backend (#499 design §10.2).
|
|
|
|
Runs only on macOS with ``PM_NATIVE_TRASH=macos`` (the ``native-trash`` job in
|
|
portability.yml), because it puts files into the real Trash of the user
|
|
running it. With the variable set, a bridge that cannot load fails the test
|
|
rather than skipping it. Files moved into a trash folder are left there: the
|
|
runner is disposable and this test never deletes from a trash folder.
|
|
"""
|
|
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
|
|
from utils import trash # noqa: E402
|
|
|
|
NATIVE = sys.platform == "darwin" and os.environ.get("PM_NATIVE_TRASH") == "macos"
|
|
HDIUTIL_SECONDS = 60
|
|
|
|
|
|
class SpyBridge:
|
|
"""The real bridge, recording every ``trash`` result."""
|
|
|
|
def __init__(self, real):
|
|
self.real = real
|
|
self.results = []
|
|
|
|
def facts(self, path):
|
|
return self.real.facts(path)
|
|
|
|
def trash_dir(self, path):
|
|
return self.real.trash_dir(path)
|
|
|
|
def trash(self, path):
|
|
result = self.real.trash(path)
|
|
self.results.append(result)
|
|
return result
|
|
|
|
|
|
def _hdiutil(*args):
|
|
return subprocess.run(
|
|
["hdiutil", *args], capture_output=True, text=True, timeout=HDIUTIL_SECONDS
|
|
)
|
|
|
|
|
|
@unittest.skipUnless(NATIVE, "macOS native trash proof (PM_NATIVE_TRASH=macos)")
|
|
class MacTrashNativeTests(unittest.TestCase):
|
|
def setUp(self):
|
|
from utils import trash_macos
|
|
|
|
self.trash_macos = trash_macos
|
|
self.spy = SpyBridge(trash_macos.load_bridge()) # fails, never skips
|
|
self.backend = trash_macos.MacTrashBackend(bridge=self.spy)
|
|
self._tmp = tempfile.TemporaryDirectory(prefix="pm-native-trash-")
|
|
self.tmp = os.path.realpath(self._tmp.name)
|
|
self._mounts = []
|
|
self._modes = []
|
|
|
|
def tearDown(self):
|
|
failures = []
|
|
try:
|
|
for path, mode in reversed(self._modes):
|
|
os.chmod(path, mode)
|
|
for mount in reversed(self._mounts):
|
|
done = _hdiutil("detach", mount)
|
|
if done.returncode != 0:
|
|
_hdiutil("detach", mount, "-force")
|
|
failures.append(f"detach {mount}: {done.stderr.strip()}")
|
|
finally:
|
|
self._tmp.cleanup()
|
|
self.assertEqual(failures, [])
|
|
|
|
def _file(self, folder, name="a.png", content=None):
|
|
os.makedirs(folder, exist_ok=True)
|
|
path = os.path.join(folder, name)
|
|
with open(path, "wb") as handle:
|
|
handle.write(content or os.urandom(64))
|
|
return path
|
|
|
|
def _image(self, name, fs="APFS"):
|
|
image = os.path.join(self.tmp, name + ".dmg")
|
|
mount = os.path.join(self.tmp, name + "-mnt")
|
|
os.makedirs(mount)
|
|
made = _hdiutil("create", "-size", "40m", "-fs", fs, "-volname", name, image)
|
|
self.assertEqual(made.returncode, 0, made.stderr)
|
|
attached = _hdiutil("attach", image, "-nobrowse", "-mountpoint", mount)
|
|
self.assertEqual(attached.returncode, 0, attached.stderr)
|
|
self._mounts.append(mount)
|
|
return image, mount
|
|
|
|
def _assert_trashed(self, source, content, info, folder):
|
|
self.assertFalse(os.path.lexists(source))
|
|
ok, resulting, error = self.spy.results[-1]
|
|
self.assertTrue(ok, error)
|
|
self.assertEqual(os.path.realpath(os.path.dirname(resulting)), folder)
|
|
moved = os.lstat(resulting)
|
|
self.assertEqual((moved.st_dev, moved.st_ino), (info.st_dev, info.st_ino))
|
|
with open(resulting, "rb") as handle:
|
|
self.assertEqual(handle.read(), content)
|
|
return resulting
|
|
|
|
def _assert_kept(self, path, content):
|
|
with open(path, "rb") as handle:
|
|
self.assertEqual(handle.read(), content)
|
|
|
|
def _assert_refused(self, path, roots, code):
|
|
with open(path, "rb") as handle:
|
|
content = handle.read()
|
|
self.assertEqual(self.backend.available(path, roots), code)
|
|
with self.assertRaises(trash.TrashRefused) as caught:
|
|
self.backend.move(path, roots)
|
|
self.assertEqual(caught.exception.code, code)
|
|
self._assert_kept(path, content)
|
|
|
|
def test_n1_startup_volume_goes_to_the_home_trash(self):
|
|
root = os.path.join(self.tmp, "output")
|
|
source = self._file(root)
|
|
with open(source, "rb") as handle:
|
|
content = handle.read()
|
|
info = os.lstat(source)
|
|
self.assertIsNone(self.backend.available(source, [root]))
|
|
self.assertEqual(self.backend.move(source, [root]), trash.HOME_TRASH_LABEL)
|
|
home_trash = os.path.realpath(os.path.expanduser("~/.Trash"))
|
|
self._assert_trashed(source, content, info, home_trash)
|
|
|
|
def test_n2_a_name_clash_never_replaces(self):
|
|
first_root = os.path.join(self.tmp, "one")
|
|
second_root = os.path.join(self.tmp, "two")
|
|
first = self._file(first_root, "same.png", b"first")
|
|
second = self._file(second_root, "same.png", b"second")
|
|
self.backend.move(first, [first_root])
|
|
self.backend.move(second, [second_root])
|
|
(_, one, _), (_, two, _) = self.spy.results
|
|
self.assertNotEqual(one, two)
|
|
self._assert_kept(one, b"first")
|
|
self._assert_kept(two, b"second")
|
|
|
|
def test_n3_fresh_external_volume_uses_its_own_trash(self):
|
|
for fs in ("APFS", "MS-DOS FAT32"):
|
|
with self.subTest(fs=fs):
|
|
_image, mount = self._image("fresh" + fs.split()[0], fs)
|
|
root = os.path.join(mount, "output")
|
|
source = self._file(root)
|
|
with open(source, "rb") as handle:
|
|
content = handle.read()
|
|
info = os.lstat(source)
|
|
self.assertFalse(os.path.lexists(os.path.join(mount, ".Trashes")))
|
|
self.assertIsNone(self.backend.available(source, [root]))
|
|
label = self.backend.move(source, [root])
|
|
uid = str(os.getuid())
|
|
self.assertEqual(label, f".Trashes/{uid} on its drive")
|
|
folder = os.path.realpath(os.path.join(mount, ".Trashes", uid))
|
|
self._assert_trashed(source, content, info, folder)
|
|
|
|
def test_n4_volume_root_as_output_root_is_trash_in_root(self):
|
|
_image, mount = self._image("rooted")
|
|
source = self._file(os.path.join(mount, "output"))
|
|
self._assert_refused(source, [mount], trash.TRASH_IN_ROOT)
|
|
self.assertEqual(self.spy.results, [])
|
|
|
|
def test_n5_read_only_volume_is_no_trash(self):
|
|
image, mount = self._image("readonly")
|
|
source = self._file(mount, "ro.png", b"read-only bytes")
|
|
done = _hdiutil("detach", mount)
|
|
self.assertEqual(done.returncode, 0, done.stderr)
|
|
self._mounts.remove(mount)
|
|
attached = _hdiutil(
|
|
"attach", image, "-nobrowse", "-readonly", "-mountpoint", mount
|
|
)
|
|
self.assertEqual(attached.returncode, 0, attached.stderr)
|
|
self._mounts.append(mount)
|
|
self._assert_refused(source, [os.path.join(mount, "output")], trash.NO_TRASH)
|
|
|
|
def test_n6_unwritable_folder_is_no_permission_and_macos_agrees(self):
|
|
root = os.path.join(self.tmp, "locked")
|
|
source = self._file(root, "p.png", b"kept")
|
|
self._modes.append((root, 0o755))
|
|
os.chmod(root, 0o555)
|
|
self._assert_refused(source, [self.tmp], trash.NO_PERMISSION)
|
|
self.assertEqual(self.spy.results, [])
|
|
ok, _resulting, error = self.spy.real.trash(source)
|
|
self.assertFalse(ok)
|
|
self.assertEqual((error.domain, error.code), ("NSCocoaErrorDomain", 513))
|
|
self._assert_kept(source, b"kept")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|