Spaces:
Running
Running
File size: 5,060 Bytes
6ea3b3e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 | """Local FTS5 sparse retrieval.
Builds a small real sidecar rather than mocking sqlite: the failure modes worth
covering here are FTS5 syntax errors from user text and the AND/OR fallback, and
neither survives a mock.
"""
import asyncio
import sqlite3
import pytest
from app import config, fts_svc, local_meta
@pytest.fixture
def sidecar(tmp_path, monkeypatch):
path = tmp_path / "metadata.sqlite"
c = sqlite3.connect(path)
c.execute("""CREATE TABLE papers (
arxiv_id TEXT PRIMARY KEY, title TEXT, authors TEXT,
abstract_preview TEXT, categories TEXT, primary_topic TEXT,
update_date TEXT, citation_count INTEGER,
influential_citations INTEGER)""")
rows = [
("2506.00001", "Diffusion transformers for video generation",
"A diffusion transformer that generates video frames.", "2025-06-01"),
("0704.0002", "Sparsity-certifying graph decompositions",
"A pebble game algorithm for sparse graphs.", "2007-04-01"),
("2606.00001", "Retrieval augmented generation for agents",
"An agent framework using retrieval augmented generation.", "2026-06-01"),
]
c.executemany("INSERT INTO papers (arxiv_id,title,abstract_preview,update_date,"
"citation_count,influential_citations) VALUES (?,?,?,?,0,0)",
[(a, t, ab, d) for a, t, ab, d in rows])
c.execute("""CREATE VIRTUAL TABLE papers_fts USING fts5(
title, abstract_preview, content='papers',
content_rowid='rowid', tokenize='porter unicode61')""")
c.execute("INSERT INTO papers_fts(papers_fts) VALUES('rebuild')")
c.commit(); c.close()
# Reset both modules' cached probes so each test opens this sidecar.
monkeypatch.setattr(local_meta, "SIDECAR_PATH", str(path))
monkeypatch.setattr(local_meta, "_conn", None)
monkeypatch.setattr(local_meta, "_probed", False)
monkeypatch.setattr(local_meta, "_available", False)
monkeypatch.setattr(fts_svc, "_probed", False)
monkeypatch.setattr(fts_svc, "_available", False)
return path
def run(q, limit=10):
return asyncio.run(fts_svc.search_sparse(q, limit=limit))
def test_finds_paper_by_title(sidecar):
out = run("diffusion transformer video")
assert [r["arxiv_id"] for r in out][:1] == ["2506.00001"]
def test_covers_papers_newer_than_the_snapshot(sidecar):
"""The whole point of the switch: post-2025-06 papers must be reachable."""
out = run("retrieval augmented generation agents")
assert "2606.00001" in [r["arxiv_id"] for r in out]
def test_falls_back_from_and_to_or(sidecar):
"""A term absent from the corpus must not empty the result set."""
out = run("diffusion zzzznotacorpusterm")
assert [r["arxiv_id"] for r in out][:1] == ["2506.00001"]
@pytest.mark.parametrize("q", [
'quantum "unbalanced', # unterminated quote
"graph AND OR NOT", # bare operators
"sparse NEAR/2 graph", # NEAR syntax
"video*", # prefix operator
"^anchored", # column-anchor operator
"it's a graph", # apostrophe
"-- drop table papers", # sql-ish noise
])
def test_user_text_is_never_parsed_as_fts_syntax(sidecar, q):
"""Any of these raises OperationalError if passed to MATCH unescaped,
which would silently kill the sparse arm for that query."""
out = run(q)
assert isinstance(out, list)
def test_empty_and_punctuation_only_queries(sidecar):
assert run("") == []
assert run(" ") == []
assert run("!!! ??? ...") == []
def test_single_characters_are_dropped(sidecar):
assert fts_svc._terms("a b diffusion c") == ["diffusion"]
def test_scores_are_higher_is_better(sidecar):
"""Negated from bm25()'s more-negative-is-better so it matches the dense arm."""
out = run("diffusion transformer video")
assert len(out) >= 1
assert out == sorted(out, key=lambda r: r["score"], reverse=True)
def test_respects_limit(sidecar):
assert len(run("generation OR graph OR diffusion", limit=2)) <= 2
def test_unavailable_without_fts_index(tmp_path, monkeypatch):
"""An older sidecar has no papers_fts; degrade instead of raising."""
path = tmp_path / "old.sqlite"
c = sqlite3.connect(path)
c.execute("CREATE TABLE papers (arxiv_id TEXT PRIMARY KEY, title TEXT)")
c.execute("INSERT INTO papers VALUES ('1','t')")
c.commit(); c.close()
monkeypatch.setattr(local_meta, "SIDECAR_PATH", str(path))
monkeypatch.setattr(local_meta, "_conn", None)
monkeypatch.setattr(local_meta, "_probed", False)
monkeypatch.setattr(local_meta, "_available", False)
monkeypatch.setattr(fts_svc, "_probed", False)
monkeypatch.setattr(fts_svc, "_available", False)
assert fts_svc.is_available() is False
assert run("anything") == []
def test_stats_reports_index_size(sidecar):
s = fts_svc.stats()
assert s["available"] is True
assert s["indexed"] == 3
|