diff --git a/src/basic_memory/sync/file_change_scanner.py b/src/basic_memory/sync/file_change_scanner.py index 7125897f..ed4a6418 100644 --- a/src/basic_memory/sync/file_change_scanner.py +++ b/src/basic_memory/sync/file_change_scanner.py @@ -2,14 +2,14 @@ from dataclasses import dataclass, field from pathlib import Path -from typing import Dict, Sequence +from typing import Dict, Sequence, Optional from loguru import logger from basic_memory.file_utils import compute_checksum from basic_memory.models import Entity from basic_memory.repository.entity_repository import EntityRepository -from basic_memory.sync.utils import SyncReport +from basic_memory.sync.utils import SyncReport, FileChange @dataclass @@ -87,13 +87,19 @@ class FileChangeScanner: return result async def find_changes( - self, directory: Path, db_file_state: Dict[str, FileState] + self, + db_file_state: Dict[str, FileState], + directory: Optional[Path] = None, ) -> SyncReport: """Find changes between filesystem and database.""" # Get current files and checksums - scan_result = await self.scan_directory(directory) - current_files = scan_result.files + # scan the directory provided + scan_result = await self.scan_directory(directory) + + # the set of all of the current files and their checksums + current_files = scan_result.files + # Build report report = SyncReport() @@ -157,5 +163,6 @@ class FileChangeScanner: async def find_knowledge_changes(self, directory: Path) -> SyncReport: """Find changes in knowledge directory.""" + db_file_state = await self.get_db_file_state(await self.entity_repository.find_all()) return await self.find_changes(directory=directory, db_file_state=db_file_state) diff --git a/src/basic_memory/sync/sync_service.py b/src/basic_memory/sync/sync_service.py index 94225288..fb7ae0b5 100644 --- a/src/basic_memory/sync/sync_service.py +++ b/src/basic_memory/sync/sync_service.py @@ -61,28 +61,10 @@ class SyncService: else: logger.debug(f"No entity found to delete: {file_path}") - async def sync(self, directory: Optional[Path] = None, file_changes: Optional[dict[str, FileChange]] = None) -> SyncReport: - """Sync knowledge files with database.""" - if file_changes is not None: - changes = SyncReport() - for path, file_change in file_changes.items(): - logger.debug(f"path {path} file_change {file_change}") - match file_change.change_type: - case Change.added: - changes.new.add(path) - changes.checksums[path] = file_change.checksum - case Change.modified: - changes.modified.add(path) - changes.checksums[path] = file_change.checksum - case Change.deleted: - changes.deleted.add(path) - else: - # Traditional directory scan mode - if directory is None: - raise ValueError("Must provide either directory or file_changes") - - changes = await self.scanner.find_knowledge_changes(directory) - logger.info(f"Found {changes.total_changes} knowledge changes") + async def sync(self, directory: Path) -> SyncReport: + """Sync knowledge files with database.""" + changes = await self.scanner.find_knowledge_changes(directory) + logger.info(f"Found {changes.total_changes} knowledge changes") # Handle moves first for old_path, new_path in changes.moves.items(): diff --git a/src/basic_memory/sync/utils.py b/src/basic_memory/sync/utils.py index fcaf7dbd..57d917ec 100644 --- a/src/basic_memory/sync/utils.py +++ b/src/basic_memory/sync/utils.py @@ -55,5 +55,10 @@ class SyncReport: @property def total_changes(self) -> int: - """Total number of files that need attention.""" + """Total number of changes.""" return len(self.new) + len(self.modified) + len(self.deleted) + len(self.moves) + + @property + def total_files(self) -> int: + """Total number of files synced.""" + return len(self.new) + len(self.modified) + len(self.moves) diff --git a/src/basic_memory/sync/watch_service.py b/src/basic_memory/sync/watch_service.py index 6047085d..096219aa 100644 --- a/src/basic_memory/sync/watch_service.py +++ b/src/basic_memory/sync/watch_service.py @@ -20,7 +20,7 @@ from basic_memory.sync.utils import FileChange class WatchEvent(BaseModel): timestamp: datetime path: str - action: str # sync, delete, etc + action: str # new, delete, etc status: str # success, error error: Optional[str] = None @@ -32,8 +32,6 @@ class WatchServiceState(BaseModel): pid: int = dataclasses.field(default_factory=os.getpid) # Stats - files_synced: int = 0 - bytes_processed: int = 0 error_count: int = 0 last_error: Optional[datetime] = None last_scan: Optional[datetime] = None @@ -54,6 +52,7 @@ class WatchServiceState(BaseModel): def record_error(self, error: str): self.error_count += 1 + self.add_event(path="", action="sync", status="error", error=error) self.last_error = datetime.now() @@ -79,7 +78,7 @@ class WatchService: debounce=self.config.sync_delay, recursive=True, ): - await self.handle_changes(changes) + await self.handle_changes(self.config.home) except Exception as e: self.state.record_error(str(e)) @@ -97,35 +96,28 @@ class WatchService: """Filter to only watch markdown files""" return path.endswith(".md") and not Path(path).name.startswith(".") - async def handle_changes(self, changes: set[tuple[Change, str]]): + async def handle_changes(self, directory: Path): """Process a batch of file changes""" - # Group changes by file path - changes_by_file = {} try: - for change_type, path in changes: - file_change = await FileChange.from_path(path, change_type, self.file_service) - - # store changes by relative path - changes_by_file[str(file_change.path)] = file_change - # Process changes with timeout - await self.sync_service.sync(file_changes=changes_by_file) + report = await self.sync_service.sync(directory) + self.state.last_scan = datetime.now() + self.state.total_files = report.total_files # Update stats - self.state.files_synced += len(changes_by_file) - for path, change in changes_by_file.items(): - if change.change_type != Change.deleted: - size = self.file_service.path(path,absolute=True).stat().st_size - self.state.bytes_processed += size - - self.state.add_event(path=path, action="sync", status="success") + for path in report.new: + self.state.add_event(path=path, action="new", status="success") + for path in report.modified: + self.state.add_event(path=path, action="modified", status="success") + for path in report.moves: + self.state.add_event(path=path, action="moved", status="success") + for path in report.deleted: + self.state.add_event(path=path, action="deleted", status="success") await self.write_status() except Exception as e: self.state.record_error(str(e)) - for path in changes_by_file: - self.state.add_event(path=path, action="sync", status="error", error=str(e)) await self.write_status() raise diff --git a/tests/conftest.py b/tests/conftest.py index 3ec72d32..58c96da5 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -28,6 +28,7 @@ from basic_memory.services.link_resolver import LinkResolver from basic_memory.services.search_service import SearchService from basic_memory.sync import FileChangeScanner from basic_memory.sync.sync_service import SyncService +from basic_memory.sync.watch_service import WatchService @pytest_asyncio.fixture @@ -382,3 +383,13 @@ async def test_graph( "observations": [e.observations for e in entities], "relations": relations, } + + +@pytest_asyncio.fixture +def watch_service(sync_service, file_service, test_config): + return WatchService( + sync_service=sync_service, + file_service=file_service, + config=test_config + ) + diff --git a/tests/sync/test_watch_service.py b/tests/sync/test_watch_service.py index 6026a905..a3e569b4 100644 --- a/tests/sync/test_watch_service.py +++ b/tests/sync/test_watch_service.py @@ -1,27 +1,15 @@ """Tests for the watch service.""" -from dataclasses import asdict import pytest import pytest_asyncio from datetime import datetime -from pathlib import Path from watchfiles import Change -from basic_memory.sync.utils import FileChange -from basic_memory.sync.watch_service import WatchService, WatchEvent, WatchServiceState +from basic_memory.sync.watch_service import WatchServiceState @pytest_asyncio.fixture -def watch_service(sync_service, file_service, test_config): - return WatchService( - sync_service=sync_service, - file_service=file_service, - config=test_config - ) - - -@pytest_asyncio.fixture -async def sample_markdown(tmp_path): +async def sample_markdown_file(test_config): content = """--- title: Test Note type: note @@ -29,59 +17,82 @@ type: note # Test Note This is a test note. """ - file_path = tmp_path / "test.md" + file_path = test_config.home / "test.md" file_path.write_text(content) return file_path @pytest.mark.asyncio -async def test_handle_file_added(watch_service, sync_service, sample_markdown): +async def test_handle_file_added(test_config, watch_service, sync_service, sample_markdown_file): """Test handling a new file event""" - changes = {(Change.added, str(sample_markdown))} - - await watch_service.handle_changes(changes) - + + await watch_service.handle_changes(test_config.home) + # Check stats updated - assert watch_service.state.files_synced == 1 - assert watch_service.state.bytes_processed > 0 - + assert watch_service.state.total_files == 1 + assert watch_service.state.last_scan is not None + # Check event recorded assert len(watch_service.state.recent_events) == 1 event = watch_service.state.recent_events[0] assert event.path == "test.md" - assert event.action == "sync" + assert event.action == "new" assert event.status == "success" @pytest.mark.asyncio -async def test_handle_file_modified(watch_service, sync_service, sample_markdown): +async def test_handle_file_modified(test_config, watch_service, sync_service, sample_markdown_file): """Test handling a modified file event""" # First add the file - await watch_service.handle_changes({(Change.added, str(sample_markdown))}) - + await watch_service.handle_changes(test_config.home) + # Modify the file - sample_markdown.write_text(sample_markdown.read_text() + "\nModified content") - await watch_service.handle_changes({(Change.modified, str(sample_markdown))}) - + sample_markdown_file.write_text(sample_markdown_file.read_text() + "\nModified content") + await watch_service.handle_changes(test_config.home) + # Should have two events assert len(watch_service.state.recent_events) == 2 - assert watch_service.state.files_synced == 2 + assert watch_service.state.total_files == 1 + event = watch_service.state.recent_events[0] + assert event.path == "test.md" + assert event.action == "modified" + assert event.status == "success" @pytest.mark.asyncio -async def test_handle_file_deleted(watch_service, sync_service, sample_markdown): +async def test_handle_file_moved(test_config, watch_service, sync_service, sample_markdown_file): + """Test handling a moved file event""" + # First add the file + await watch_service.handle_changes(test_config.home) + + # Modify the file + renamed = sample_markdown_file.rename(test_config.home / "moved.md") + + await watch_service.handle_changes(test_config.home) + + # Should have two events + assert len(watch_service.state.recent_events) == 2 + assert watch_service.state.total_files == 1 + event = watch_service.state.recent_events[0] + assert event.path == "test.md" + assert event.action == "moved" + assert event.status == "success" + + +@pytest.mark.asyncio +async def test_handle_file_deleted(test_config, watch_service, sync_service, sample_markdown_file): """Test handling a deleted file event""" # First add the file - await watch_service.handle_changes({(Change.added, str(sample_markdown))}) - + await watch_service.handle_changes(test_config.home) + # Delete the file - sample_markdown.unlink() - await watch_service.handle_changes({(Change.deleted, str(sample_markdown))}) - + sample_markdown_file.unlink() + await watch_service.handle_changes(test_config.home) + # Should have two events assert len(watch_service.state.recent_events) == 2 delete_event = watch_service.state.recent_events[0] - assert delete_event.action == "sync" + assert delete_event.action == "deleted" assert delete_event.path == "test.md" @@ -93,30 +104,12 @@ async def test_filter_changes(watch_service): assert watch_service.filter_changes(Change.added, ".test.md") is False -@pytest.mark.asyncio -async def test_handle_error(watch_service, sync_service, ): - """Test error handling during sync""" - - changes = {(Change.added, "test.md")} - - # ValueError raised because "test.md" path does not exist - with pytest.raises(ValueError): - await watch_service.handle_changes(changes) - - assert watch_service.state.error_count == 1 - assert watch_service.state.last_error is not None - - @pytest.mark.asyncio async def test_recent_events_limit(watch_service): """Test that recent events are limited to 100""" for i in range(150): - watch_service.state.add_event( - path=f"test{i}.md", - action="sync", - status="success" - ) - + watch_service.state.add_event(path=f"test{i}.md", action="sync", status="success") + assert len(watch_service.state.recent_events) == 100 # Most recent should be at the start assert watch_service.state.recent_events[0].path == "test149.md" @@ -127,26 +120,20 @@ async def test_state_serialization(watch_service): """Test state serializes to dict correctly""" # Add some test state watch_service.state.running = True - watch_service.state.files_synced = 42 - watch_service.state.add_event( - path="test.md", - action="sync", - status="success" - ) - + watch_service.state.add_event(path="test.md", action="sync", status="success") + data = WatchServiceState.model_dump(watch_service.state) - + # Check basic fields - assert data['running'] is True - assert data['files_synced'] == 42 - assert isinstance(data['start_time'], datetime) - + assert data["running"] is True + assert isinstance(data["start_time"], datetime) + # Check events serialized - assert len(data['recent_events']) == 1 - event = data['recent_events'][0] - assert event['path'] == "test.md" - assert event['action'] == "sync" - assert isinstance(event['timestamp'], datetime) + assert len(data["recent_events"]) == 1 + event = data["recent_events"][0] + assert event["path"] == "test.md" + assert event["action"] == "sync" + assert isinstance(event["timestamp"], datetime) @pytest.mark.asyncio @@ -154,13 +141,14 @@ async def test_status_file(watch_service, tmp_path): """Test status file writing""" watch_service.state.running = True await watch_service.write_status() - + status_file = watch_service.status_path assert status_file.exists() - + # Should be valid JSON with our fields import json + data = json.loads(status_file.read_text()) - assert data['running'] is True - assert isinstance(data['start_time'], str) - assert data['pid'] > 0 + assert data["running"] is True + assert isinstance(data["start_time"], str) + assert data["pid"] > 0 diff --git a/tests/sync/test_watch_sync.py b/tests/sync/test_watch_sync.py deleted file mode 100644 index 6ba74a53..00000000 --- a/tests/sync/test_watch_sync.py +++ /dev/null @@ -1,129 +0,0 @@ -"""Tests for watch service integration with sync service.""" - -import pytest -import pytest_asyncio -from pathlib import Path -from watchfiles import Change - -from basic_memory.sync.utils import FileChange - - -@pytest.mark.asyncio -async def test_sync_with_file_changes(sync_service, tmp_path): - """Test sync with file changes from watch service""" - # Create a test file - content = """--- -title: Test Note -type: note ---- -# Test Note -This is a test.""" - - file_path = tmp_path / "test.md" - file_path.write_text(content) - - # Create FileChange for a new file - changes = { - str(file_path): FileChange( - change_type=Change.added, - path=str(file_path), - checksum="abc123" - ) - } - - report = await sync_service.sync(file_changes=changes) - assert str(file_path) in report.new - assert report.checksums[str(file_path)] == "abc123" - - -@pytest.mark.asyncio -async def test_sync_mixed_changes(sync_service, tmp_path): - """Test sync with multiple types of changes""" - # Setup initial files - new_file = tmp_path / "new.md" - new_file.write_text("New file") - - mod_file = tmp_path / "modified.md" - mod_file.write_text("Original content") - - del_file = tmp_path / "deleted.md" - - changes = { - str(new_file): FileChange( - change_type=Change.added, - path=str(new_file), - checksum="new123" - ), - str(mod_file): FileChange( - change_type=Change.modified, - path=str(mod_file), - checksum="mod123" - ), - str(del_file): FileChange( - change_type=Change.deleted, - path=str(del_file) - ) - } - - report = await sync_service.sync(file_changes=changes) - - # Verify report contains all changes - assert str(new_file) in report.new - assert str(mod_file) in report.modified - assert str(del_file) in report.deleted - - # Check checksums - assert report.checksums[str(new_file)] == "new123" - assert report.checksums[str(mod_file)] == "mod123" - assert str(del_file) not in report.checksums - - -@pytest.mark.asyncio -async def test_sync_requires_params(sync_service): - """Test sync requires either directory or file_changes""" - with pytest.raises(ValueError): - await sync_service.sync() - - -@pytest.mark.asyncio -async def test_sync_with_invalid_change_type(sync_service, tmp_path): - """Test handling of invalid change types""" - file_path = tmp_path / "test.md" - file_path.touch() - - # Type checker won't let us create this directly, so ignore the type - changes = { - str(file_path): FileChange( - change_type="invalid", # type: ignore - path=str(file_path), - checksum="test123" - ) - } - - with pytest.raises(Exception): - await sync_service.sync(file_changes=changes) - - -@pytest.mark.asyncio -async def test_end_to_end_watch_sync(watch_service, sync_service, tmp_path): - """Test complete watch -> sync flow""" - # Create a test file that will trigger a watch event - test_file = tmp_path / "test.md" - test_file.write_text("""--- -title: Test Note -type: note ---- -# Test -Testing watch -> sync flow -""") - - # Simulate watchfiles event - changes = {(Change.added, str(test_file))} - await watch_service.handle_changes(changes) - - # Check watch service state - assert watch_service.state.files_synced == 1 - assert len(watch_service.state.recent_events) == 1 - event = watch_service.state.recent_events[0] - assert event.path == str(test_file) - assert event.status == "success"