File size: 2,671 Bytes
c6c1581 03b3d27 cbd62a3 03b3d27 c6c1581 03b3d27 c6c1581 9ca8ee5 cbd62a3 c6c1581 9ca8ee5 cbd62a3 c6c1581 cbd62a3 c6c1581 03b3d27 c6c1581 | 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 | import numpy as np
from fastapi.testclient import TestClient
from common.db import init_db, upsert_paper
from common.vector_index import create_index, add_vector
from app.search_service import SearchService
from app import main as app_main
class FakeEmbeddingsClient:
def embed_batch(self, texts):
return [np.array([1.0, 0.0], dtype=np.float32) for _ in texts]
def _client(tmp_path):
conn = init_db(str(tmp_path / "test.db"))
index = create_index(dim=2)
faiss_id = add_vector(index, np.array([1.0, 0.0], dtype=np.float32))
upsert_paper(conn, {"id": "p1", "title": "Neural Nets", "abstract": "About neural nets", "authors": "A", "venue": "ACL", "year": 2023, "url": "http://x", "bibtex": "@inproceedings{p1}", "pdf_url": "http://x/p1.pdf", "active": True, "faiss_id": faiss_id})
service = SearchService(conn=conn, index=index, embeddings_client=FakeEmbeddingsClient())
app_main.app.state.search_service = service
app_main.app.state.startup_ok = True
app_main.app.state.last_synced_at = "2026-07-03T00:00:00+00:00"
return TestClient(app_main.app)
def test_keyword_search_endpoint(tmp_path):
client = _client(tmp_path)
response = client.get("/search/keyword", params={"q": "neural"})
assert response.status_code == 200
assert response.json()[0]["id"] == "p1"
assert response.json()[0]["bibtex"] == "@inproceedings{p1}"
assert response.json()[0]["pdf_url"] == "http://x/p1.pdf"
def test_similarity_search_endpoint(tmp_path):
client = _client(tmp_path)
response = client.get("/search/similarity", params={"q": "anything", "k": 5})
assert response.status_code == 200
assert response.json()[0]["id"] == "p1"
assert response.json()[0]["bibtex"] == "@inproceedings{p1}"
assert response.json()[0]["pdf_url"] == "http://x/p1.pdf"
def test_get_paper_endpoint(tmp_path):
client = _client(tmp_path)
assert client.get("/paper/p1").status_code == 200
assert client.get("/paper/missing").status_code == 404
result = client.get("/paper/p1").json()
assert result["bibtex"] == "@inproceedings{p1}"
assert result["pdf_url"] == "http://x/p1.pdf"
def test_health_endpoint_ok_after_startup(tmp_path):
client = _client(tmp_path)
response = client.get("/health")
assert response.status_code == 200
assert response.json()["status"] == "ok"
assert response.json()["last_synced_at"] == "2026-07-03T00:00:00+00:00"
def test_health_endpoint_fails_if_startup_not_ok():
app_main.app.state.startup_ok = False
client = TestClient(app_main.app, raise_server_exceptions=False)
response = client.get("/health")
assert response.status_code == 503
|