tests/test_app.py

"""Phase 1: drive the FastAPI /search and /ingest endpoints over a tiny index.

Builds a 3-doc embedded index into a temp dir, then exercises the app the same
way the live demo does.
"""

import importlib
import json
import sys
from pathlib import Path

import pytest

ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT / "src"))
sys.path.insert(0, str(ROOT / "app"))
sys.path.insert(0, str(ROOT / "scripts"))

MINI = [
    {"id": "a1", "title": "Graph neural networks", "text": "Message passing over graph structured data."},
    {"id": "a2", "title": "Streaming systems", "text": "Kafka and Spark process unbounded event streams."},
    {"id": "a3", "title": "Vector search", "text": "HNSW indexes power approximate nearest neighbor retrieval."},
]


@pytest.fixture()
def client(tmp_path, monkeypatch):
    corpus = tmp_path / "corpus.jsonl"
    corpus.write_text("\n".join(json.dumps(d) for d in MINI), encoding="utf-8")
    monkeypatch.setenv("PREBUILT_DIR", str(tmp_path / "prebuilt"))
    monkeypatch.setenv("CORPUS_PATH", str(corpus))
    monkeypatch.delenv("QDRANT_URL", raising=False)

    # Reload config so it picks up the temp paths, then build the index.
    import streamsearch.config as config
    importlib.reload(config)
    build = importlib.import_module("build_index") if "build_index" not in sys.modules \
        else importlib.reload(sys.modules["build_index"])
    build.main()

    # Import the app fresh against the same reloaded config.
    import service
    importlib.reload(service)
    service.get_searcher.cache_clear()
    main = importlib.import_module("main") if "main" not in sys.modules \
        else importlib.reload(sys.modules["main"])

    from fastapi.testclient import TestClient
    with TestClient(main.app) as c:
        yield c


def test_health_reports_doc_count(client):
    r = client.get("/health")
    assert r.status_code == 200
    assert r.json()["documents"] == 3


def test_search_returns_relevant_result(client):
    r = client.get("/search", params={"q": "kafka spark streaming", "top_k": 3})
    assert r.status_code == 200
    assert r.json()["results"][0]["id"] == "a2"


def test_ingest_then_searchable(client):
    r = client.post("/ingest", json={"title": "Zephyr protocol",
                                      "text": "The zephyrium snark lattice teleports widgets."})
    assert r.status_code == 200
    new_id = r.json()["id"]

    r2 = client.get("/search", params={"q": "zephyrium snark lattice widgets", "top_k": 3})
    assert new_id in [h["id"] for h in r2.json()["results"]]
    assert client.get("/health").json()["documents"] == 4