basicmachines-co-basic-memory/tests/sync/test_watch_service_edge_cases.py

"""Test edge cases in the WatchService."""

import pytest
from watchfiles import Change


def test_filter_changes_valid_path(watch_service, project_config):
    """Test the filter_changes method with valid non-hidden paths."""
    # Regular file path
    assert (
        watch_service.filter_changes(Change.added, str(project_config.home / "valid_file.txt"))
        is True
    )

    # Nested path
    assert (
        watch_service.filter_changes(
            Change.added, str(project_config.home / "nested" / "valid_file.txt")
        )
        is True
    )


def test_filter_changes_hidden_path(watch_service, project_config):
    """Test the filter_changes method with hidden files/directories."""
    # Hidden file (starts with dot)
    assert (
        watch_service.filter_changes(Change.added, str(project_config.home / ".hidden_file.txt"))
        is False
    )

    # File in hidden directory
    assert (
        watch_service.filter_changes(
            Change.added, str(project_config.home / ".hidden_dir" / "file.txt")
        )
        is False
    )

    # Deeply nested hidden directory
    assert (
        watch_service.filter_changes(
            Change.added, str(project_config.home / "valid" / ".hidden" / "file.txt")
        )
        is False
    )


@pytest.mark.asyncio
async def test_handle_changes_empty_set(watch_service, project_config, test_project):
    """Test handle_changes with an empty set (no processed files)."""
    await watch_service.handle_changes(test_project, set())

    # Verify last_scan was updated
    assert watch_service.state.last_scan is not None

    # Verify synced_files wasn't changed
    assert watch_service.state.synced_files == 0


@pytest.mark.asyncio
async def test_handle_vim_atomic_write_delete_still_exists(
    watch_service, project_config, test_project, sync_service
):
    """Test vim atomic write scenario: DELETE event but file still exists on disk."""
    project_dir = project_config.home

    # Create initial file and sync it
    test_file = project_dir / "vim_test.md"
    initial_content = """---
type: note
title: vim test
---
# Vim Test
Initial content for atomic write test
"""
    test_file.write_text(initial_content)
    await sync_service.sync(project_dir)

    # Get initial entity state
    initial_entity = await sync_service.entity_repository.get_by_file_path("vim_test.md")
    assert initial_entity is not None
    initial_checksum = initial_entity.checksum

    # Simulate vim's atomic write: modify content but send DELETE event
    # (vim moves original file, creates new content, then deletes old inode)
    modified_content = """---
type: note
title: vim test
---
# Vim Test
Modified content after atomic write
"""
    test_file.write_text(modified_content)

    # Setup DELETE event even though file still exists (vim's atomic write behavior)
    # Use absolute path like the real watch service would
    changes = {(Change.deleted, str(test_file))}

    # Handle the change
    await watch_service.handle_changes(test_project, changes)

    # Verify the entity still exists and was updated (not deleted)
    entity = await sync_service.entity_repository.get_by_file_path("vim_test.md")
    assert entity is not None
    assert entity.id == initial_entity.id  # Same entity
    assert entity.checksum != initial_checksum  # Checksum should be updated

    # Verify the file content was properly synced
    actual_content = test_file.read_text()
    assert "Modified content after atomic write" in actual_content

    # Check that correct event was recorded (should be "modified", not "deleted")
    events = [e for e in watch_service.state.recent_events if e.path == "vim_test.md"]
    assert len(events) == 1
    assert events[0].action == "modified"
    assert events[0].status == "success"


