Spaces:
Running
Running
File size: 3,861 Bytes
634117a | 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 | """
tests/test_api.py
FastAPI endpoint tests using httpx + Starlette TestClient.
No HF token or real LLM calls are needed.
"""
import os
import sys
import pytest
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
@pytest.fixture(autouse=True)
def reset_engine():
"""Reset the server's singleton engine before each test."""
from kerdos_rag.server import _engine
_engine.reset()
yield
_engine.reset()
@pytest.fixture
def client():
from fastapi.testclient import TestClient
from kerdos_rag.server import app
return TestClient(app)
# ── /health ───────────────────────────────────────────────────────────────────
def test_health(client):
r = client.get("/health")
assert r.status_code == 200
assert r.json()["status"] == "ok"
# ── /status ───────────────────────────────────────────────────────────────────
def test_status_empty(client):
r = client.get("/status")
assert r.status_code == 200
data = r.json()
assert data["chunk_count"] == 0
assert data["indexed_sources"] == []
# ── /index ────────────────────────────────────────────────────────────────────
def test_index_txt_file(client, tmp_path):
doc = tmp_path / "info.txt"
doc.write_text("The return policy allows 30-day refunds.", encoding="utf-8")
with open(doc, "rb") as f:
r = client.post("/index", files={"files": ("info.txt", f, "text/plain")})
assert r.status_code == 200
body = r.json()
assert "info.txt" in body["indexed"]
assert body["chunk_count"] > 0
def test_index_reflects_in_status(client, tmp_path):
doc = tmp_path / "data.txt"
doc.write_text("Important enterprise data.", encoding="utf-8")
with open(doc, "rb") as f:
client.post("/index", files={"files": ("data.txt", f, "text/plain")})
status = client.get("/status").json()
assert "data.txt" in status["indexed_sources"]
assert status["chunk_count"] > 0
def test_index_skips_duplicate(client, tmp_path):
doc = tmp_path / "dup.txt"
doc.write_text("Some content.", encoding="utf-8")
with open(doc, "rb") as f:
client.post("/index", files={"files": ("dup.txt", f, "text/plain")})
with open(doc, "rb") as f:
r = client.post("/index", files={"files": ("dup.txt", f, "text/plain")})
body = r.json()
assert "dup.txt" in body["skipped"]
assert body["indexed"] == []
# ── /chat ─────────────────────────────────────────────────────────────────────
def test_chat_422_when_empty(client):
r = client.post("/chat", json={"query": "What is the policy?"})
assert r.status_code == 422
# ── /reset ────────────────────────────────────────────────────────────────────
def test_reset(client, tmp_path):
doc = tmp_path / "file.txt"
doc.write_text("Some data.", encoding="utf-8")
with open(doc, "rb") as f:
client.post("/index", files={"files": ("file.txt", f, "text/plain")})
assert client.get("/status").json()["chunk_count"] > 0
r = client.delete("/reset")
assert r.status_code == 200
assert r.json()["ok"] is True
assert client.get("/status").json()["chunk_count"] == 0
|