| 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 |
|
|