nanobot-runtime/skills/wiki/tests/test_wiki_sync.py

"""Sync driver: idempotence, the lock, degraded mode, the meta guard, coverage."""

import json
import os
import sys
from pathlib import Path

SCRIPTS = Path(__file__).parent.parent / "scripts"
sys.path.insert(0, str(SCRIPTS))

import wiki_store as store
import wiki_sync
from wiki_config import load_config
from wiki_embed import expected_meta

DOC = """# Zálohování

Záloha dat na disk probíhá každý den v noci pomocí rsyncu.

## Retence

Držíme třicet denních snapshotů a dvanáct měsíčních.
"""

OTHER_DOC = """# Cestování

Tokio metro je nejrychlejší cesta z Narity do centra.
"""


def _stats(env):
    with store.connection(env.db_path) as conn:
        return store.index_stats(conn)


def _run(argv=None):
    return wiki_sync.main(argv or [])


def _file_row(conn, path):
    row = store.get_file(conn, "workspace", path)
    assert row is not None, f"{path} is not indexed"
    return row


def test_indexes_covered_workspace_files(wiki_env, fake_embedder):
    wiki_env.write_file("notes/zalohy.md", DOC)
    wiki_env.write_file("develop/knowledge.md", OTHER_DOC)

    assert _run() == 0

    stats = _stats(wiki_env)
    assert stats["files"] == 2
    assert stats["chunks"] > 0
    assert stats["vectors"] == stats["chunks"]
    assert stats["pending"] == 0
    assert "indexed 2 files" in wiki_env.log_text() or "indexed 1 files" in wiki_env.log_text()


def test_paths_whitelist_and_exclude_are_honoured(wiki_env, fake_embedder):
    wiki_env.write_file("notes/keep.md", DOC)
    wiki_env.write_file("notes/inbox/raw.md", DOC)  # excluded
    wiki_env.write_file("develop/history.md", DOC)  # excluded
    wiki_env.write_file("tmp/clone/README.md", DOC)  # outside paths
    wiki_env.write_file("notes/script.py", "# TODO: fix this\n")  # outside include

    assert _run() == 0

    with store.connection(wiki_env.db_path) as conn:
        paths = sorted(store.list_source_files(conn, "workspace"))
    assert paths == ["notes/keep.md"]


def test_second_run_over_unchanged_tree_is_a_no_op(wiki_env, fake_embedder):
    """Verification 2: counts identical and nothing appended to the log."""
    wiki_env.write_file("notes/zalohy.md", DOC)
    assert _run() == 0

    first_stats = _stats(wiki_env)
    with store.connection(wiki_env.db_path) as conn:
        first_indexed_at = _file_row(conn, "notes/zalohy.md")["indexed_at"]
    log_after_first = wiki_env.log_text()

    assert _run() == 0

    assert _stats(wiki_env) == first_stats
    assert wiki_env.log_text() == log_after_first
    with store.connection(wiki_env.db_path) as conn:
        assert _file_row(conn, "notes/zalohy.md")["indexed_at"] == first_indexed_at


def test_touching_a_file_without_changing_it_only_refreshes_stats(wiki_env, fake_embedder):
    wiki_env.write_file("notes/zalohy.md", DOC)
    assert _run() == 0
    before = _stats(wiki_env)
    with store.connection(wiki_env.db_path) as conn:
        indexed_at = _file_row(conn, "notes/zalohy.md")["indexed_at"]

    os.utime(wiki_env.workspace / "notes/zalohy.md", (1_600_000_000, 1_600_000_000))
    assert _run() == 0

    assert _stats(wiki_env) == before
    with store.connection(wiki_env.db_path) as conn:
        row = _file_row(conn, "notes/zalohy.md")
    assert row["indexed_at"] == indexed_at  # content untouched -> no re-chunk
    assert row["mtime"] == 1_600_000_000


def test_edited_file_is_rechunked_and_reembedded(wiki_env, fake_embedder):
    wiki_env.write_file("notes/zalohy.md", DOC)
    assert _run() == 0

    wiki_env.write_file("notes/zalohy.md", DOC + "\n## Offsite\n\nKopie jede do S3 každý týden.\n")
    assert _run() == 0

    stats = _stats(wiki_env)
    assert stats["pending"] == 0
    assert stats["vectors"] == stats["chunks"]
    with store.connection(wiki_env.db_path) as conn:
        assert store.orphan_vector_ids(conn) == []
        assert len(store.bm25_ranked_ids(conn, "offsite", 10)) == 1


def test_deleted_file_leaves_no_chunks_or_vectors(wiki_env, fake_embedder):
    wiki_env.write_file("notes/zalohy.md", DOC)
    wiki_env.write_file("notes/travel.md", OTHER_DOC)
    assert _run() == 0

    (wiki_env.workspace / "notes/travel.md").unlink()
    assert _run() == 0

    with store.connection(wiki_env.db_path) as conn:
        assert sorted(store.list_source_files(conn, "workspace")) == ["notes/zalohy.md"]
        assert store.orphan_vector_ids(conn) == []
        assert store.bm25_ranked_ids(conn, "tokio", 10) == []
    stats = _stats(wiki_env)
    assert stats["vectors"] == stats["chunks"]


def test_live_lock_makes_the_run_a_silent_no_op(wiki_env, fake_embedder):
    """Verification 3: a second concurrent sync exits 0 without writing."""
    wiki_env.write_file("notes/zalohy.md", DOC)
    wiki_sync.LOCK_PATH.write_text(
        json.dumps({"pid": os.getpid(), "started_at": wiki_sync.datetime.now().astimezone().isoformat()}),
        encoding="utf-8",
    )

    assert _run() == 0

    assert not wiki_env.db_path.exists()
    assert wiki_env.log_text() == ""


