pv-agent/tests/test_api.py

"""Offline-Tests für den versionierten API-Vertrag und seine Schutzgrenzen."""
from __future__ import annotations

from dataclasses import dataclass

from fastapi.testclient import TestClient

from agent.api import app
from agent.cli import _cmd_serve, _is_loopback_bind
from agent.config import Config


class FakeClient:
    def is_up(self):
        return True

    def close(self):
        pass


class FakeRetriever:
    def stats(self):
        return {
            "n_entries": 3,
            "n_chunks": 9,
            "n_vectors": 9,
            "dense_available": True,
            "stand_min": "202601",
            "stand_max": "202612",
            "built_at": "2026-09-16T00:00:00Z",
        }

    def close(self):
        pass


def answer_result(answer: str = "Belegte Antwort [lb-min-01].") -> dict:
    return {
        "question": "Was gilt?",
        "answer": answer,
        "refused": False,
        "verified": True,
        "citations": ["lb-min-01"],
        "sources": [
            {
                "id": "lb-min-01",
                "title": "Testquelle",
                "section": "Zusammenfassung",
                "stand": "2026-01",
                "work": "Testwerk",
            }
        ],
        "n_context": 1,
        "model": "test-model",
        "regenerations": 0,
        "latency_ms": 12,
        "draft": None,
        "answer_type": "specific",
        "planned": False,
        "planned_queries": [
            {"text": "Was gilt?", "stand_year": None, "scope": None}
        ],
    }


def configure_state(*, api_key: str = "", admin_key: str = "") -> None:
    app.state.rag.cfg = Config(api_key=api_key, admin_api_key=admin_key)
    app.state.rag.client = FakeClient()
    app.state.rag.retriever = FakeRetriever()


def test_frontend_is_served_with_security_headers():
    with TestClient(app) as client:
        configure_state(api_key="service-secret")
        response = client.get("/")
        assert response.status_code == 200
        assert "PV Wissen" in response.text
        assert "service-secret" not in response.text
        assert response.headers["cache-control"] == "no-store"
        assert response.headers["x-frame-options"] == "DENY"
        assert response.headers["x-content-type-options"] == "nosniff"
        assert "script-src 'self'" in response.headers["content-security-policy"]


def test_frontend_assets_use_v1_api_and_unknown_assets_are_hidden():
    with TestClient(app) as client:
        configure_state()
        script = client.get("/assets/app.js")
        styles = client.get("/assets/styles.css")
        missing = client.get("/assets/index.html")
        assert script.status_code == 200
        assert styles.status_code == 200
        assert 'fetch("/v1/ask"' in script.text
        assert "innerHTML" not in script.text
        assert missing.status_code == 404


def test_nonlocal_bind_without_api_key_is_rejected(capsys):
    class Args:
        host = "0.0.0.0"

    assert _is_loopback_bind("127.0.0.1") is True
    assert _is_loopback_bind("::1") is True
    assert _is_loopback_bind("0.0.0.0") is False
    assert _cmd_serve(Args(), Config()) == 2
    assert "PV_API_KEY" in capsys.readouterr().err


def test_v1_ask_requires_configured_bearer(monkeypatch):
    monkeypatch.setattr("agent.api.answer_question", lambda *args, **kwargs: answer_result())
    with TestClient(app) as client:
        configure_state(api_key="service-secret")
        unauthorized = client.post("/v1/ask", json={"question": "Was gilt?"})
        assert unauthorized.status_code == 401
        assert unauthorized.headers["www-authenticate"] == "Bearer"

        response = client.post(
            "/v1/ask",
            json={"question": "Was gilt?"},
            headers={
                "Authorization": "Bearer service-secret",
                "X-Request-ID": "odoo-42",
            },
        )
        assert response.status_code == 200
        body = response.json()
        assert body["api_version"] == "v1"
        assert body["request_id"] == "odoo-42"
        assert response.headers["x-request-id"] == "odoo-42"
        assert body["status"] == "answered"
        assert body["grounding"] == {
            "data_scope": "knowledge_base_only",
            "citations_verified": True,
            "context_count": 1,
            "regenerations": 0,
        }
        assert body["planned_queries"][0]["text"] == "Was gilt?"


def test_v1_contract_rejects_payroll_context(monkeypatch):
    monkeypatch.setattr("agent.api.answer_question", lambda *args, **kwargs: answer_result())
    with TestClient(app) as client:
        configure_state()
        response = client.post(
            "/v1/ask",
            json={
                "question": "Was gilt?",
                "employee_data": {"name": "Max", "salary": 5000},
            },
        )
        assert response.status_code == 422


def test_v1_extracts_grounded_conflict_and_clarification(monkeypatch):
    answer = (
        "⚠ Quelle A und Quelle B widersprechen einander [lb-min-01].\n\n"
        "Für welche Branche soll die Aussage geprüft werden?"
    )
    monkeypatch.setattr(
        "agent.api.answer_question", lambda *args, **kwargs: answer_result(answer)
    )
    with TestClient(app) as client:
        configure_state()
        response = client.post("/v1/ask", json={"question": "Was gilt?"})
        assert response.status_code == 200
        body = response.json()
        assert body["conflicts"][0]["source_ids"] == ["lb-min-01"]
        assert body["clarification_question"] == (
            "Für welche Branche soll die Aussage geprüft werden?"
        )
        assert body["assumptions"] == []
        assert body["alternatives"] == []


def test_health_does_not_expose_internal_ollama_url():
    with TestClient(app) as client:
        configure_state(api_key="service-secret")
        response = client.get("/v1/health")
        assert response.status_code == 200
        body = response.json()
        assert body["status"] == "ok"
        assert body["authentication_enabled"] is True
        assert "ollama_url" not in response.text
        assert "100.103.83.12" not in response.text


@dataclass
class FakeStats:
    embed_error: str | None = None

    def as_dict(self):
        return {"n_entries": 3, "n_chunks": 9}


def test_reindex_uses_separate_admin_key(monkeypatch):
    monkeypatch.setattr(
        "agent.api.build_index", lambda cfg, client=None: FakeStats()
    )
    with TestClient(app) as client:
        configure_state(api_key="service-secret", admin_key="admin-secret")
        denied = client.post(
            "/v1/reindex",
            headers={"Authorization": "Bearer service-secret"},
        )
        assert denied.status_code == 401

        response = client.post(
            "/v1/reindex",
            headers={"Authorization": "Bearer admin-secret"},
        )
        assert response.status_code == 200
        assert response.json()["api_version"] == "v1"
        assert response.json()["n_entries"] == 3