tests/test_api.py

import json
from unittest.mock import patch

import pytest
from fastapi.testclient import TestClient


@pytest.fixture()
def client(monkeypatch):
    monkeypatch.setenv("LANGFUSE_PUBLIC_KEY", "pk-lf-test")
    monkeypatch.setenv("LANGFUSE_SECRET_KEY", "sk-lf-test")
    monkeypatch.setenv("LANGFUSE_TRACING_ENABLED", "false")
    from app import config
    from app.main import app

    monkeypatch.setattr(config, "APP_PASSCODE", "sesame")
    return TestClient(app)


HEADERS = {"X-Contimp-Passcode": "sesame"}


class FakeMessage:
    def __init__(self, content, tool_calls=None):
        self.content = content
        self.tool_calls = tool_calls


def fake_completion(content):
    class Choice:
        message = FakeMessage(content)

    class Response:
        choices = [Choice()]

    return Response()


def test_health_unauthenticated(client):
    assert client.get("/api/health").json()["ok"] is True


def test_passcode_required(client):
    assert client.get("/api/tasks").status_code == 401
    assert client.get("/api/tasks", headers={"X-Contimp-Passcode": "wrong"}).status_code == 401
    assert client.get("/api/tasks", headers=HEADERS).status_code == 200


def test_tasks_and_sample(client):
    tasks = client.get("/api/tasks", headers=HEADERS).json()
    assert {t["id"] for t in tasks} >= {"pr-area", "config-copilot"}
    s = client.post("/api/tasks/pr-area/sample", headers=HEADERS).json()
    assert s["input_id"].startswith("pr-") and s["text"]
    assert client.post("/api/tasks/nope/sample", headers=HEADERS).status_code == 404


def test_run_pr_area_with_mocked_llm(client):
    from app.tasks.pr_area import _RECORDS

    input_id, record = next(iter(_RECORDS.items()))
    answer = json.dumps({"area": record["area"]})
    with patch("app.engine._openai") as mock_llm:
        mock_llm.return_value.chat.completions.create.return_value = fake_completion(answer)
        r = client.post(
            "/api/tasks/pr-area/run",
            headers=HEADERS,
            json={"text": "Title: x", "input_id": input_id, "user": "tester", "session_id": "s1"},
        )
    body = r.json()
    assert r.status_code == 200, body
    assert body["scores"] == {"format_ok": 1.0, "exact_match": 1.0}
    assert body["truth"]["area"] == record["area"]
    assert body["output"]["area"] == record["area"]
    # complete trajectory is returned: system -> user -> final assistant answer
    msgs = body["messages"]
    assert msgs[0]["role"] == "system" and msgs[1]["role"] == "user"
    assert msgs[-1] == {"role": "assistant", "content": answer}


def test_run_without_input_id_has_no_scores(client):
    with patch("app.engine._openai") as mock_llm:
        mock_llm.return_value.chat.completions.create.return_value = fake_completion(
            '{"area": "docs"}'
        )
        r = client.post(
            "/api/tasks/pr-area/run", headers=HEADERS, json={"text": "Title: my own PR"}
        )
    body = r.json()
    assert body["scores"] == {} and body["truth"] is None


def test_run_rejects_empty_text(client):
    r = client.post("/api/tasks/pr-area/run", headers=HEADERS, json={"text": "  "})
    assert r.status_code == 422