"""Query layers: RRF merge, the FTS expression, degraded mode, toc and grep.""" import shutil import sys from pathlib import Path import pytest SCRIPTS = Path(__file__).parent.parent / "scripts" sys.path.insert(0, str(SCRIPTS)) import wiki_search import wiki_store as store import wiki_sync from wiki_config import SourceConfig, load_config BACKUP_DOC = """--- title: Zálohování dat tags: [devops, backup] --- # Zálohování dat Záloha běží každou noc přes rsync na druhý disk. ## Retence snapshotů Držíme třicet denních snapshotů a dvanáct měsíčních. """ TRAVEL_DOC = """--- title: Tokijské metro tags: [transit] --- # Tokijské metro Z Narity do centra jede Skyliner za osmatřicet minut. """ def _index(env): assert wiki_sync.main([]) == 0 # --------------------------------------------------------------------------- # pure functions # --------------------------------------------------------------------------- def test_rrf_merge_over_known_ranks(): merged = wiki_search.rrf_merge([[1, 2, 3], [3, 1, 4]]) assert [chunk_id for chunk_id, _ in merged] == [1, 3, 2, 4] scores = dict(merged) assert scores[1] == pytest.approx(1 / 61 + 1 / 62) assert scores[3] == pytest.approx(1 / 63 + 1 / 61) assert scores[4] == pytest.approx(1 / 63) def test_rrf_merge_of_a_single_list_keeps_its_order(): assert [i for i, _ in wiki_search.rrf_merge([[7, 8, 9]])] == [7, 8, 9] def test_fts_expression_adds_prefix_wildcards(): """Czech inflection is covered by the wildcard; a 2-char prefix is too broad to keep.""" assert wiki_search.fts_match_expression("záloha dat") == '"záloha"* OR "dat"*' assert wiki_search.fts_match_expression("v ok dva") == '"v" OR "ok" OR "dva"*' def test_fts_expression_neutralises_operators_and_punctuation(): expression = wiki_search.fts_match_expression("NOT (a AND b) -c*") assert expression == '"NOT"* OR "a" OR "AND"* OR "b" OR "c"' def test_fts_expression_of_empty_query_is_empty(): assert wiki_search.fts_match_expression("!!! ") == "" # --------------------------------------------------------------------------- # search # --------------------------------------------------------------------------- def test_search_reports_which_half_found_each_hit(wiki_env, fake_embedder, capsys): wiki_env.write_file("notes/zalohy.md", BACKUP_DOC) wiki_env.write_file("notes/tokio.md", TRAVEL_DOC) _index(wiki_env) config = load_config(wiki_env.config_path) assert wiki_search.run_search(config, "retence snapshotů", limit=5) == 0 out = capsys.readouterr().out assert "notes/zalohy.md" in out assert "Retence snapshotů" in out assert "bm25 #" in out and "vec #" in out def test_search_prefix_match_finds_inflected_form(wiki_env, fake_embedder, capsys): wiki_env.write_file("notes/zalohy.md", BACKUP_DOC) _index(wiki_env) config = load_config(wiki_env.config_path) assert wiki_search.run_search(config, "zaloh", limit=5) == 0 assert "notes/zalohy.md" in capsys.readouterr().out def test_search_says_out_loud_when_embeddings_are_unavailable(wiki_env, fake_embedder, capsys): """Verification 4: degrade to FTS-only, never silently.""" wiki_env.write_file("notes/zalohy.md", BACKUP_DOC) _index(wiki_env) capsys.readouterr() # Drop the fake and let the config's unreachable endpoint take over. wiki_search.OllamaEmbedder = wiki_search.__dict__["OllamaEmbedder"] from wiki_embed import OllamaEmbedder as RealEmbedder wiki_search.OllamaEmbedder = RealEmbedder config = load_config(wiki_env.config_path) assert wiki_search.run_search(config, "záloha", limit=5) == 0 out = capsys.readouterr().out assert "embeddings unavailable" in out assert "FTS-only" in out assert "notes/zalohy.md" in out # the lexical half still answers def test_search_refuses_an_index_from_another_contract(wiki_env, fake_embedder, capsys): """Verification 5: a mismatch must say `reindex needed`, not return results.""" wiki_env.write_file("notes/zalohy.md", BACKUP_DOC) _index(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:8b" ), encoding="utf-8", ) config = load_config(wiki_env.config_path) assert wiki_search.run_search(config, "záloha", limit=5) == 1 captured = capsys.readouterr() assert "reindex needed" in captured.err assert "notes/zalohy.md" not in captured.out def test_search_warns_while_vectors_are_still_pending(wiki_env, monkeypatch, capsys): wiki_env.write_file("notes/zalohy.md", BACKUP_DOC) assert wiki_sync.main([]) == 1 # unreachable endpoint -> chunks stay pending from conftest import FakeEmbedder monkeypatch.setattr(wiki_search, "OllamaEmbedder", FakeEmbedder) config = load_config(wiki_env.config_path) assert wiki_search.run_search(config, "záloha", limit=5) == 0 out = capsys.readouterr().out assert "awaiting vectors" in out def test_search_over_an_empty_index_says_no_matches(wiki_env, fake_embedder, capsys): config = load_config(wiki_env.config_path) assert wiki_search.run_search(config, "záloha", limit=5) == 0 assert "(no matches)" in capsys.readouterr().out def test_an_unrelated_query_still_returns_nearest_neighbours(wiki_env, fake_embedder, capsys): """KNN has no distance floor: the semantic half always offers its k closest chunks. That is deliberate for retrieval — the agent reads the chunk and judges it — but it means an empty result set only ever means an empty index. """ wiki_env.write_file("notes/zalohy.md", BACKUP_DOC) _index(wiki_env) capsys.readouterr() config = load_config(wiki_env.config_path) assert wiki_search.run_search(config, "kajakářství", limit=5) == 0 out = capsys.readouterr().out assert "notes/zalohy.md" in out assert "bm25 #" not in out # nothing lexical matched; every hit came from the vectors def test_search_with_an_empty_query(wiki_env, fake_embedder, capsys): wiki_env.write_file("notes/zalohy.md", BACKUP_DOC) _index(wiki_env) capsys.readouterr() config = load_config(wiki_env.config_path) assert wiki_search.run_search(config, "???", limit=5) == 0 assert "(empty query)" in capsys.readouterr().out def test_search_limit_caps_the_result_count(wiki_env, fake_embedder, capsys): for index in range(8): wiki_env.write_file(f"notes/doc{index}.md", f"# Dokument {index}\n\nZáloha dat číslo {index}.\n") _index(wiki_env) capsys.readouterr() config = load_config(wiki_env.config_path) assert wiki_search.run_search(config, "záloha dat", limit=3) == 0 out = capsys.readouterr().out assert len([line for line in out.splitlines() if line.startswith(("1. ", "2. ", "3. ", "4. "))]) == 3 # --------------------------------------------------------------------------- # toc # --------------------------------------------------------------------------- def test_toc_groups_by_directory_and_shows_titles_and_tags(wiki_env, fake_embedder, capsys): wiki_env.write_file("notes/devops/zalohy.md", BACKUP_DOC) wiki_env.write_file("notes/travel/tokio.md", TRAVEL_DOC) _index(wiki_env) capsys.readouterr() assert wiki_search.run_toc(None, None) == 0 out = capsys.readouterr().out assert "workspace (2 files)" in out assert "notes/devops/" in out assert "zalohy.md" in out assert "Zálohování dat" in out assert "[devops, backup]" in out def test_toc_filters_by_tag(wiki_env, fake_embedder, capsys): wiki_env.write_file("notes/zalohy.md", BACKUP_DOC) wiki_env.write_file("notes/tokio.md", TRAVEL_DOC) _index(wiki_env) capsys.readouterr() assert wiki_search.run_toc(None, "transit") == 0 out = capsys.readouterr().out assert "tokio.md" in out assert "zalohy.md" not in out def test_toc_on_an_empty_index(wiki_env, fake_embedder, capsys): assert wiki_search.run_toc(None, None) == 0 assert "(no indexed files match)" in capsys.readouterr().out # --------------------------------------------------------------------------- # grep # --------------------------------------------------------------------------- @pytest.mark.skipif(shutil.which("rg") is None, reason="ripgrep not installed") def test_grep_sees_source_code_the_index_never_touches(wiki_env, fake_embedder, capsys): """Layer 1 is the substitute for indexing code, and it costs nothing extra.""" wiki_env.write_file("notes/zalohy.md", BACKUP_DOC) wiki_env.write_file("notes/tools/backup.py", "def rotate_snapshots(keep=30):\n return keep\n") _index(wiki_env) capsys.readouterr() config = load_config(wiki_env.config_path) assert wiki_search.run_grep(config, "rotate_snapshots", None) == 0 out = capsys.readouterr().out assert "backup.py" in out assert "rotate_snapshots" in out # The same identifier is absent from the index — grep is the only layer that has it. with store.connection(wiki_env.db_path) as conn: assert store.bm25_ranked_ids(conn, "rotate_snapshots", 10) == [] @pytest.mark.skipif(shutil.which("rg") is None, reason="ripgrep not installed") def test_grep_stays_inside_the_source_paths(wiki_env, fake_embedder, capsys): wiki_env.write_file("notes/zalohy.md", BACKUP_DOC) wiki_env.write_file("tmp/dump/leak.md", "tajnastruna v tmp\n") config = load_config(wiki_env.config_path) assert wiki_search.run_grep(config, "tajnastruna", None) == 0 assert "(no matches)" in capsys.readouterr().out @pytest.mark.skipif(shutil.which("rg") is None, reason="ripgrep not installed") def test_grep_reports_an_unknown_source(wiki_env, fake_embedder, capsys): config = load_config(wiki_env.config_path) assert wiki_search.run_grep(config, "cokoli", "nope") == 1 assert "unknown source" in capsys.readouterr().err def test_grep_roots_use_the_literal_prefix_of_each_glob(wiki_env): wiki_env.write_file("notes/a.md", "x") wiki_env.write_file("develop/b.md", "x") source = load_config(wiki_env.config_path).source("workspace") assert source is not None assert wiki_search.grep_roots(source) == [ wiki_env.workspace / "notes", wiki_env.workspace / "develop", ] def test_grep_roots_of_a_git_source_is_the_whole_clone(wiki_env): clone = wiki_env.wiki_dir / "remote" / "travel" clone.mkdir(parents=True) source = SourceConfig( source_id="travel", kind="git", url="git@example:travel.git", paths=("**",), include=("*.md",), exclude=(), ) assert wiki_search.grep_roots(source) == [clone]