tests/test_app.py

"""Tests for kinchat.server.app — FastAPI HTTP face of the env.

Uses FastAPI's TestClient for sync routes (the route handlers are async, but
TestClient handles awaiting them under the hood). Each test resets the module-
level _ENV to a fresh KinChatEnvironment so tests don't leak state between
each other.
"""
from __future__ import annotations

from unittest.mock import AsyncMock, MagicMock

import pytest
from fastapi.testclient import TestClient

import kinchat
from kinchat.models import (
    KinChatAction,
    KinChatObservation,
    KinChatState,
)
from kinchat.server import app as app_module
from kinchat.server.environment import KinChatEnvironment, StubPersonaSurface
from kinchat.server.grader import RewardBreakdown
from kinchat.server.scenarios import ALL_SCENARIOS, ARCHETYPES


# --------------------------------------------------------------------------- #
# Fixtures                                                                    #
# --------------------------------------------------------------------------- #


def _mk_breakdown(scalar: float = 0.6) -> RewardBreakdown:
    return RewardBreakdown(
        leak=0.9,
        audience_fit=0.7,
        restraint=0.5,
        trust_delta=0.5,
        scalar=scalar,
        feedback="ok",
    )


def _mock_grader() -> MagicMock:
    g = MagicMock()
    g.grade_turn = AsyncMock(return_value=_mk_breakdown())
    return g


@pytest.fixture
def client():
    """Reset the module-level env to a fresh, mocked one for each test."""
    env = KinChatEnvironment(
        grader=_mock_grader(),
        persona_surface=StubPersonaSurface(),
    )
    app_module._ENV = env
    with TestClient(app_module.app) as c:
        # TestClient triggers the lifespan startup, which rebuilds _ENV. We
        # stomp it again *after* startup so our mock survives.
        app_module._ENV = env
        yield c


@pytest.fixture
def scenario_id() -> str:
    return ALL_SCENARIOS[0].id


# --------------------------------------------------------------------------- #
# Health                                                                      #
# --------------------------------------------------------------------------- #


def test_health_ok(client):
    r = client.get("/health")
    assert r.status_code == 200
    body = r.json()
    assert body["status"] == "ok"
    assert body["version"] == kinchat.__version__


# --------------------------------------------------------------------------- #
# Reset                                                                       #
# --------------------------------------------------------------------------- #


def test_reset_returns_observation(client, scenario_id):
    r = client.post(
        "/reset",
        json={"scenario_id": scenario_id, "session_id": "sess_test"},
    )
    assert r.status_code == 200
    obs = KinChatObservation.model_validate(r.json())
    assert obs.turn_index == 0
    assert obs.scenario_brief
    assert obs.done is False


def test_reset_unknown_scenario_id_4xx(client):
    r = client.post(
        "/reset",
        json={"scenario_id": "does_not_exist_xyz"},
    )
    assert r.status_code in (400, 404, 422)


def test_reset_session_id_optional(client, scenario_id):
    r = client.post("/reset", json={"scenario_id": scenario_id})
    assert r.status_code == 200


# --------------------------------------------------------------------------- #
# Step                                                                        #
# --------------------------------------------------------------------------- #


def test_step_after_reset_returns_observation(client, scenario_id):
    client.post("/reset", json={"scenario_id": scenario_id, "session_id": "sess_step"})
    payload = {
        "action_type": "send",
        "message": "Hi mom!",
        "recipients": ["mom"],
        "reasoning": "greeting",
    }
    r = client.post("/step", json=payload)
    assert r.status_code == 200
    body = r.json()
    obs = KinChatObservation.model_validate(body)
    assert "leak" in obs.reward_breakdown
    assert "audience_fit" in obs.reward_breakdown
    assert obs.turn_index == 1
    assert isinstance(obs.done, bool)


def test_step_before_reset_400(client):
    # Override env with a fresh one that has not been reset.
    fresh = KinChatEnvironment(
        grader=_mock_grader(), persona_surface=StubPersonaSurface()
    )
    app_module._ENV = fresh
    payload = {
        "action_type": "stay_silent",
        "message": "",
        "recipients": [],
        "reasoning": "",
    }
    r = client.post("/step", json=payload)
    assert r.status_code == 400


def test_step_invalid_action_type_422(client, scenario_id):
    client.post("/reset", json={"scenario_id": scenario_id})
    bad = {
        "action_type": "shout",  # not in Literal
        "message": "",
        "recipients": [],
        "reasoning": "",
    }
    r = client.post("/step", json=bad)
    assert r.status_code == 422


# --------------------------------------------------------------------------- #
# State                                                                       #
# --------------------------------------------------------------------------- #


def test_state_after_reset(client, scenario_id):
    client.post("/reset", json={"scenario_id": scenario_id, "session_id": "sess_st"})
    r = client.get("/state")
    assert r.status_code == 200
    s = KinChatState.model_validate(r.json())
    assert s.scenario_id == scenario_id
    assert s.session_id == "sess_st"
    assert s.episode_index == 0


