137 lines
5.7 KiB
Python
137 lines
5.7 KiB
Python
"""Schema-level guarantees: the cascade, and the one hole the cascade leaves."""
|
|
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
SCRIPTS = Path(__file__).parent.parent / "scripts"
|
|
sys.path.insert(0, str(SCRIPTS))
|
|
|
|
import wiki_store as store
|
|
from wiki_db import EMBEDDING_DIMS
|
|
|
|
NOW = "2026-09-09T12:00:00+00:00"
|
|
|
|
|
|
def _seed_file(conn, source_id, path, chunk_texts):
|
|
store.upsert_source(conn, source_id, "workspace")
|
|
store.upsert_file(conn, source_id, path, "T", ["t"], ["H"], "sha", 10, 1.0, NOW)
|
|
store.replace_chunks(conn, source_id, path, [(f"{path} > s", t) for t in chunk_texts])
|
|
for row in store.pending_chunks(conn, 100):
|
|
store.store_embedding(conn, row["id"], [0.1] * EMBEDDING_DIMS, NOW)
|
|
|
|
|
|
def test_delete_file_leaves_no_orphan_vectors(tmp_path):
|
|
"""The FK cascade reaches chunks and chunks_fts but never vec0 — deletes must be explicit."""
|
|
db_path = tmp_path / "index.sqlite"
|
|
with store.transaction(db_path) as conn:
|
|
_seed_file(conn, "ws", "a.md", ["alpha text", "beta text"])
|
|
_seed_file(conn, "ws", "b.md", ["gamma text"])
|
|
|
|
with store.connection(db_path) as conn:
|
|
assert store.index_stats(conn) == {"files": 2, "chunks": 3, "vectors": 3, "pending": 0}
|
|
|
|
with store.transaction(db_path) as conn:
|
|
store.delete_file(conn, "ws", "a.md")
|
|
|
|
with store.connection(db_path) as conn:
|
|
assert store.index_stats(conn) == {"files": 1, "chunks": 1, "vectors": 1, "pending": 0}
|
|
assert store.orphan_vector_ids(conn) == []
|
|
assert store.bm25_ranked_ids(conn, "alpha", 10) == []
|
|
assert len(store.bm25_ranked_ids(conn, "gamma", 10)) == 1
|
|
|
|
|
|
def test_replace_chunks_drops_old_vectors(tmp_path):
|
|
db_path = tmp_path / "index.sqlite"
|
|
with store.transaction(db_path) as conn:
|
|
_seed_file(conn, "ws", "a.md", ["alpha text", "beta text", "delta text"])
|
|
|
|
with store.transaction(db_path) as conn:
|
|
store.replace_chunks(conn, "ws", "a.md", [("a.md > s", "only one now")])
|
|
|
|
with store.connection(db_path) as conn:
|
|
assert store.index_stats(conn) == {"files": 1, "chunks": 1, "vectors": 0, "pending": 1}
|
|
assert store.orphan_vector_ids(conn) == []
|
|
assert store.bm25_ranked_ids(conn, "alpha", 10) == []
|
|
|
|
|
|
def test_delete_by_path_does_not_touch_other_source(tmp_path):
|
|
"""The key is (source_id, path); a path-only delete would eat a foreign source's chunks."""
|
|
db_path = tmp_path / "index.sqlite"
|
|
with store.transaction(db_path) as conn:
|
|
_seed_file(conn, "ws", "notes.md", ["shared name one"])
|
|
_seed_file(conn, "git", "notes.md", ["shared name two"])
|
|
|
|
with store.transaction(db_path) as conn:
|
|
store.delete_file(conn, "ws", "notes.md")
|
|
|
|
with store.connection(db_path) as conn:
|
|
assert store.index_stats(conn)["chunks"] == 1
|
|
assert store.orphan_vector_ids(conn) == []
|
|
remaining = list(conn.execute("SELECT source_id FROM chunks"))
|
|
assert remaining[0]["source_id"] == "git"
|
|
|
|
|
|
def test_embedded_at_update_keeps_fts_row(tmp_path):
|
|
"""Step 4b flips embedded_at on unchanged text; the WHEN guard must keep FTS intact."""
|
|
db_path = tmp_path / "index.sqlite"
|
|
with store.transaction(db_path) as conn:
|
|
store.upsert_source(conn, "ws", "workspace")
|
|
store.upsert_file(conn, "ws", "a.md", "T", [], [], "sha", 10, 1.0, NOW)
|
|
store.replace_chunks(conn, "ws", "a.md", [("a.md", "zaloha dat na disk")])
|
|
|
|
with store.connection(db_path) as conn:
|
|
assert len(store.bm25_ranked_ids(conn, "zaloha", 10)) == 1
|
|
|
|
with store.transaction(db_path) as conn:
|
|
pending = store.pending_chunks(conn, 10)
|
|
store.store_embedding(conn, pending[0]["id"], [0.2] * EMBEDDING_DIMS, NOW)
|
|
|
|
with store.connection(db_path) as conn:
|
|
assert len(store.bm25_ranked_ids(conn, "zaloha", 10)) == 1
|
|
assert store.index_stats(conn) == {"files": 1, "chunks": 1, "vectors": 1, "pending": 0}
|
|
|
|
|
|
def test_fts_folds_czech_diacritics(tmp_path):
|
|
"""remove_diacritics 2 is what makes `zaloha` find `záloha`."""
|
|
db_path = tmp_path / "index.sqlite"
|
|
with store.transaction(db_path) as conn:
|
|
store.upsert_source(conn, "ws", "workspace")
|
|
store.upsert_file(conn, "ws", "a.md", "T", [], [], "sha", 10, 1.0, NOW)
|
|
store.replace_chunks(conn, "ws", "a.md", [("a.md", "zálohování dat probíhá denně")])
|
|
|
|
with store.connection(db_path) as conn:
|
|
assert len(store.bm25_ranked_ids(conn, "zaloh*", 10)) == 1
|
|
assert len(store.bm25_ranked_ids(conn, "záloh*", 10)) == 1
|
|
|
|
|
|
def test_meta_roundtrip_and_rollback(tmp_path):
|
|
db_path = tmp_path / "index.sqlite"
|
|
with store.transaction(db_path) as conn:
|
|
store.write_meta(conn, {"embedding_model": "qwen3-embedding:0.6b", "embedding_dims": "1024"})
|
|
|
|
with store.connection(db_path) as conn:
|
|
assert store.read_meta(conn)["embedding_dims"] == "1024"
|
|
|
|
try:
|
|
with store.transaction(db_path) as conn:
|
|
store.write_meta(conn, {"embedding_dims": "768"})
|
|
raise RuntimeError("boom")
|
|
except RuntimeError:
|
|
pass
|
|
|
|
with store.connection(db_path) as conn:
|
|
assert store.read_meta(conn)["embedding_dims"] == "1024"
|
|
|
|
|
|
def test_toc_filters_by_tag(tmp_path):
|
|
db_path = tmp_path / "index.sqlite"
|
|
with store.transaction(db_path) as conn:
|
|
store.upsert_source(conn, "travel", "git")
|
|
store.upsert_file(conn, "travel", "japan/metro.md", "Metro", ["transit"], [], "s", 1, 1.0, NOW)
|
|
store.upsert_file(conn, "travel", "alps/gear.md", "Gear", ["gear", "safety"], [], "s", 1, 1.0, NOW)
|
|
|
|
with store.connection(db_path) as conn:
|
|
assert [r["path"] for r in store.list_toc_files(conn, tag="gear")] == ["alps/gear.md"]
|
|
assert len(store.list_toc_files(conn, source_id="travel")) == 2
|
|
assert store.list_toc_files(conn, source_id="nope") == []
|