@pytest.mark.asyncio
async def test_handle_true_deletion_vs_vim_atomic(
    watch_service, project_config, test_project, sync_service
):
    """Test that true deletions are still handled correctly vs vim atomic writes."""
    project_dir = project_config.home

    # Create and sync two files
    atomic_file = project_dir / "atomic_test.md"
    delete_file = project_dir / "delete_test.md"

    content = """---
type: note
---
# Test File
Content for testing
"""

    atomic_file.write_text(content)
    delete_file.write_text(content)
    await sync_service.sync(project_dir)

    # For atomic_file: modify content but keep file (vim atomic write scenario)
    modified_content = content.replace("Content for testing", "Modified content")
    atomic_file.write_text(modified_content)

    # For delete_file: actually delete it (true deletion)
    delete_file.unlink()

    # Setup DELETE events for both files
    # Use absolute paths like the real watch service would
    changes = {
        (Change.deleted, str(atomic_file)),  # File still exists - atomic write
        (Change.deleted, str(delete_file)),  # File deleted - true deletion
    }

    # Handle the changes
    await watch_service.handle_changes(test_project, changes)

    # Verify atomic_file was treated as modification (still exists in DB)
    atomic_entity = await sync_service.entity_repository.get_by_file_path("atomic_test.md")
    assert atomic_entity is not None

    # Verify delete_file was truly deleted (no longer exists in DB)
    delete_entity = await sync_service.entity_repository.get_by_file_path("delete_test.md")
    assert delete_entity is None

    # Check events were recorded correctly
    events = watch_service.state.recent_events
    atomic_events = [e for e in events if e.path == "atomic_test.md"]
    delete_events = [e for e in events if e.path == "delete_test.md"]

    assert len(atomic_events) == 1
    assert atomic_events[0].action == "modified"

    assert len(delete_events) == 1
    assert delete_events[0].action == "deleted"


@pytest.mark.asyncio
async def test_handle_vim_atomic_write_markdown_with_relations(
    watch_service, project_config, test_project, sync_service
):
    """Test vim atomic write with markdown files that contain relations."""
    project_dir = project_config.home

    # Create target file for relations
    target_file = project_dir / "target.md"
    target_content = """---
type: note
title: Target Note
---
# Target Note
This is the target of relations.
"""
    target_file.write_text(target_content)

    # Create main file with relations
    main_file = project_dir / "main.md"
    initial_content = """---
type: note
title: Main Note
---
# Main Note
This note links to [[Target Note]].

- relates_to [[Target Note]]
"""
    main_file.write_text(initial_content)
    await sync_service.sync(project_dir)

    # Get initial state
    main_entity = await sync_service.entity_repository.get_by_file_path("main.md")
    assert main_entity is not None
    initial_relations = len(main_entity.relations)

    # Simulate vim atomic write with content change that adds more relations
    modified_content = """---
type: note
title: Main Note
---
# Main Note
This note links to [[Target Note]] multiple times.

- relates_to [[Target Note]]
- references [[Target Note]]
"""
    main_file.write_text(modified_content)

    # Setup DELETE event (vim atomic write)
    # Use absolute path like the real watch service would
    changes = {(Change.deleted, str(main_file))}

    # Handle the change
    await watch_service.handle_changes(test_project, changes)

    # Verify entity still exists and relations were updated
    updated_entity = await sync_service.entity_repository.get_by_file_path("main.md")
    assert updated_entity is not None
    assert updated_entity.id == main_entity.id

    # Verify relations were processed correctly
    updated_relations = len(updated_entity.relations)
    assert updated_relations >= initial_relations  # Should have at least as many relations

    # Check event was recorded as modification
    events = [e for e in watch_service.state.recent_events if e.path == "main.md"]
    assert len(events) == 1
    assert events[0].action == "modified"


@pytest.mark.asyncio
async def test_handle_vim_atomic_write_directory_path_ignored(
    watch_service, project_config, test_project
):
    """Test that directories are properly ignored even in atomic write detection."""
    project_dir = project_config.home

    # Create directory
    test_dir = project_dir / "test_directory"
    test_dir.mkdir()

    # Setup DELETE event for directory (should be ignored)
    # Use absolute path like the real watch service would
    changes = {(Change.deleted, str(test_dir))}

    # Handle the change - should not cause errors
    await watch_service.handle_changes(test_project, changes)

    # Verify no events were recorded for the directory
    events = [e for e in watch_service.state.recent_events if "test_directory" in e.path]
    assert len(events) == 0