tests/test_api.py

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