def test_stale_lock_is_reclaimed_and_logged(wiki_env, fake_embedder):
    wiki_env.write_file("notes/zalohy.md", DOC)
    dead_pid = 999_999
    wiki_sync.LOCK_PATH.write_text(
        json.dumps({"pid": dead_pid, "started_at": wiki_sync.datetime.now().astimezone().isoformat()}),
        encoding="utf-8",
    )

    assert _run() == 0

    assert "stale lock, reclaiming" in wiki_env.log_text()
    assert _stats(wiki_env)["files"] == 1
    assert not wiki_sync.LOCK_PATH.exists()


def test_unreadable_lock_is_treated_as_stale(wiki_env, fake_embedder):
    wiki_env.write_file("notes/zalohy.md", DOC)
    wiki_sync.LOCK_PATH.write_text("not json", encoding="utf-8")

    assert _run() == 0
    assert _stats(wiki_env)["files"] == 1


def test_coverage_report_names_uncovered_directories(wiki_env, fake_embedder):
    """Verification 6: markdown outside `paths` must surface as a log line."""
    wiki_env.write_file("notes/zalohy.md", DOC)
    wiki_env.write_file("recepty/gulas.md", "# Guláš\n\nCibule na dva kusy hovězího.\n")
    # The skill's own clone storage holds markdown but is never a candidate.
    (wiki_env.wiki_dir / "remote" / "index").mkdir(parents=True)
    (wiki_env.wiki_dir / "remote" / "index" / "foreign.md").write_text("# Cizí\n", encoding="utf-8")

    assert _run() == 0

    coverage = [line for line in wiki_env.log_text().splitlines() if "coverage" in line]
    assert len(coverage) == 1
    assert "recepty" in coverage[0]
    assert "wiki" not in coverage[0]


def test_degraded_mode_then_recovery_without_file_change(wiki_env, monkeypatch):
    """Verification 4: no Ollama -> FTS-only pending; once back, step 4b fills the vectors."""
    wiki_env.write_file("notes/zalohy.md", DOC)

    # The fixture config points at embed.invalid, so no embedder is reachable here.
    assert _run() == 1
    stats = _stats(wiki_env)
    assert stats["chunks"] > 0
    assert stats["vectors"] == 0
    assert stats["pending"] == stats["chunks"]
    assert "embeddings unavailable" in wiki_env.log_text()

    with store.connection(wiki_env.db_path) as conn:
        assert len(store.bm25_ranked_ids(conn, "zaloh*", 10)) >= 1  # FTS works meanwhile

    from conftest import FakeEmbedder

    monkeypatch.setattr(wiki_sync, "OllamaEmbedder", FakeEmbedder)
    assert _run() == 0  # no file changed, yet the pending chunks get vectors

    stats = _stats(wiki_env)
    assert stats["pending"] == 0
    assert stats["vectors"] == stats["chunks"]


def test_meta_mismatch_blocks_indexing_until_full(wiki_env, fake_embedder):
    """Verification 5: never mix vectors from two contracts."""
    wiki_env.write_file("notes/zalohy.md", DOC)
    assert _run() == 0
    before = _stats(wiki_env)

    wiki_env.config_path.write_text(
        wiki_env.config_path.read_text(encoding="utf-8").replace(
            "model: qwen3-embedding:0.6b", "model: qwen3-embedding:4b"
        ),
        encoding="utf-8",
    )
    wiki_env.write_file("notes/new.md", OTHER_DOC)

    assert _run() == 1
    assert "index identity mismatch" in wiki_env.log_text()
    assert _stats(wiki_env) == before  # nothing indexed under the new contract

    assert _run(["--full"]) == 0
    with store.connection(wiki_env.db_path) as conn:
        assert store.read_meta(conn)["embedding_model"] == "qwen3-embedding:4b"
        assert sorted(store.list_source_files(conn, "workspace")) == ["notes/new.md", "notes/zalohy.md"]
        assert store.orphan_vector_ids(conn) == []


def test_full_rebuild_wipes_and_reindexes(wiki_env, fake_embedder):
    wiki_env.write_file("notes/zalohy.md", DOC)
    assert _run() == 0
    before = _stats(wiki_env)

    assert _run(["--full"]) == 0

    assert _stats(wiki_env) == before
    with store.connection(wiki_env.db_path) as conn:
        assert store.orphan_vector_ids(conn) == []


def test_meta_is_written_on_a_fresh_index(wiki_env, fake_embedder):
    wiki_env.write_file("notes/zalohy.md", DOC)
    assert _run() == 0

    config = load_config(wiki_env.config_path)
    with store.connection(wiki_env.db_path) as conn:
        assert store.read_meta(conn) == expected_meta(config.embedding)


def test_unknown_source_argument_is_reported(wiki_env, fake_embedder):
    wiki_env.write_file("notes/zalohy.md", DOC)
    assert _run(["--source", "nope"]) == 1
    assert "unknown source" in wiki_env.log_text()


def test_batching_respects_the_configured_size(wiki_env, fake_embedder):
    """batch: 4 in the fixture config — the embedder must be called in chunks of 4."""
    for index in range(6):
        wiki_env.write_file(f"notes/doc{index}.md", f"# Dokument {index}\n\n{DOC}\n")

    assert _run() == 0

    calls = [len(call) for embedder in fake_embedder for call in embedder.calls]
    assert calls, "the embedder was never called"
    assert max(calls) <= 4