tests/test_api.py

"""Exercise the HTTP surface against the fakes: routes, admin gate, SSE.

    python tests/test_api.py
"""

from __future__ import annotations

import json
import os
import sys
import threading
import time
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent.parent))

from tests.fake_servers import LLM_PORT, MCP_PORT, start_background  # noqa: E402

ADMIN = "admin-secret"
os.environ.update(
    {
        "LLM_BASE_URL": f"http://127.0.0.1:{LLM_PORT}/v1",
        "LLM_MODEL": "fake-model",
        "LLM_API_KEY": "test-key",
        "VFRPLAN_MCP_URL": f"http://127.0.0.1:{MCP_PORT}/mcp",
        "VFRPLAN_MCP_TOKEN": "",
        "APP_ADMIN_TOKEN": ADMIN,
    }
)

import httpx  # noqa: E402
import uvicorn  # noqa: E402

from app.main import app  # noqa: E402

APP_PORT = 8933
BASE = f"http://127.0.0.1:{APP_PORT}"

results: list[tuple[bool, str]] = []


def check(label: str, condition: bool, detail: str = "") -> None:
    results.append((condition, label))
    print(f"  [{'PASS' if condition else 'FAIL'}] {label}" + (f" — {detail}" if detail else ""))


def main() -> int:
    start_background()
    threading.Thread(
        target=lambda: uvicorn.run(app, host="127.0.0.1", port=APP_PORT, log_level="error"),
        daemon=True,
    ).start()
    time.sleep(2.5)

    with httpx.Client(timeout=60) as client:
        r = client.get(f"{BASE}/")
        check("GET / serves the UI", r.status_code == 200 and "Planificador VFR" in r.text)

        for asset in ("/static/app.js", "/static/styles.css"):
            r = client.get(f"{BASE}{asset}")
            check(f"GET {asset}", r.status_code == 200 and len(r.text) > 500)

        r = client.get(f"{BASE}/health")
        check("GET /health reports configured", r.json().get("llm_configured") is True, r.text[:100])

        r = client.get(f"{BASE}/api/tools")
        body = r.json()
        check("GET /api/tools lists MCP tools", body.get("ok") and body.get("count") == 2, str(body)[:120])

        r = client.get(f"{BASE}/api/model/status")
        body = r.json()
        check("GET /api/model/status reachable", body.get("state") == "running", str(body)[:140])

        # --- admin gate ---
        r = client.get(f"{BASE}/api/config")
        check("GET /api/config without token is 401", r.status_code == 401)

        r = client.get(f"{BASE}/api/config", headers={"X-Admin-Token": "wrong"})
        check("GET /api/config with wrong token is 401", r.status_code == 401)

        r = client.get(f"{BASE}/api/config", headers={"X-Admin-Token": ADMIN})
        body = r.json()
        masked = body["llm"]["api_key_masked"]
        check("GET /api/config with token returns config", r.status_code == 200)
        check("API key is masked, never returned raw", "test-key" not in json.dumps(body), masked)

        r = client.post(f"{BASE}/api/config", json={"llm": {"model": "x"}})
        check("POST /api/config without token is 401", r.status_code == 401)

        r = client.post(
            f"{BASE}/api/config",
            json={"llm": {"model": "changed-model", "api_key": ""}},
            headers={"X-Admin-Token": ADMIN},
        )
        check("POST /api/config applies override", r.json().get("model") == "changed-model", r.text[:120])

        r = client.get(f"{BASE}/api/config", headers={"X-Admin-Token": ADMIN})
        check(
            "blank api_key left the existing key untouched",
            r.json()["llm"]["has_api_key"] is True,
        )

        # restore, or the chat below would ask the fake LLM for a model it has
        # no opinion about (harmless here, but the test should be honest)
        client.post(
            f"{BASE}/api/config",
            json={"llm": {"model": "fake-model"}},
            headers={"X-Admin-Token": ADMIN},
        )

        # --- activation ---
        r = client.post(f"{BASE}/api/model/activate")
        check("POST /api/model/activate without token is 401", r.status_code == 401)

        r = client.post(f"{BASE}/api/model/activate", headers={"X-Admin-Token": ADMIN})
        body = r.json()
        check(
            "activate on an unmanaged endpoint probes instead of resuming",
            r.status_code == 200 and body.get("managed") is False and body.get("ok") is True,
            str(body)[:160],
        )

        # --- chat SSE ---
        events: list[dict] = []
        with client.stream(
            "POST", f"{BASE}/api/chat", json={"message": "METAR at LEBL?", "history": []}
        ) as response:
            check("POST /api/chat streams", response.status_code == 200)
            for line in response.iter_lines():
                if line.startswith("data: "):
                    events.append(json.loads(line[6:]))

        kinds = [e["type"] for e in events]
        check("stream includes tools_loaded", "tools_loaded" in kinds, str(kinds))
        check("stream includes reasoning", "reasoning" in kinds)
        check("stream includes tool_call", "tool_call" in kinds)
        check("stream includes tool_result", "tool_result" in kinds)
        check("stream includes final", "final" in kinds)
        check("stream terminates with done", kinds[-1] == "done", str(kinds[-3:]))
        check("no error events", "error" not in kinds, str([e for e in events if e["type"] == "error"])[:200])

        r = client.post(f"{BASE}/api/chat", json={"message": "  "})
        check("empty message is rejected", r.status_code == 400)

    print("-" * 60)
    failed = [label for ok, label in results if not ok]
    print(f"  {len(results) - len(failed)}/{len(results)} passed")
    for label in failed:
        print(f"  FAILED: {label}")
    return 1 if failed else 0


if __name__ == "__main__":
    sys.exit(main())