"""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)