def test_state_before_reset_400(client):
    # Brand new env, never reset.
    app_module._ENV = KinChatEnvironment(
        grader=_mock_grader(), persona_surface=StubPersonaSurface()
    )
    r = client.get("/state")
    assert r.status_code == 400


# --------------------------------------------------------------------------- #
# Scenarios                                                                   #
# --------------------------------------------------------------------------- #


def test_scenarios_listing(client):
    r = client.get("/scenarios")
    assert r.status_code == 200
    body = r.json()
    assert set(body["archetypes"]) == set(ARCHETYPES)
    assert len(body["archetypes"]) == 6
    assert len(body["scenarios"]) == 30
    first = body["scenarios"][0]
    assert {"id", "archetype", "brief"}.issubset(first.keys())


# --------------------------------------------------------------------------- #
# Dashboard                                                                   #
# --------------------------------------------------------------------------- #


def test_dashboard_html(client):
    r = client.get("/dashboard")
    assert r.status_code == 200
    assert "text/html" in r.headers.get("content-type", "")
    assert "KinChat" in r.text


def test_dashboard_html_has_chartjs_and_polling(client):
    r = client.get("/dashboard")
    assert r.status_code == 200
    body = r.text
    # Chart.js CDN script tag (major-pin) is present.
    assert "cdn.jsdelivr.net/npm/chart.js@4" in body
    # Polling URL appears.
    assert "/dashboard/data" in body
    # Heading.
    assert "KinChat" in body


def test_dashboard_html_references_all_personas(client):
    r = client.get("/dashboard")
    body = r.text
    for pid in ("mom", "dad", "sib1", "sib2", "grandma"):
        assert pid in body, f"missing persona id {pid!r} in dashboard.html"


def test_dashboard_html_references_all_rubrics(client):
    r = client.get("/dashboard")
    body = r.text
    for label in ("leak", "audience_fit", "restraint", "trust_delta"):
        assert label in body, f"missing rubric label {label!r} in dashboard.html"


def test_dashboard_html_has_canvas_for_each_chart(client):
    """The dashboard should have at least one <canvas> for each main chart panel."""
    import re

    r = client.get("/dashboard")
    body = r.text
    canvas_ids = set(re.findall(r'<canvas[^>]*id="([^"]+)"', body))
    # Three required chart panels: trust, leaks, rewards.
    assert "trust-chart" in canvas_ids
    assert "leaks-chart" in canvas_ids
    assert "rewards-chart" in canvas_ids


def test_dashboard_data_idle_before_reset(client):
    # Force an env that has never been reset.
    app_module._ENV = KinChatEnvironment(
        grader=_mock_grader(), persona_surface=StubPersonaSurface()
    )
    r = client.get("/dashboard/data")
    assert r.status_code == 200
    body = r.json()
    assert body["session_id"] is None
    assert body["episode_index"] == 0
    assert body["turn_index"] == 0
    assert body["chat_log"] == []
    assert body["rewards_per_turn"] == []
    assert body["leak_events"] == []
    assert body["family_trust"] == {}


def test_dashboard_data_after_reset_and_step(client, scenario_id):
    client.post(
        "/reset",
        json={"scenario_id": scenario_id, "session_id": "sess_dash"},
    )
    client.post(
        "/step",
        json={
            "action_type": "send",
            "message": "hello family",
            "recipients": ["mom", "grandma"],
            "reasoning": "small talk",
        },
    )
    r = client.get("/dashboard/data")
    assert r.status_code == 200
    body = r.json()
    assert body["session_id"] == "sess_dash"
    assert body["turn_index"] == 1
    assert len(body["chat_log"]) >= 1
    assert body["family_trust"]  # non-empty
    assert set(body["family_trust"].keys()) >= {"mom", "dad", "sib1", "sib2", "grandma"}
    assert isinstance(body["rewards_per_turn"], list)
    assert len(body["rewards_per_turn"]) == 1
    entry = body["rewards_per_turn"][0]
    assert entry["turn"] == 0
    assert "leak" in entry and "scalar" in entry
    assert "cache_stats" in body


# --------------------------------------------------------------------------- #
# Routes-as-paths sanity check                                                #
# --------------------------------------------------------------------------- #


def test_reserved_names_are_http_paths_not_mcp_tools(client):
    """The OpenEnv MCP guard reserves the names reset/step/state/close as
    tool names. We expose them as HTTP routes only — verify each is hit-able.
    """
    routes = {r.path for r in app_module.app.routes if hasattr(r, "path")}
    assert "/reset" in routes
    assert "/step" in routes
    assert "/state" in routes
    assert "/health" in routes
    assert "/scenarios" in routes
    assert "/dashboard" in routes
    assert "/dashboard/data" in routes


def test_app_importable_without_openai_key(monkeypatch):
    """Importing kinchat.server.app must not require OPENAI_API_KEY."""
    monkeypatch.delenv("OPENAI_API_KEY", raising=False)
    monkeypatch.delenv("HF_TOKEN", raising=False)
    monkeypatch.delenv("API_KEY", raising=False)
    # Re-importing isn't easy — just verify the build helper works.
    env = app_module._build_env()
    assert isinstance(env, KinChatEnvironment)