feat(monitor_dir): Add path/inode polling mode option

monitor_dir_test.py

@@ -0,0 +1,91 @@
+#!/usr/bin/env -S uv run --script
+# /// script
+# dependencies = ["pytest", "watchdog"]
+# ///
+
+from pathlib import Path
+import stat
+import sys
+from types import SimpleNamespace
+from unittest.mock import Mock, call, patch
+
+import pytest
+
+import monitor_dir as script
+
+
+@pytest.mark.parametrize("recursive", [False, True])
+def test_recursive_scan_does_not_follow_symlinks(tmp_path: Path, recursive: bool) -> None:
+    child = tmp_path / "child"
+    child.mkdir()
+    file_path = child / "file.txt"
+    file_path.touch()
+    link = tmp_path / "loop"
+    link.symlink_to(target=tmp_path, target_is_directory=True)
+
+    expected = {child: True, link: False}
+    if recursive:
+        expected[file_path] = False
+    assert script.take_snapshot(base_dir=tmp_path, recursive=recursive) == expected
+
+
+def test_changing_inode_numbers_do_not_emit_events(tmp_path: Path) -> None:
+    file_path = tmp_path / "file.txt"
+    file_path.touch()
+    handler = Mock()
+    with patch.object(
+        target=Path, attribute="lstat",
+        side_effect=[
+            SimpleNamespace(st_mode=stat.S_IFREG, st_ino=1),
+            SimpleNamespace(st_mode=stat.S_IFREG, st_ino=999),
+        ],
+    ):
+        previous = script.take_snapshot(base_dir=tmp_path, recursive=False)
+        current = script.take_snapshot(base_dir=tmp_path, recursive=False)
+
+    script.dispatch_changes(previous=previous, current=current, event_handler=handler)
+    handler._process_event_type.assert_not_called()
+
+
+def test_rename_emits_deleted_and_created() -> None:
+    source = Path("/watched/before.txt")
+    destination = Path("/watched/after.txt")
+    handler = Mock()
+    script.dispatch_changes(previous={source: False}, current={destination: False}, event_handler=handler)
+    assert handler._process_event_type.call_args_list == [
+        call(event_name="deleted", path_to_filter_on=source, src_path=source),
+        call(event_name="created", path_to_filter_on=destination, src_path=destination),
+    ]
+
+
+def test_failed_scan_preserves_last_snapshot() -> None:
+    base_dir = Path("/watched")
+    existing = base_dir / "existing.txt"
+    created = base_dir / "created.txt"
+    handler = Mock()
+    with patch.object(
+        target=script, attribute="take_snapshot",
+        side_effect=[{existing: False}, OSError("SSHFS disconnected"), {existing: False, created: False}],
+    ), patch.object(
+        target=script.time, attribute="sleep", side_effect=[None, None, KeyboardInterrupt],
+    ):
+        with pytest.raises(KeyboardInterrupt):
+            script.monitor_paths(base_dir=base_dir, recursive=True, interval=1, event_handler=handler)
+    handler._process_event_type.assert_called_once_with(
+        event_name="created", path_to_filter_on=created, src_path=created,
+    )
+
+
+def test_partial_scan_is_discarded(tmp_path: Path) -> None:
+    (tmp_path / "first.txt").touch()
+    (tmp_path / "second.txt").touch()
+    with patch.object(
+        target=Path, attribute="lstat",
+        side_effect=[SimpleNamespace(st_mode=stat.S_IFREG), PermissionError("scan interrupted")],
+    ):
+        with pytest.raises(PermissionError):
+            script.take_snapshot(base_dir=tmp_path, recursive=True)
+
+
+if __name__ == "__main__":
+    sys.exit(pytest.main(args=[str(Path(__file__).resolve()), *sys.argv[1:]]))