243 lines
8.2 KiB
Python
243 lines
8.2 KiB
Python
|
"""Tests for file_watcher module."""
|
||
|
|
|
||
|
|
import time
|
||
|
|
from pathlib import Path
|
||
|
|
from threading import Event
|
||
|
|
|
||
|
|
import pytest
|
||
|
|
|
||
|
|
from src.core.file_watcher import EventType, FileEvent, FileWatcher
|
||
|
|
|
||
|
|
|
||
|
|
class TestFileEvent:
|
||
|
|
"""Tests for FileEvent dataclass."""
|
||
|
|
|
||
|
|
def test_create_file_event(self) -> None:
|
||
|
|
event = FileEvent(
|
||
|
|
event_type=EventType.CREATED,
|
||
|
|
path="test.txt",
|
||
|
|
is_directory=False,
|
||
|
|
)
|
||
|
|
assert event.event_type == EventType.CREATED
|
||
|
|
assert event.path == "test.txt"
|
||
|
|
assert event.is_directory is False
|
||
|
|
assert event.dest_path is None
|
||
|
|
|
||
|
|
def test_move_event_with_dest(self) -> None:
|
||
|
|
event = FileEvent(
|
||
|
|
event_type=EventType.MOVED,
|
||
|
|
path="old.txt",
|
||
|
|
is_directory=False,
|
||
|
|
dest_path="new.txt",
|
||
|
|
)
|
||
|
|
assert event.event_type == EventType.MOVED
|
||
|
|
assert event.path == "old.txt"
|
||
|
|
assert event.dest_path == "new.txt"
|
||
|
|
|
||
|
|
def test_str_representation(self) -> None:
|
||
|
|
event = FileEvent(EventType.CREATED, "test.txt", False)
|
||
|
|
assert str(event) == "created: test.txt"
|
||
|
|
|
||
|
|
def test_str_representation_moved(self) -> None:
|
||
|
|
event = FileEvent(EventType.MOVED, "old.txt", False, "new.txt")
|
||
|
|
assert str(event) == "moved: old.txt -> new.txt"
|
||
|
|
|
||
|
|
|
||
|
|
class TestFileWatcher:
|
||
|
|
"""Tests for FileWatcher class."""
|
||
|
|
|
||
|
|
def test_start_and_stop(self, tmp_path: Path) -> None:
|
||
|
|
events: list[FileEvent] = []
|
||
|
|
watcher = FileWatcher(tmp_path, callback=events.append)
|
||
|
|
|
||
|
|
assert not watcher.is_running()
|
||
|
|
watcher.start()
|
||
|
|
assert watcher.is_running()
|
||
|
|
watcher.stop()
|
||
|
|
assert not watcher.is_running()
|
||
|
|
|
||
|
|
def test_context_manager(self, tmp_path: Path) -> None:
|
||
|
|
events: list[FileEvent] = []
|
||
|
|
|
||
|
|
with FileWatcher(tmp_path, callback=events.append) as watcher:
|
||
|
|
assert watcher.is_running()
|
||
|
|
|
||
|
|
assert not watcher.is_running()
|
||
|
|
|
||
|
|
def test_start_nonexistent_path_raises(self, tmp_path: Path) -> None:
|
||
|
|
nonexistent = tmp_path / "nonexistent"
|
||
|
|
events: list[FileEvent] = []
|
||
|
|
watcher = FileWatcher(nonexistent, callback=events.append)
|
||
|
|
|
||
|
|
with pytest.raises(FileNotFoundError):
|
||
|
|
watcher.start()
|
||
|
|
|
||
|
|
def test_double_start_is_safe(self, tmp_path: Path) -> None:
|
||
|
|
events: list[FileEvent] = []
|
||
|
|
watcher = FileWatcher(tmp_path, callback=events.append)
|
||
|
|
|
||
|
|
watcher.start()
|
||
|
|
watcher.start() # Should not raise
|
||
|
|
assert watcher.is_running()
|
||
|
|
watcher.stop()
|
||
|
|
|
||
|
|
def test_double_stop_is_safe(self, tmp_path: Path) -> None:
|
||
|
|
events: list[FileEvent] = []
|
||
|
|
watcher = FileWatcher(tmp_path, callback=events.append)
|
||
|
|
|
||
|
|
watcher.start()
|
||
|
|
watcher.stop()
|
||
|
|
watcher.stop() # Should not raise
|
||
|
|
assert not watcher.is_running()
|
||
|
|
|
||
|
|
def test_detects_file_creation(self, tmp_path: Path) -> None:
|
||
|
|
events: list[FileEvent] = []
|
||
|
|
event_received = Event()
|
||
|
|
|
||
|
|
def callback(event: FileEvent) -> None:
|
||
|
|
events.append(event)
|
||
|
|
event_received.set()
|
||
|
|
|
||
|
|
with FileWatcher(tmp_path, callback=callback):
|
||
|
|
# Create a file
|
||
|
|
test_file = tmp_path / "test.txt"
|
||
|
|
test_file.write_text("hello")
|
||
|
|
|
||
|
|
# Wait for event
|
||
|
|
event_received.wait(timeout=2.0)
|
||
|
|
|
||
|
|
# Check that we got a CREATED event
|
||
|
|
created_events = [e for e in events if e.event_type == EventType.CREATED]
|
||
|
|
assert len(created_events) >= 1
|
||
|
|
assert any(e.path == "test.txt" for e in created_events)
|
||
|
|
|
||
|
|
def test_detects_file_deletion(self, tmp_path: Path) -> None:
|
||
|
|
# Create file first
|
||
|
|
test_file = tmp_path / "test.txt"
|
||
|
|
test_file.write_text("hello")
|
||
|
|
|
||
|
|
events: list[FileEvent] = []
|
||
|
|
event_received = Event()
|
||
|
|
|
||
|
|
def callback(event: FileEvent) -> None:
|
||
|
|
events.append(event)
|
||
|
|
if event.event_type == EventType.DELETED:
|
||
|
|
event_received.set()
|
||
|
|
|
||
|
|
with FileWatcher(tmp_path, callback=callback):
|
||
|
|
# Delete the file
|
||
|
|
test_file.unlink()
|
||
|
|
|
||
|
|
# Wait for event
|
||
|
|
event_received.wait(timeout=2.0)
|
||
|
|
|
||
|
|
# Check that we got a DELETED event
|
||
|
|
deleted_events = [e for e in events if e.event_type == EventType.DELETED]
|
||
|
|
assert len(deleted_events) >= 1
|
||
|
|
assert any(e.path == "test.txt" for e in deleted_events)
|
||
|
|
|
||
|
|
def test_detects_file_move(self, tmp_path: Path) -> None:
|
||
|
|
# Create file first
|
||
|
|
test_file = tmp_path / "old.txt"
|
||
|
|
test_file.write_text("hello")
|
||
|
|
|
||
|
|
events: list[FileEvent] = []
|
||
|
|
event_received = Event()
|
||
|
|
|
||
|
|
def callback(event: FileEvent) -> None:
|
||
|
|
events.append(event)
|
||
|
|
if event.event_type == EventType.MOVED:
|
||
|
|
event_received.set()
|
||
|
|
|
||
|
|
with FileWatcher(tmp_path, callback=callback):
|
||
|
|
# Move the file
|
||
|
|
new_file = tmp_path / "new.txt"
|
||
|
|
test_file.rename(new_file)
|
||
|
|
|
||
|
|
# Wait for event
|
||
|
|
event_received.wait(timeout=2.0)
|
||
|
|
|
||
|
|
# Check that we got a MOVED event
|
||
|
|
moved_events = [e for e in events if e.event_type == EventType.MOVED]
|
||
|
|
assert len(moved_events) >= 1
|
||
|
|
assert any(e.path == "old.txt" and e.dest_path == "new.txt" for e in moved_events)
|
||
|
|
|
||
|
|
def test_ignores_vault_directory(self, tmp_path: Path) -> None:
|
||
|
|
# Create .vault directory
|
||
|
|
vault_dir = tmp_path / ".vault"
|
||
|
|
vault_dir.mkdir()
|
||
|
|
|
||
|
|
events: list[FileEvent] = []
|
||
|
|
|
||
|
|
with FileWatcher(tmp_path, callback=events.append):
|
||
|
|
# Create file inside .vault
|
||
|
|
(vault_dir / "manifest.json").write_text("{}")
|
||
|
|
time.sleep(0.5)
|
||
|
|
|
||
|
|
# No events should be recorded for .vault directory
|
||
|
|
assert all(".vault" not in e.path for e in events)
|
||
|
|
|
||
|
|
def test_custom_ignore_patterns(self, tmp_path: Path) -> None:
|
||
|
|
events: list[FileEvent] = []
|
||
|
|
event_received = Event()
|
||
|
|
|
||
|
|
def callback(event: FileEvent) -> None:
|
||
|
|
events.append(event)
|
||
|
|
event_received.set()
|
||
|
|
|
||
|
|
with FileWatcher(tmp_path, callback=callback, ignore_patterns=[".vault", "__pycache__"]):
|
||
|
|
# Create ignored directory
|
||
|
|
cache_dir = tmp_path / "__pycache__"
|
||
|
|
cache_dir.mkdir()
|
||
|
|
(cache_dir / "test.pyc").write_text("cached")
|
||
|
|
time.sleep(0.2)
|
||
|
|
|
||
|
|
# Create non-ignored file
|
||
|
|
(tmp_path / "regular.txt").write_text("hello")
|
||
|
|
event_received.wait(timeout=2.0)
|
||
|
|
|
||
|
|
# Only regular.txt events should be recorded
|
||
|
|
assert all("__pycache__" not in e.path for e in events)
|
||
|
|
assert any("regular.txt" in e.path for e in events)
|
||
|
|
|
||
|
|
def test_detects_nested_file_creation(self, tmp_path: Path) -> None:
|
||
|
|
# Create nested directory
|
||
|
|
nested = tmp_path / "subdir" / "nested"
|
||
|
|
nested.mkdir(parents=True)
|
||
|
|
|
||
|
|
events: list[FileEvent] = []
|
||
|
|
event_received = Event()
|
||
|
|
|
||
|
|
def callback(event: FileEvent) -> None:
|
||
|
|
events.append(event)
|
||
|
|
if event.event_type == EventType.CREATED and "deep.txt" in event.path:
|
||
|
|
event_received.set()
|
||
|
|
|
||
|
|
with FileWatcher(tmp_path, callback=callback):
|
||
|
|
# Create file in nested directory
|
||
|
|
(nested / "deep.txt").write_text("nested content")
|
||
|
|
event_received.wait(timeout=2.0)
|
||
|
|
|
||
|
|
# Check event has correct relative path
|
||
|
|
created_events = [e for e in events if e.event_type == EventType.CREATED]
|
||
|
|
assert any("subdir/nested/deep.txt" in e.path or "subdir\\nested\\deep.txt" in e.path for e in created_events)
|
||
|
|
|
||
|
|
def test_detects_directory_creation(self, tmp_path: Path) -> None:
|
||
|
|
events: list[FileEvent] = []
|
||
|
|
event_received = Event()
|
||
|
|
|
||
|
|
def callback(event: FileEvent) -> None:
|
||
|
|
events.append(event)
|
||
|
|
if event.is_directory and event.event_type == EventType.CREATED:
|
||
|
|
event_received.set()
|
||
|
|
|
||
|
|
with FileWatcher(tmp_path, callback=callback):
|
||
|
|
# Create directory
|
||
|
|
(tmp_path / "newdir").mkdir()
|
||
|
|
event_received.wait(timeout=2.0)
|
||
|
|
|
||
|
|
# Check directory creation event
|
||
|
|
dir_events = [e for e in events if e.is_directory and e.event_type == EventType.CREATED]
|
||
|
|
assert len(dir_events) >= 1
|
||
|
|
assert any(e.path == "newdir" for e in dir_events)
|