92 lines
3.2 KiB
Python
92 lines
3.2 KiB
Python
#!/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:]]))
|