simbarag/tests/unit/test_obsidian_sync.py

"""Tests for the incremental Obsidian sync: off-loop execution and delete SQL."""

import json
import os
import threading
from unittest.mock import patch

import pytest

from blueprints.rag import logic


class FakeConnection:
    def __init__(self, store):
        self.store = store

    def execute(self, stmt, params=None):
        self.store.append((str(stmt), params or {}))

    def commit(self):
        pass

    def __enter__(self):
        return self

    def __exit__(self, *exc_info):
        return False


class FakeEngine:
    def __init__(self):
        self.executed: list[tuple[str, dict]] = []

    def connect(self):
        return FakeConnection(self.executed)


class FakeVectorStore:
    def __init__(self):
        self.calls: list[list[str]] = []

    async def aadd_documents(self, documents):
        self.calls.append([d.page_content for d in documents])


@pytest.fixture
def vault_dir(tmp_path):
    (tmp_path / "new.md").write_text("# New\nNew note body")
    (tmp_path / "updated.md").write_text("# Updated\nUpdated note body")
    with patch.dict(os.environ, {"OBSIDIAN_VAULT_PATH": str(tmp_path)}):
        yield tmp_path


@pytest.fixture
async def synced(monkeypatch, vault_dir):
    """Run a sync with a fake vault: one new file, one updated file, one
    file that's been deleted off disk since the last sync."""
    updated_path = str(vault_dir / "updated.md")
    missing_path = str(vault_dir / "gone.md")

    monkeypatch.setattr(
        logic, "_obsidian_index_cache", {updated_path: 1.0, missing_path: 1.0}
    )

    engine = FakeEngine()
    monkeypatch.setattr(logic, "_get_engine", lambda: engine)
    monkeypatch.setattr(logic, "_get_collection_id", lambda: "collection-uuid")

    def _unexpected_cache_load():
        raise AssertionError("cache load should be skipped when already populated")

    monkeypatch.setattr(logic, "_load_obsidian_index_cache", _unexpected_cache_load)

    store = FakeVectorStore()
    monkeypatch.setattr(logic, "_get_vector_store", lambda: store)

    threads: dict[str, threading.Thread] = {}

    real_walk = logic._walk_vault_mtimes

    def recording_walk(service):
        threads["walk"] = threading.current_thread()
        return real_walk(service)

    monkeypatch.setattr(logic, "_walk_vault_mtimes", recording_walk)

    real_read = logic._read_obsidian_documents

    def recording_read(service, filepaths, now):
        threads["read"] = threading.current_thread()
        return real_read(service, filepaths, now)

    monkeypatch.setattr(logic, "_read_obsidian_documents", recording_read)

    real_delete = logic.delete_documents_by_filepaths

    def recording_delete(filepaths):
        threads["delete"] = threading.current_thread()
        return real_delete(filepaths)

    monkeypatch.setattr(logic, "delete_documents_by_filepaths", recording_delete)

    result = await logic.sync_obsidian_documents()
    return result, threads, engine, store, updated_path, missing_path


class TestSyncOffLoop:
    async def test_counts_match_vault_changes(self, synced):
        result, *_ = synced
        assert result == {"added": 1, "updated": 1, "deleted": 1}

    async def test_walk_runs_off_the_main_thread(self, synced):
        _, threads, *_ = synced
        assert threads["walk"] is not threading.main_thread()

    async def test_reads_run_off_the_main_thread(self, synced):
        _, threads, *_ = synced
        assert threads["read"] is not threading.main_thread()

    async def test_deletes_run_off_the_main_thread(self, synced):
        _, threads, *_ = synced
        assert threads["delete"] is not threading.main_thread()

    async def test_embeds_new_and_updated_content(self, synced):
        _, _, _, store, *_ = synced
        embedded = {text for call in store.calls for text in call}
        assert any("New note body" in t for t in embedded)
        assert any("Updated note body" in t for t in embedded)

    async def test_cache_reflects_add_update_delete(self, synced):
        result, _, _, _, updated_path, missing_path = synced
        assert missing_path not in logic._obsidian_index_cache
        assert updated_path in logic._obsidian_index_cache

    async def test_deletes_use_jsonb_containment_with_bound_params(self, synced):
        _, _, engine, *_ = synced
        deletes = [(sql, params) for sql, params in engine.executed if "DELETE" in sql]
        assert deletes, "expected at least one delete statement"
        for sql, params in deletes:
            assert "cmetadata @> CAST(:filter AS jsonb)" in sql
            assert "->>:" not in sql  # not the old, unindexed text-extraction filter
            assert "filter" in params
            parsed_filter = json.loads(params["filter"])
            assert "filepath" in parsed_filter


async def test_delete_failure_keeps_deleted_path_in_cache_and_propagates(
    monkeypatch, vault_dir
):
    """If the batched delete raises, the sync must not have already forgotten
    the deleted file: otherwise the next sync never retries the delete and
    its chunks stay searchable forever with no record that they're stale."""
    updated_path = str(vault_dir / "updated.md")
    missing_path = str(vault_dir / "gone.md")

    monkeypatch.setattr(
        logic, "_obsidian_index_cache", {updated_path: 1.0, missing_path: 1.0}
    )

    def _raise_on_delete(filepaths):
        raise RuntimeError("db unavailable")

    monkeypatch.setattr(logic, "delete_documents_by_filepaths", _raise_on_delete)

    store = FakeVectorStore()
    monkeypatch.setattr(logic, "_get_vector_store", lambda: store)

    with pytest.raises(RuntimeError, match="db unavailable"):
        await logic.sync_obsidian_documents()

    assert missing_path in logic._obsidian_index_cache
    # The updated file's old chunks were never actually deleted, so its cache
    # entry must not be bumped to look freshly indexed.
    assert logic._obsidian_index_cache[updated_path] == 1.0
    assert store.calls == []


async def test_delete_documents_by_metadata_uses_containment(monkeypatch):
    engine = FakeEngine()
    monkeypatch.setattr(logic, "_get_engine", lambda: engine)
    monkeypatch.setattr(logic, "_get_collection_id", lambda: "collection-uuid")

    logic.delete_documents_by_metadata("source", "obsidian")

    assert len(engine.executed) == 1
    sql, params = engine.executed[0]
    assert "cmetadata @> CAST(:filter AS jsonb)" in sql
    assert json.loads(params["filter"]) == {"source": "obsidian"}