diff --git a/.github/workflows/portability.yml b/.github/workflows/portability.yml index 43782e03..dc084cfb 100644 --- a/.github/workflows/portability.yml +++ b/.github/workflows/portability.yml @@ -25,6 +25,9 @@ env: # 40240706568 (fb9134f), 40228167328 (f9835aa), before these legs moved here in #420: # slowest py3.10 on windows-latest: 12m43s / 15m54s / 14m53s -> 27 min # (macos 6m14s-8m30s, windows 11m42s-15m54s) +# native-trash (#499) has no green main runs yet: its probe, run 40949370943 (2026-10-29), +# took about 1 minute per macOS leg -> ceil(1.5 x 2) + 3 = 6 min (step 4). Recalibrate after +# three green runs. jobs: test: name: py${{ matrix.python-version }} on ${{ matrix.os }} @@ -75,3 +78,28 @@ jobs: env: PM_TEST_SLOWEST: "40" run: python -m tests.run_suite -v + + # The real Trash on both macOS CPUs (#499 design §10.2). It runs apart from the suite, so a red + # suite never skips the native proof. + native-trash: + name: native trash on ${{ matrix.os }} + runs-on: ${{ matrix.os }} + timeout-minutes: 6 + strategy: + fail-fast: false + matrix: + os: [macos-latest, macos-15-intel] + + steps: + - uses: actions/checkout@v7 + + - name: Set up Python + uses: actions/setup-python@v7 + with: + python-version: "3.12" + + - name: Native trash proof + timeout-minutes: 4 + env: + PM_NATIVE_TRASH: macos + run: python -m unittest tests.test_trash_macos_native tests.test_trash_macos -v diff --git a/TESTING.md b/TESTING.md index 16a22edb..11f24fca 100644 --- a/TESTING.md +++ b/TESTING.md @@ -2462,10 +2462,26 @@ files to the system trash. **No code path unlinks or overwrites a media file.** freedesktop Trash specification (home trash on the same mount, else the mount's `.Trash/$uid` or `.Trash-$uid`), reserve the name with `O_EXCL` and move with `renameat2(RENAME_NOREPLACE)`; without it, across devices, on a read-only mount or with - a trash inside an output root the move is refused (`no-trash`, `trash-in-root`). Windows - and macOS report `no-trash` until their own backends land; library-only removal works - everywhere. Tests use `tests/fake_trash.py`; the browser test server installs it, so no - test touches a real trash. + a trash inside an output root the move is refused (`no-trash`, `trash-in-root`). macOS + (#499, `utils/trash_macos.py`) calls `-[NSFileManager trashItemAtURL:resultingItemURL:error:]` + through `ctypes` (ten fixed `objc_msgSend` prototypes, nil checks before every send, one + autorelease pool per call, loaded on first use). It answers in the review dialog, without + writing: `network` (not a local volume), `no-trash` (read-only, or no trash folder + macOS would use), `cloud-sync` (iCloud Drive, `~/Library/CloudStorage`, or ubiquitous), + `trash-in-root` (the trash folder is inside an output root, compared by `(st_dev, + st_ino)` of every ancestor) and `no-permission` (folder not writable, an immutable or + append flag, a foreign file in a sticky folder). A fresh volume's `.Trashes/` is + predicted (`URLForDirectory` answers 3328/`ENOENT` until macOS makes it). A refused move + keeps the file and logs the `NSError`; a move macOS reports done but whose place cannot be + shown (no result URL, or one inside an output root) goes under `errors` as + `trash-unconfirmed`, keeps its records and stops the job. Windows reports `no-trash` + until #500; library-only removal works everywhere. Tests use `tests/fake_trash.py`; the + browser test server installs it, so no test touches a real trash. The native proof, + `tests/test_trash_macos_native.py`, runs only with `PM_NATIVE_TRASH=macos` in + portability.yml's `native-trash` job (`macos-latest` and `macos-15-intel`): the startup + volume, a name clash, fresh APFS and FAT32 disk images, an image used as an output root, + a read-only image and an unwritable folder (design and probe: + `docs/egm/499-macos-trash/`, local). ### Prompt library cards and recorded deviations (#249 PR E) diff --git a/py/api/image_delete.py b/py/api/image_delete.py index fe011149..5abe4dbc 100644 --- a/py/api/image_delete.py +++ b/py/api/image_delete.py @@ -62,6 +62,9 @@ TITLES = { } MAX_DETAIL_ROWS = 50 TRASH_FOLDERS = frozenset({".trash", ".trashes", "$recycle.bin"}) +# The system moved the file but its place in a trash could not be shown (#499). +TRASH_UNCONFIRMED = "trash-unconfirmed" +RECORDS_FAILED_CODE = "records-failed" # Reason codes (design §4.4) and the sentence each shows. REASONS = { @@ -80,6 +83,17 @@ REASONS = { trash.TRASH_IN_ROOT: "This drive's trash is inside an output folder.", trash.LOCKED: "The file is open in another program. Close it and try again.", trash.FAILED: "The file could not be moved to the trash.", + trash.NO_PERMISSION: ( + "PromptManager isn't allowed to move this file (no permission, or it is " + "locked in Finder)." + ), + trash.CLOUD_SYNC: ( + "Files in iCloud Drive or a cloud-synced folder can't be moved to the " + "trash here." + ), + TRASH_UNCONFIRMED: ( + "Moved by macOS, but its place in the trash could not be confirmed." + ), "not-attempted": "Not attempted: the job stopped after an error.", } _PROBLEM_CODES = { @@ -327,7 +341,7 @@ class ImageDeleteMixin: job.update(total=len(targets), status_line=TITLES[mode]) roots = [str(root) for root in self._get_all_output_dirs()] done_group = "trashed" if mode == "trash" else "removed" - changed, failed = 0, False + changed, failed = 0, None for done, target in enumerate(targets, start=1): if failed: self._record_delete(job, "not_changed", target, "not-attempted") @@ -336,7 +350,8 @@ class ImageDeleteMixin: self._delete_one_image, db, target, mode, roots ) if group == "errors": - failed = True + failed = code or RECORDS_FAILED_CODE + text = REASONS[code] if code else RECORDS_FAILED job.record( "errors", { @@ -344,9 +359,9 @@ class ImageDeleteMixin: "root": target["root_label"], "root_index": target["root_index"], "root_scope": target["root_scope"], - "detail": RECORDS_FAILED, + "detail": text, }, - f"! {target['root_label']}/{target['rel']} · {RECORDS_FAILED}", + f"! {target['root_label']}/{target['rel']} · {text}", ) elif group == done_group: changed += 1 @@ -365,8 +380,10 @@ class ImageDeleteMixin: job.log_line(line) job.update(progress=100, status_line=f"{line} · {backup_note(backup)}") if failed: - job.scope["failure"] = "records-failed" - job.publish({"type": "error", "code": "records-failed", "message": line}) + # "records-failed" names every stop at an errors row (the page's + # "Stopped after N of M" toast); the row says which error it was. + job.scope["failure"] = RECORDS_FAILED_CODE + job.publish({"type": "error", "code": failed, "message": line}) job.finish("failed") @staticmethod @@ -405,7 +422,19 @@ class ImageDeleteMixin: label = trash.move_to_trash(target["abs"], roots) except trash.TrashRefused as refused: code = refused.code if refused.code in REASONS else trash.FAILED + self.logger.info( + "Not moved to the trash (%s): %s %s", + code, + target["rel"], + refused.detail, + ) return "not_changed", code, None + except trash.TrashUnconfirmed as unconfirmed: + # The file left its folder: keep its records, stop the job. + self.logger.warning( + "Trash move not confirmed: %s %s", target["rel"], unconfirmed.detail + ) + return "errors", TRASH_UNCONFIRMED, None mark = None else: mark = (target["size"], target["mtime_ns"]) diff --git a/tests/test_image_delete_routes.py b/tests/test_image_delete_routes.py index 1fb364df..464b3b91 100644 --- a/tests/test_image_delete_routes.py +++ b/tests/test_image_delete_routes.py @@ -313,15 +313,83 @@ class TestFileCheck(ImageDeleteTestCase): self.assertIn(code, image_delete.REASONS) +class TestMacOSOutcomes(ImageDeleteTestCase): + """The codes and the unconfirmed outcome the macOS backend adds (#499).""" + + async def test_permission_and_cloud_refusals_show_their_sentences(self): + self.write("a.png", b"aaa") + self.write("b.png", b"bbb") + self.fake.refuse[str(self.output_dir / "a.png")] = "no-permission" + self.fake.refuse[str(self.output_dir / "b.png")] = "cloud-sync" + plan = await self.review(["a.png", "b.png"]) + self.assertEqual(plan["trash"], {"count": 0}) + self.assertEqual( + plan["trash_refused"]["counts"], {"no-permission": 1, "cloud-sync": 1} + ) + details = {row["detail"] for row in plan["trash_refused"]["rows"]} + self.assertEqual( + details, + {image_delete.REASONS["no-permission"], image_delete.REASONS["cloud-sync"]}, + ) + + async def test_an_unconfirmed_move_keeps_records_and_stops_the_job(self): + from utils import trash + + self.write("a.png", b"aaa", mtime=1_000_000) + self.write("b.png", b"bbb", mtime=2_000_000) + self.link("a.png") + original = self.fake.move + + def move(path, roots): + if path.endswith("a.png"): + original(path, roots) # the file did leave its folder + raise trash.TrashUnconfirmed("resultingItemURL=None") + return original(path, roots) + + self.fake.move = move + _plan, job = await self.delete(["a.png", "b.png"], "trash") + self.assertEqual(job.state, "failed") + self.assertEqual(job.scope.get("failure"), "records-failed") + self.assertEqual( + self.details(job, "errors"), [image_delete.REASONS["trash-unconfirmed"]] + ) + self.assertEqual(self.count("SELECT COUNT(*) FROM generated_images"), 1) + self.assertEqual( + self.details(job, "not_changed"), [image_delete.REASONS["not-attempted"]] + ) + self.assertTrue((self.output_dir / "b.png").exists()) + + async def test_a_refusal_detail_is_logged(self): + from utils import trash + + self.write("a.png", b"aaa") + + def move(path, roots): + raise trash.TrashRefused("no-permission", "NSCocoaErrorDomain 513") + + self.fake.move = move + with self.assertLogs("prompt_manager.api", level="INFO") as logs: + _plan, job = await self.delete(["a.png"], "trash") + self.assertEqual( + self.details(job, "not_changed"), [image_delete.REASONS["no-permission"]] + ) + self.assertTrue(any("NSCocoaErrorDomain 513" in line for line in logs.output)) + self.assertTrue((self.output_dir / "a.png").exists()) + + class TestNativeBackend(ImageDeleteTestCase): - """The real backend of this OS: Windows and macOS have no trash yet (#480).""" + """The real backend of this OS: Windows has no trash yet (#500). + + Only library mode is applied, so even on macOS and Linux no file reaches + a real trash. + """ async def test_library_only_works_and_the_trash_is_unavailable_here(self): fake_trash.uninstall() self.write("a.png", b"aaa") self.link("a.png") plan = await self.review(["a.png"]) - native_trash = sys.platform not in ("win32", "darwin") + native_trash = sys.platform != "win32" if not native_trash: self.assertEqual(plan["trash"], {"count": 0}) self.assertEqual(plan["trash_refused"]["counts"], {"no-trash": 1}) diff --git a/tests/test_trash_freedesktop.py b/tests/test_trash_freedesktop.py index a0f3f3ec..29e9ff8c 100644 --- a/tests/test_trash_freedesktop.py +++ b/tests/test_trash_freedesktop.py @@ -277,12 +277,13 @@ class UnavailableBackendTests(unittest.TestCase): 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 - ) + def test_each_platform_gets_its_backend(self): + from utils import trash_macos + + self.assertIsInstance(trash.default_backend("win32"), trash.UnavailableBackend) + self.assertIsInstance( + trash.default_backend("darwin"), trash_macos.MacTrashBackend + ) self.assertIsInstance(trash.default_backend("linux"), trash.FreedesktopBackend) def test_module_functions_use_the_replaceable_backend(self): diff --git a/tests/test_trash_macos.py b/tests/test_trash_macos.py new file mode 100644 index 00000000..28b65586 --- /dev/null +++ b/tests/test_trash_macos.py @@ -0,0 +1,358 @@ +"""The macOS trash backend's decisions (#499), with a fake Foundation bridge. + +No Objective-C is involved: the bridge is faked, ``uid``/``home``/``lstat``/ +``access`` are injected, and the only real files are temp files used for the +"source still exists" and output-root identity checks. Runs on every OS. +""" + +import os +import shutil +import stat +import sys +import tempfile +import unittest +from types import SimpleNamespace +from unittest import mock + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +from utils import trash, trash_macos # noqa: E402 +from utils.trash_macos import Facts, MacTrashBackend, NSErr # noqa: E402 + +LOCAL = Facts(local=True, read_only=False, root_fs=True, volume="/", ubiquitous=False) +COCOA = "NSCocoaErrorDomain" +NO_TRASH_DIR = NSErr(COCOA, 3328, "NSPOSIXErrorDomain", 2) + + +class FakeBridge: + """The §6.3 bridge with canned answers; ``trash`` moves nothing by default.""" + + def __init__(self, facts=LOCAL, trash_dir=("/Users/me/.Trash", None), result=None): + self.facts_value = facts + self.trash_dir_value = trash_dir + self.result = result or (True, None, None) + self.calls = [] + + def facts(self, path): + self.calls.append(("facts", path)) + return self.facts_value + + def trash_dir(self, path): + self.calls.append(("trash_dir", path)) + return self.trash_dir_value + + def trash(self, path): + self.calls.append(("trash", path)) + result = self.result + return result(path) if callable(result) else result + + +def _stat(mode=stat.S_IFREG | 0o644, uid=501, flags=0): + return SimpleNamespace(st_mode=mode, st_uid=uid, st_flags=flags) + + +class MacBackendTestCase(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.mkdtemp(prefix="pm-mac-trash-") + self.addCleanup(shutil.rmtree, self.tmp, True) + self.home = os.path.join(self.tmp, "home") + self.root = os.path.join(self.tmp, "output") + os.makedirs(os.path.join(self.home, ".Trash")) + os.makedirs(self.root) + self.file = os.path.join(self.root, "a.png") + with open(self.file, "wb") as handle: + handle.write(b"png") + self.home_trash = os.path.join(self.home, ".Trash") + self.stats = {} + self.access_ok = True + + def _lstat(self, path): + if path in self.stats: + value = self.stats[path] + if isinstance(value, Exception): + raise value + return value + return os.lstat(path) + + def backend(self, **bridge_args): + bridge_args.setdefault("trash_dir", (self.home_trash, None)) + self.bridge = FakeBridge(**bridge_args) + return MacTrashBackend( + bridge=self.bridge, + uid=501, + home=self.home, + lstat=self._lstat, + access=lambda path, mode: self.access_ok, + ) + + def moved_to(self, destination): + """A ``trash`` result that really renames the file to *destination*.""" + + def run(path): + os.makedirs(os.path.dirname(destination), exist_ok=True) + os.rename(path, destination) + return True, destination, None + + return run + + def assert_refused(self, backend, code): + self.assertEqual(backend.available(self.file, [self.root]), code) + with self.assertRaises(trash.TrashRefused) as caught: + backend.move(self.file, [self.root]) + self.assertEqual(caught.exception.code, code) + self.assertNotIn("trash", [name for name, _ in self.bridge.calls]) + self.assertTrue(os.path.exists(self.file)) + return caught.exception + + +class AvailableOrderTests(MacBackendTestCase): + def test_a_writable_local_file_is_available(self): + self.assertIsNone(self.backend().available(self.file, [self.root])) + + def test_network_volume_is_refused_without_trying(self): + facts = Facts(False, False, False, "/Volumes/nas", False) + self.assert_refused(self.backend(facts=facts), trash.NETWORK) + + def test_unknown_locality_counts_as_network(self): + facts = Facts(None, False, True, "/", False) + self.assert_refused(self.backend(facts=facts), trash.NETWORK) + + def test_read_only_volume_or_unknown_is_no_trash(self): + for read_only in (True, None): + with self.subTest(read_only=read_only): + facts = Facts(True, read_only, False, "/Volumes/ro", False) + self.assert_refused(self.backend(facts=facts), trash.NO_TRASH) + + def test_ubiquitous_item_is_cloud_sync(self): + facts = Facts(True, False, True, "/", True) + self.assert_refused(self.backend(facts=facts), trash.CLOUD_SYNC) + + def test_icloud_and_file_provider_folders_are_cloud_sync(self): + for parts in (("Library", "Mobile Documents"), ("Library", "CloudStorage")): + with self.subTest(parts=parts): + folder = os.path.join(self.home, *parts, "Box") + os.makedirs(folder) + path = os.path.join(folder, "b.png") + with open(path, "wb") as handle: + handle.write(b"x") + backend = self.backend() + self.assertEqual(backend.available(path, [folder]), trash.CLOUD_SYNC) + with self.assertRaises(trash.TrashRefused): + backend.move(path, [folder]) + self.assertTrue(os.path.exists(path)) + + def test_found_trash_folder_inside_an_output_root_is_trash_in_root(self): + self.assertEqual( + self.backend().available(self.file, [self.home]), trash.TRASH_IN_ROOT + ) + + def test_trash_dir_error_other_than_missing_is_no_trash(self): + backend = self.backend( + trash_dir=(None, NSErr(COCOA, 3328, "NSPOSIXErrorDomain", 30)) + ) + self.assert_refused(backend, trash.NO_TRASH) + + def test_missing_trash_on_the_startup_volume_is_no_trash(self): + self.assert_refused( + self.backend(trash_dir=(None, NO_TRASH_DIR)), trash.NO_TRASH + ) + + +class FreshVolumeTests(MacBackendTestCase): + def setUp(self): + super().setUp() + self.volume = os.path.join(self.tmp, "vol") + self.root = os.path.join(self.volume, "output") + os.makedirs(self.root) + self.file = os.path.join(self.root, "a.png") + with open(self.file, "wb") as handle: + handle.write(b"png") + self.facts = Facts(True, False, False, self.volume, False) + + def test_missing_volume_trash_is_predicted(self): + backend = self.backend(facts=self.facts, trash_dir=(None, NO_TRASH_DIR)) + self.assertIsNone(backend.available(self.file, [self.root])) + + def test_prediction_refused_when_trashes_is_a_link(self): + os.symlink(self.tmp, os.path.join(self.volume, ".Trashes")) + backend = self.backend(facts=self.facts, trash_dir=(None, NO_TRASH_DIR)) + self.assert_refused(backend, trash.NO_TRASH) + + def test_predicted_trash_inside_the_volume_root_is_trash_in_root(self): + backend = self.backend(facts=self.facts, trash_dir=(None, NO_TRASH_DIR)) + self.assert_refused_root = backend.available(self.file, [self.volume]) + self.assertEqual(self.assert_refused_root, trash.TRASH_IN_ROOT) + + def test_move_to_a_predicted_trash_returns_the_volume_label(self): + destination = os.path.join(self.volume, ".Trashes", "501", "a.png") + backend = self.backend( + facts=self.facts, + trash_dir=(None, NO_TRASH_DIR), + result=self.moved_to(destination), + ) + self.assertEqual( + backend.move(self.file, [self.root]), ".Trashes/501 on its drive" + ) + self.assertFalse(os.path.exists(self.file)) + + +class PermissionTests(MacBackendTestCase): + def test_parent_not_writable_is_no_permission(self): + self.access_ok = False + self.assert_refused(self.backend(), trash.NO_PERMISSION) + + def test_immutable_or_append_flags_on_file_or_parent(self): + for flag in ("UF_IMMUTABLE", "SF_IMMUTABLE", "UF_APPEND", "SF_APPEND"): + for target in (self.file, self.root): + with self.subTest(flag=flag, target=target): + mode = ( + stat.S_IFDIR | 0o755 + if target == self.root + else stat.S_IFREG | 0o644 + ) + self.stats = {target: _stat(mode=mode, flags=getattr(stat, flag))} + self.assert_refused(self.backend(), trash.NO_PERMISSION) + + def test_sticky_parent_and_foreign_owner(self): + self.stats = { + self.root: _stat(mode=stat.S_IFDIR | 0o1777), + self.file: _stat(uid=502), + } + self.assert_refused(self.backend(), trash.NO_PERMISSION) + + def test_sticky_parent_with_own_file_is_allowed(self): + self.stats = {self.root: _stat(mode=stat.S_IFDIR | 0o1777), self.file: _stat()} + self.assertIsNone(self.backend().available(self.file, [self.root])) + + +class MoveOutcomeTests(MacBackendTestCase): + def test_success_into_the_home_trash(self): + destination = os.path.join(self.home_trash, "a.png") + backend = self.backend(result=self.moved_to(destination)) + self.assertEqual(backend.move(self.file, [self.root]), trash.HOME_TRASH_LABEL) + self.assertTrue(os.path.exists(destination)) + + def test_cocoa_errors_map_to_codes_with_detail(self): + cases = { + 3328: trash.NO_TRASH, + 642: trash.NO_TRASH, + 513: trash.NO_PERMISSION, + 257: trash.NO_PERMISSION, + 4: trash.FAILED, + 640: trash.FAILED, + } + for code, expected in cases.items(): + with self.subTest(code=code): + error = NSErr(COCOA, code, "NSOSStatusErrorDomain", -5000) + backend = self.backend(result=(False, None, error)) + with self.assertRaises(trash.TrashRefused) as caught: + backend.move(self.file, [self.root]) + self.assertEqual(caught.exception.code, expected) + self.assertEqual( + caught.exception.detail, + f"NSCocoaErrorDomain {code} (NSOSStatusErrorDomain -5000)", + ) + self.assertTrue(os.path.exists(self.file)) + + def test_other_domain_or_missing_error_is_failed(self): + for error in (NSErr("NSPOSIXErrorDomain", 13), None): + with self.subTest(error=error): + backend = self.backend(result=(False, None, error)) + with self.assertRaises(trash.TrashRefused) as caught: + backend.move(self.file, [self.root]) + self.assertEqual(caught.exception.code, trash.FAILED) + + def test_yes_with_the_source_still_present_is_failed(self): + backend = self.backend( + result=(True, os.path.join(self.home_trash, "a.png"), None) + ) + with self.assertRaises(trash.TrashRefused) as caught: + backend.move(self.file, [self.root]) + self.assertEqual(caught.exception.code, trash.FAILED) + self.assertTrue(os.path.exists(self.file)) + + def test_yes_with_no_result_path_is_unconfirmed(self): + def vanished(path): + os.rename(path, os.path.join(self.tmp, "elsewhere.png")) + return True, None, None + + backend = self.backend(result=vanished) + with self.assertRaises(trash.TrashUnconfirmed): + backend.move(self.file, [self.root]) + + def test_yes_into_an_output_root_is_unconfirmed(self): + backend = self.backend( + result=self.moved_to(os.path.join(self.root, "x", "a.png")) + ) + with self.assertRaises(trash.TrashUnconfirmed): + backend.move(self.file, [self.root]) + + def test_yes_into_an_unplanned_folder_is_trashed_and_logged(self): + other = os.path.join(self.tmp, "other-trash", "a.png") + backend = self.backend(result=self.moved_to(other)) + with self.assertLogs("prompt_manager.trash", level="WARNING") as logs: + label = backend.move(self.file, [self.root]) + self.assertEqual(label, os.path.realpath(os.path.dirname(other))) + self.assertIn(other, "\n".join(logs.output)) + + +class IdentityTests(MacBackendTestCase): + @unittest.skipUnless(hasattr(os, "symlink") and os.name == "posix", "symlinks") + def test_root_given_through_a_symlink_still_contains_its_trash(self): + alias = os.path.join(self.tmp, "home-alias") + os.symlink(self.home, alias) + self.assertEqual( + self.backend().available(self.file, [alias]), trash.TRASH_IN_ROOT + ) + + def test_missing_roots_are_ignored(self): + missing = os.path.join(self.tmp, "nope") + self.assertIsNone(self.backend().available(self.file, [missing, self.root])) + + +class LoadingTests(unittest.TestCase): + def test_default_backend_for_darwin_does_not_load_the_bridge(self): + with mock.patch.object(trash_macos, "load_bridge") as loader: + backend = trash.default_backend("darwin") + self.assertIsInstance(backend, MacTrashBackend) + loader.assert_not_called() + + def test_failed_load_answers_no_trash_and_is_cached_and_logged_once(self): + backend = MacTrashBackend(uid=501) + with ( + mock.patch.object( + trash_macos, "load_bridge", side_effect=OSError("no Foundation") + ) as loader, + self.assertLogs("prompt_manager.trash", level="WARNING") as logs, + ): + self.assertEqual(backend.available("/x/a.png", ["/x"]), trash.NO_TRASH) + with self.assertRaises(trash.TrashRefused) as caught: + backend.move("/x/a.png", ["/x"]) + self.assertEqual(caught.exception.code, trash.NO_TRASH) + self.assertEqual(loader.call_count, 1) + self.assertEqual(len(logs.output), 1) + + def test_load_bridge_off_macos_raises(self): + if sys.platform == "darwin": + self.skipTest("the real bridge loads on macOS (native test covers it)") + with self.assertRaises(OSError): + trash_macos.load_bridge() + + +class ErrorTextTests(unittest.TestCase): + def test_detail_without_underlying(self): + self.assertEqual( + trash_macos.error_detail(NSErr(COCOA, 4)), "NSCocoaErrorDomain 4" + ) + self.assertEqual(trash_macos.error_detail(None), "no NSError") + + def test_refusal_keeps_detail(self): + refused = trash.TrashRefused(trash.FAILED, "why") + self.assertEqual( + (refused.code, refused.detail, str(refused)), ("failed", "why", "why") + ) + self.assertEqual(trash.TrashRefused(trash.FAILED).detail, "") + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_trash_macos_native.py b/tests/test_trash_macos_native.py new file mode 100644 index 00000000..f8c92a49 --- /dev/null +++ b/tests/test_trash_macos_native.py @@ -0,0 +1,191 @@ +"""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() diff --git a/tests/vendor/ci-workflow.test.mjs b/tests/vendor/ci-workflow.test.mjs index c146bb58..c1a4d08e 100644 --- a/tests/vendor/ci-workflow.test.mjs +++ b/tests/vendor/ci-workflow.test.mjs @@ -353,6 +353,20 @@ test('Windows and macOS run 3.12-3.14 on main, nightly and on demand, never on a assert.doesNotMatch(text, /cancel-in-progress: true|cancel-in-progress: \$/, 'no portability run is cancelled'); }); +// #499: the native trash proof runs on both macOS CPUs, apart from the suite so a red suite never skips it. +test('portability runs the native trash proof on arm64 and Intel macOS', () => { + const body = jobs(readWorkflow('portability.yml')).get('native-trash'); + assert.ok(body, 'portability.yml has a native-trash job'); + assert.deepEqual(matrixList(body, 'os'), ['macos-latest', 'macos-15-intel']); + assert.match(body, /^ {6}fail-fast: false$/m); + assert.doesNotMatch(body, /^ {4}needs:/m, 'it does not wait for the suite job'); + assert.match(body, /^ {10}PM_NATIVE_TRASH: macos$/m); + assert.match(body, /^ {8}run: python -m unittest tests\.test_trash_macos_native tests\.test_trash_macos -v$/m); + const jobMinutes = Number(/^ {4}timeout-minutes: (\d+)$/m.exec(body)[1]); + const stepMinutes = Number(/^ {8}timeout-minutes: (\d+)$/m.exec(body)?.[1]); + assert.ok(stepMinutes > 0 && jobMinutes - stepMinutes >= 2, 'the proof step stops before the job limit'); +}); + test('real ComfyUI runs only outside pull requests; the other test.yml checks run on every event', () => { const found = testJobs(); assert.match(found.get('comfyui'), QUEUE_ONLY_IF); diff --git a/utils/trash.py b/utils/trash.py index 242236b5..5134b3a2 100644 --- a/utils/trash.py +++ b/utils/trash.py @@ -14,8 +14,9 @@ same mount, else the file's mount's ``.Trash/$uid`` (only under a real, sticky ``.Trash``) or ``.Trash-$uid``. Names are reserved by creating the ``.trashinfo`` with ``O_EXCL`` and the file is moved with ``renameat2(RENAME_NOREPLACE)``; without that call the move is refused. -Windows and macOS have no backend yet (:class:`UnavailableBackend`): their -own PRs add one with a native proof. +macOS uses :class:`utils.trash_macos.MacTrashBackend` (#499): NSFileManager's +``trashItemAtURL`` through ctypes, after refusing what it cannot show to be +recoverable. Windows has no backend yet (:class:`UnavailableBackend`, #500). ``BACKEND`` is the one replaceable seam; tests install a fake. """ @@ -36,7 +37,20 @@ TRASH_IN_ROOT = "trash-in-root" PATH_TOO_LONG = "path-too-long" LOCKED = "locked" FAILED = "failed" -CODES = frozenset({NO_TRASH, NETWORK, TRASH_IN_ROOT, PATH_TOO_LONG, LOCKED, FAILED}) +NO_PERMISSION = "no-permission" +CLOUD_SYNC = "cloud-sync" +CODES = frozenset( + { + NO_TRASH, + NETWORK, + TRASH_IN_ROOT, + PATH_TOO_LONG, + LOCKED, + FAILED, + NO_PERMISSION, + CLOUD_SYNC, + } +) HOME_TRASH_LABEL = "home trash" RENAME_NOREPLACE = 1 @@ -61,6 +75,19 @@ class TrashRefused(Exception): def __init__(self, code: str, detail: str = ""): super().__init__(detail or code) self.code = code + self.detail = detail + + +class TrashUnconfirmed(Exception): + """The system reported the file moved, but not to a place shown to be a trash. + + The file is no longer where it was, so this is not a refusal: the caller + keeps the file's records and stops (#499 design §7.4). + """ + + def __init__(self, detail: str): + super().__init__(detail) + self.detail = detail def parse_mountinfo(text: str) -> list[str]: @@ -345,8 +372,16 @@ class FreedesktopBackend: def default_backend(platform: str = sys.platform): - """The backend for *platform*; Windows and macOS have none yet (#480).""" - if platform in ("win32", "darwin") or platform.startswith("cygwin"): + """The backend for *platform*; Windows has none yet (#500). + + macOS gets :class:`utils.trash_macos.MacTrashBackend`, which loads its + Foundation bridge only on first use (#499). + """ + if platform == "darwin": + from . import trash_macos + + return trash_macos.MacTrashBackend() + if platform == "win32" or platform.startswith("cygwin"): return UnavailableBackend() return FreedesktopBackend() diff --git a/utils/trash_macos.py b/utils/trash_macos.py new file mode 100644 index 00000000..4ce422da --- /dev/null +++ b/utils/trash_macos.py @@ -0,0 +1,489 @@ +"""The macOS Trash backend (#499): NSFileManager through ctypes, or refuse. + +``-[NSFileManager trashItemAtURL:resultingItemURL:error:]`` moves a file into +its volume's trash folder (``~/.Trash`` on the startup volume, +``/.Trashes/`` elsewhere) or fails with an ``NSError``; it renames +name clashes itself and never replaces a file. Before calling it, +:class:`MacTrashBackend` refuses what it cannot show to be recoverable: +network volumes, read-only volumes, iCloud Drive and File Provider sync +folders, a trash folder inside an output root, and files macOS would refuse +for permission. Nothing here unlinks, copies or overwrites a file. + +The :class:`Bridge` is the only code that touches the Objective-C runtime. It +sends ten fixed messages, each through ``objc_msgSend`` cast to its exact +prototype (required on arm64), checks every object for nil before a send (a +nil can raise an Objective-C exception, which would abort the process), and +runs each call inside its own autorelease pool. It is loaded lazily, on the +first ``available``/``move``; when it cannot load, every file answers +``no-trash`` as before. The facts behind these rules were measured on macOS +14, 15 (x86_64) and 26 (docs/egm/499-macos-trash/design.md §4). +""" + +from __future__ import annotations + +import ctypes +import logging +import os +import stat +import sys +import threading +from dataclasses import dataclass + +from . import trash + +logger = logging.getLogger("prompt_manager.trash") + +COCOA = "NSCocoaErrorDomain" +POSIX = "NSPOSIXErrorDomain" +FEATURE_UNSUPPORTED = 3328 +_ENOENT = 2 +# Top-level Cocoa codes of a refused move (design §7.5); anything else is "failed". +_MOVE_CODES = { + FEATURE_UNSUPPORTED: trash.NO_TRASH, + 642: trash.NO_TRASH, # NSFileWriteVolumeReadOnlyError + 513: trash.NO_PERMISSION, # NSFileWriteNoPermissionError + 257: trash.NO_PERMISSION, # NSFileReadNoPermissionError +} +_LOCK_FLAGS = stat.UF_IMMUTABLE | stat.SF_IMMUTABLE | stat.UF_APPEND | stat.SF_APPEND +_CLOUD_FOLDERS = (("Library", "Mobile Documents"), ("Library", "CloudStorage")) + + +@dataclass(frozen=True) +class NSErr: + domain: str + code: int + under_domain: str | None = None + under_code: int | None = None + + +@dataclass(frozen=True) +class Facts: + """Resource values of a file's URL; ``None`` means the lookup failed or was nil.""" + + local: bool | None + read_only: bool | None + root_fs: bool | None + volume: str | None + ubiquitous: bool | None + + +def error_detail(error: NSErr | None) -> str: + """``" ( )"`` for the log.""" + if error is None: + return "no NSError" + text = f"{error.domain} {error.code}" + if error.under_domain is not None: + text += f" ({error.under_domain} {error.under_code})" + return text + + +def refusal_code(error: NSErr | None) -> str: + if error is None or error.domain != COCOA: + return trash.FAILED + return _MOVE_CODES.get(error.code, trash.FAILED) + + +def _within(path: str, folder: str) -> bool: + path = os.path.normcase(path) + folder = os.path.normcase(folder) + trimmed = folder.rstrip("\\/") or folder + return path == trimmed or path.startswith(trimmed.rstrip(os.sep) + os.sep) + + +def _identity(path: str): + info = os.stat(path) + return info.st_dev, info.st_ino + + +def inside_any_root(path: str, roots) -> bool: + """Whether *path* (existing or not) lies in an output root, by file identity. + + Every existing folder from *path* up to the top is compared with the + roots' ``(st_dev, st_ino)``, so case, NFC/NFD and symlinked spellings of a + root all match. + """ + identities = set() + for root in roots: + try: + identities.add(_identity(str(root))) + except OSError: + continue + current = os.path.abspath(path) + while True: + try: + if _identity(current) in identities: + return True + except OSError: + pass + parent = os.path.dirname(current) + if parent == current: + return False + current = parent + + +def _same_folder(left: str, right: str) -> bool: + try: + return _identity(left) == _identity(right) + except OSError: + return os.path.realpath(left) == os.path.realpath(right) + + +class MacTrashBackend: + """``available``/``move`` for macOS; the bridge and file checks are seams.""" + + def __init__( + self, bridge=None, uid=None, home=None, lstat=os.lstat, access=os.access + ): + self._bridge = bridge + self._load_failed = False + self._lock = threading.Lock() + if uid is None: + uid = os.getuid() if hasattr(os, "getuid") else -1 + self._uid = uid + self._home = home + self._lstat = lstat + self._access = access + + def _get_bridge(self): + if self._bridge is None and not self._load_failed: + with self._lock: + if self._bridge is None and not self._load_failed: + try: + self._bridge = load_bridge() + except (OSError, AttributeError, ValueError) as error: + self._load_failed = True + logger.warning("The macOS trash is unavailable: %s", error) + if self._bridge is None: + raise trash.TrashRefused(trash.NO_TRASH, "Foundation bridge not loaded") + return self._bridge + + def _home_dir(self) -> str: + return os.path.realpath(self._home or os.path.expanduser("~")) + + def _plan(self, path: str, roots): + """``(trash folder, facts)`` for *path*; raises TrashRefused (design §7.2).""" + bridge = self._get_bridge() + real = os.path.realpath(path) + facts = bridge.facts(real) + if facts.local is not True: + raise trash.TrashRefused(trash.NETWORK) + if facts.read_only is not False: + raise trash.TrashRefused(trash.NO_TRASH, "read-only volume") + home = self._home_dir() + if facts.ubiquitous is True or any( + _within(real, os.path.join(home, *parts)) for parts in _CLOUD_FOLDERS + ): + raise trash.TrashRefused(trash.CLOUD_SYNC) + folder, error = bridge.trash_dir(real) + if folder is None: + folder = self._predict(facts, error) + if inside_any_root(folder, roots): + raise trash.TrashRefused(trash.TRASH_IN_ROOT) + self._check_permission(real) + return folder, facts + + def _predict(self, facts: Facts, error: NSErr | None) -> str: + """``/.Trashes/`` for a volume whose trash is not made yet.""" + missing = ( + error is not None + and error.domain == COCOA + and error.code == FEATURE_UNSUPPORTED + and error.under_domain == POSIX + and error.under_code == _ENOENT + ) + if not missing or facts.root_fs is not False or not facts.volume: + raise trash.TrashRefused(trash.NO_TRASH, error_detail(error)) + trashes = os.path.join(facts.volume, ".Trashes") + try: + info = self._lstat(trashes) + except FileNotFoundError: + info = None + except OSError as failure: + raise trash.TrashRefused(trash.NO_TRASH, str(failure)) from failure + if info is not None and not stat.S_ISDIR(info.st_mode): + raise trash.TrashRefused(trash.NO_TRASH, ".Trashes is not a folder") + return os.path.join(trashes, str(self._uid)) + + def _check_permission(self, real: str) -> None: + parent = os.path.dirname(real) + if not self._access(parent, os.W_OK | os.X_OK): + raise trash.TrashRefused(trash.NO_PERMISSION, "folder not writable") + try: + file_info = self._lstat(real) + parent_info = self._lstat(parent) + except OSError as failure: + raise trash.TrashRefused(trash.FAILED, str(failure)) from failure + for info in (file_info, parent_info): + if getattr(info, "st_flags", 0) & _LOCK_FLAGS: + raise trash.TrashRefused(trash.NO_PERMISSION, "immutable flag") + if parent_info.st_mode & stat.S_ISVTX and file_info.st_uid != self._uid: + raise trash.TrashRefused(trash.NO_PERMISSION, "sticky folder, foreign file") + + def _label(self, folder: str, facts: Facts) -> str: + if _same_folder(folder, os.path.join(self._home_dir(), ".Trash")): + return trash.HOME_TRASH_LABEL + real = os.path.realpath(folder) + volume = os.path.realpath(facts.volume) if facts.volume else None + if volume and volume != os.path.realpath(os.sep) and _within(real, volume): + relative = os.path.relpath(real, volume).replace(os.sep, "/") + return f"{relative} on its drive" + return real + + def available(self, path: str, roots) -> str | None: + """None when *path* can be trashed, else the refusal code (no writes).""" + try: + self._plan(path, roots) + except trash.TrashRefused as refused: + return refused.code + except OSError: + return trash.NO_TRASH + return None + + def move(self, path: str, roots) -> str: + """Move *path* into its trash folder and return the label (design §7.4).""" + try: + planned, facts = self._plan(path, roots) + except OSError as failure: + raise trash.TrashRefused(trash.NO_TRASH, str(failure)) from failure + real = os.path.realpath(path) + ok, resulting, error = self._get_bridge().trash(real) + if not ok: + raise trash.TrashRefused(refusal_code(error), error_detail(error)) + if os.path.lexists(real): + raise trash.TrashRefused(trash.FAILED, "reported moved, still in place") + if not resulting or inside_any_root(os.path.dirname(resulting), roots): + raise trash.TrashUnconfirmed(f"{real} -> {resulting!r}") + folder = os.path.dirname(resulting) + if not _same_folder(folder, planned): + logger.warning( + "macOS trashed %s to %s, not the planned %s", real, resulting, planned + ) + return self._label(folder, facts) + + +# ── the Foundation bridge ──────────────────────────────────────────────── + +_NS_TRASH_DIRECTORY = 102 +_NS_USER_DOMAIN_MASK = 1 +_LIBOBJC = "/usr/lib/libobjc.A.dylib" +_FOUNDATION = "/System/Library/Frameworks/Foundation.framework/Foundation" +_KEYS = ( + "NSURLVolumeIsLocalKey", + "NSURLVolumeIsReadOnlyKey", + "NSURLVolumeIsRootFileSystemKey", + "NSURLVolumeURLKey", + "NSURLIsUbiquitousItemKey", + "NSUnderlyingErrorKey", +) +_REQUIRED_SELECTORS = ( + ("NSURL", "fileURLWithFileSystemRepresentation:isDirectory:relativeToURL:"), + ("NSURL", "getResourceValue:forKey:error:"), + ("NSFileManager", "URLForDirectory:inDomain:appropriateForURL:create:error:"), + ("NSFileManager", "trashItemAtURL:resultingItemURL:error:"), +) + + +class _Nil(Exception): + """An Objective-C object was nil where a send needs one.""" + + +class Bridge: # pragma: no cover - macOS only; the native-trash job runs it + """Foundation through ctypes: ``facts``, ``trash_dir`` and ``trash`` (design §6).""" + + def __init__(self): + if sys.platform != "darwin": + raise OSError("the Foundation bridge exists only on macOS") + objc = ctypes.CDLL(_LIBOBJC) + foundation = ctypes.CDLL(_FOUNDATION) + void_p, char_p, byte = ctypes.c_void_p, ctypes.c_char_p, ctypes.c_byte + objc.objc_getClass.argtypes = [char_p] + objc.objc_getClass.restype = void_p + objc.sel_registerName.argtypes = [char_p] + objc.sel_registerName.restype = void_p + objc.objc_autoreleasePoolPush.argtypes = [] + objc.objc_autoreleasePoolPush.restype = void_p + objc.objc_autoreleasePoolPop.argtypes = [void_p] + objc.objc_autoreleasePoolPop.restype = None + self._objc = objc + address = ctypes.cast(objc.objc_msgSend, void_p).value + + def send(restype, *argtypes): + return ctypes.CFUNCTYPE(restype, void_p, void_p, *argtypes)(address) + + out = ctypes.POINTER(void_p) + self._send_id = send(void_p) + self._send_responds = send(byte, void_p) + self._send_url = send(void_p, char_p, byte, void_p) + self._send_resource = send(byte, out, void_p, out) + self._send_bool = send(byte) + self._send_dir = send(void_p, ctypes.c_ulong, ctypes.c_ulong, void_p, byte, out) + self._send_trash = send(byte, void_p, out, out) + self._send_cstr = send(char_p) + self._send_long = send(ctypes.c_long) + self._send_object = send(void_p, void_p) + self._keys = {} + for name in _KEYS: + value = void_p.in_dll(foundation, name).value + if not value: + raise ValueError(f"Foundation constant {name} is nil") + self._keys[name] = value + for class_name, selector in _REQUIRED_SELECTORS: + if not self._send_responds( + self._class(class_name), + self._sel("respondsToSelector:"), + self._sel(selector), + ) and not self._instances_respond(class_name, selector): + raise ValueError(f"{class_name} does not respond to {selector}") + + # helpers (call inside a pool) ------------------------------------------------ + + def _class(self, name: str): + value = self._objc.objc_getClass(name.encode()) + if not value: + raise ValueError(f"no Objective-C class {name}") + return value + + def _sel(self, name: str): + return self._objc.sel_registerName(name.encode()) + + def _instances_respond(self, class_name: str, selector: str) -> bool: + return ( + self._send_responds( + self._class(class_name), + self._sel("instancesRespondToSelector:"), + self._sel(selector), + ) + != 0 + ) + + @staticmethod + def _need(value): + if not value: + raise _Nil() + return value + + def _manager(self): + return self._need( + self._send_id(self._class("NSFileManager"), self._sel("defaultManager")) + ) + + def _url(self, path: str): + return self._need( + self._send_url( + self._class("NSURL"), + self._sel( + "fileURLWithFileSystemRepresentation:isDirectory:relativeToURL:" + ), + os.fsencode(path), + 0, + None, + ) + ) + + def _path(self, url) -> str | None: + if not url: + return None + raw = self._send_cstr(url, self._sel("fileSystemRepresentation")) + return os.fsdecode(raw) if raw else None + + def _text(self, obj) -> str | None: + if not obj: + return None + raw = self._send_cstr(obj, self._sel("UTF8String")) + return raw.decode("utf-8", "replace") if raw else None + + def _error(self, error) -> NSErr | None: + if not error: + return None + domain = self._text(self._send_id(error, self._sel("domain"))) or "?" + code = self._send_long(error, self._sel("code")) + under_domain = under_code = None + info = self._send_id(error, self._sel("userInfo")) + if info: + under = self._send_object( + info, self._sel("objectForKey:"), self._keys["NSUnderlyingErrorKey"] + ) + if under: + under_domain = self._text(self._send_id(under, self._sel("domain"))) + under_code = self._send_long(under, self._sel("code")) + return NSErr(domain, code, under_domain, under_code) + + def _resource(self, url, key_name: str): + value, error = ctypes.c_void_p(), ctypes.c_void_p() + ok = self._send_resource( + url, + self._sel("getResourceValue:forKey:error:"), + ctypes.byref(value), + self._keys[key_name], + ctypes.byref(error), + ) + return value.value if ok != 0 and value.value else None + + def _flag(self, url, key_name: str) -> bool | None: + value = self._resource(url, key_name) + if value is None: + return None + return self._send_bool(value, self._sel("boolValue")) != 0 + + def _pooled(self, work): + pool = self._objc.objc_autoreleasePoolPush() + try: + return work() + except _Nil as nil: + raise trash.TrashRefused(trash.NO_TRASH, "nil Objective-C object") from nil + finally: + self._objc.objc_autoreleasePoolPop(pool) + + # the bridge API (design §6.3) ---------------------------------------------- + + def facts(self, path: str) -> Facts: + def work(): + url = self._url(path) + return Facts( + local=self._flag(url, "NSURLVolumeIsLocalKey"), + read_only=self._flag(url, "NSURLVolumeIsReadOnlyKey"), + root_fs=self._flag(url, "NSURLVolumeIsRootFileSystemKey"), + volume=self._path(self._resource(url, "NSURLVolumeURLKey")), + ubiquitous=self._flag(url, "NSURLIsUbiquitousItemKey"), + ) + + return self._pooled(work) + + def trash_dir(self, path: str): + def work(): + error = ctypes.c_void_p() + folder = self._send_dir( + self._manager(), + self._sel("URLForDirectory:inDomain:appropriateForURL:create:error:"), + _NS_TRASH_DIRECTORY, + _NS_USER_DOMAIN_MASK, + self._url(path), + 0, + ctypes.byref(error), + ) + if folder: + return self._path(folder), None + return None, self._error(error.value) + + return self._pooled(work) + + def trash(self, path: str): + def work(): + resulting, error = ctypes.c_void_p(), ctypes.c_void_p() + ok = self._send_trash( + self._manager(), + self._sel("trashItemAtURL:resultingItemURL:error:"), + self._url(path), + ctypes.byref(resulting), + ctypes.byref(error), + ) + if ok != 0: + return True, self._path(resulting.value), None + return False, None, self._error(error.value) + + return self._pooled(work) + + +def load_bridge() -> Bridge: + """The real Foundation bridge; raises OSError/ValueError where it cannot load.""" + return Bridge()