Kansho/backend/tests/test_frame.py

"""Smoke test for the local product frame. Run from backend/: python tests/test_frame.py"""
from __future__ import annotations

import json
import os
import sys
import tempfile
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT))

os.environ["KANSHO_PROVIDER_KEY"] = ""
os.environ["KANSHO_DETECT_PROVIDER_KEY"] = ""

from fastapi.testclient import TestClient
from main import app


def expect(ok: bool, message: str) -> None:
    if not ok:
        raise SystemExit(f"FAIL: {message}")
    print(f"OK  {message}")


def auth_header(token: str) -> dict:
    return {"X-Auth-Token": token}


def main() -> None:
    seed_prompts = json.loads((ROOT / "config" / "prompts.seed.json").read_text(encoding="utf-8"))
    slugs = {item["slug"] for item in seed_prompts}
    expect(slugs == {"mvp.dialogue_turn", "mvp.journal_generate", "mvp.journal_reconstruct", "mvp.entity_detect", "mvp.profile_review"}, "mvp prompts seeded")

    with TestClient(app) as client:
        health = client.get("/api/health")
        expect(health.status_code == 200 and health.json()["service"] == "kansho", "health")

        status = client.get("/api/auth/status")
        expect(status.json()["needs_setup"] is True, "needs setup before first profile")

        denied = client.get("/api/auth/me")
        expect(denied.status_code == 401, "me without session is 401")

        setup = client.post(
            "/api/auth/setup",
            json={"email": "lars@example.test", "name": "Lars", "password": "test-pass"},
        )
        expect(setup.status_code == 200, f"setup {setup.text}")
        token = setup.json()["token"]
        headers = auth_header(token)
        expect(setup.json()["role"] == "admin", "first profile is admin")

        again = client.post(
            "/api/auth/setup",
            json={"email": "other@example.test", "name": "Other", "password": "test-pass"},
        )
        expect(again.status_code == 409, "second setup is blocked")

        me = client.get("/api/auth/me", headers=headers)
        expect(me.status_code == 200 and me.json()["email"] == "lars@example.test", "me with session")

        spoof = client.get(
            "/api/auth/me",
            headers={**headers, "X-Profile-Id": "someone-else"},
        )
        expect(spoof.json()["id"] == me.json()["id"], "identity comes from session, not profile header")

        login = client.post(
            "/api/auth/login",
            json={"email": "lars@example.test", "password": "test-pass"},
        )
        expect(login.status_code == 200, "login")
        token = login.json()["token"]
        headers = auth_header(token)

        admin = client.get("/api/admin/health", headers=headers)
        expect(admin.status_code == 200, "admin health")
        expect(admin.json()["inventory"]["prompts"] == 5, "five seeded prompts")
        expect(admin.json()["inventory"]["debug_persist_enabled"] is False, "debug persist off by default")
        expect(admin.json()["inventory"]["debug_runs"] == 0, "no debug runs at start")

        debug = client.get("/api/admin/debug", headers=headers)
        expect(debug.status_code == 200 and debug.json()["persist_enabled"] is False, "debug settings default off")

        providers = client.get("/api/admin/providers", headers=headers)
        expect(providers.status_code == 200, "admin providers")
        roles = {item["role"]: item for item in providers.json()["roles"]}
        expect(set(roles) == {"generate", "detect"}, "two provider roles")
        expect(providers.json()["remote_plaintext_allowed"] is True, "test runtime allows remote detect")
        expect(providers.json()["remote_plaintext_reason"] == "non_production", "tests are not production")
        expect(roles["generate"]["ready"] is False, "generate fail-closed without key")
        expect("sk-" not in providers.text, "provider status has no secret")
        expect("KANSHO_PROVIDER_KEY=" not in providers.text, "env assignment not leaked")

        saved = client.put(
            "/api/admin/providers/generate",
            headers=headers,
            json={
                "name": "openrouter",
                "url": "https://openrouter.ai/api/v1/chat/completions",
                "model": "openai/gpt-4o",
                "zdr": True,
                "no_train": True,
            },
        )
        expect(saved.status_code == 200, f"save generate settings {saved.text}")
        saved_roles = {item["role"]: item for item in saved.json()["roles"]}
        expect(saved_roles["generate"]["model"] == "openai/gpt-4o", "generate model persisted")
        expect("profiles" in saved.json(), "llm profiles are listed")
        titles = {item["title"] for item in saved.json()["profiles"]}
        expect("Ollama (LAN)" in titles, "LAN Ollama profile is seeded")
        ollama = next(item for item in saved.json()["profiles"] if item["title"] == "Ollama (LAN)")
        expect(ollama["local"] is True, "LAN Ollama is a local trusted endpoint")
        expect(ollama["url"].startswith("http://192.168.2.144:11434"), "Ollama URL is the agreed LAN host")

        switched = client.put(
            "/api/admin/llm-profiles/" + ollama["id"],
            headers=headers,
            json={
                "title": "Ollama (LAN)",
                "name": "ollama",
                "url": ollama["url"],
                "model": "llama3.1:latest",
                "zdr": True,
                "no_train": True,
            },
        )
        expect(switched.status_code == 200, f"save ollama model {switched.text}")
        activated = client.post(
            "/api/admin/providers/detect/activate",
            headers=headers,
            json={"profile_id": ollama["id"]},
        )
        expect(activated.status_code == 200, f"activate ollama detect {activated.text}")
        detect_role = {item["role"]: item for item in activated.json()["roles"]}["detect"]
        expect(detect_role["local"] is True, "detect uses local ollama after profile switch")
        expect(detect_role["model"] == "llama3.1:latest", "detect keeps the saved ollama model")
        generate_role = {item["role"]: item for item in activated.json()["roles"]}["generate"]
        expect(generate_role["model"] == "openai/gpt-4o", "generate profile stays on its own model")

        unknown = client.put(
            "/api/admin/providers/training",
            headers=headers,
            json={"url": "https://example.test", "model": "x"},
        )
        expect(unknown.status_code == 404, "unknown provider role")

        os.environ["KANSHO_PROVIDER_KEY"] = "sk-test-not-a-real-key"
        try:
            from providers import generate_provider, detect_provider

            expect(generate_provider() is not None, "generate ready after key")
            expect(detect_provider() is not None, "detect reuses generate key")
            ready = client.get("/api/admin/providers", headers=headers)
            expect("sk-test-not-a-real-key" not in ready.text, "secret never in status")
            expect(ready.json()["roles"][0]["ready"] is True, "status ready after key")
        finally:
            os.environ.pop("KANSHO_PROVIDER_KEY", None)

        missing = client.post(
            "/api/prompts/execute",
            headers=headers,
            json={"prompt_id": "demo", "data_class": "B"},
        )
        expect(missing.status_code == 404, "execute requires a stored prompt")

        created = client.post(
            "/api/prompts",
            headers=headers,
            json={
                "slug": "frame-echo",
                "name": "Frame Echo",
                "template": "Zeit {{now}} für [[SELF]] am [[PLACE:HOME]].",
            },
        )
        expect(created.status_code == 200, f"create prompt {created.text}")
        prompt_id = created.json()["id"]

        bad_key = client.post(
            "/api/prompts",
            headers=headers,
            json={"slug": "bad", "name": "Bad", "template": "Hallo {{journal_summary}}"},
        )
        expect(bad_key.status_code == 400, "unknown context key rejected")
        expect(bad_key.json()["detail"]["code"] == "unknown_placeholder", "unknown key code")

        bad_privacy = client.post(
            "/api/prompts",
            headers=headers,
            json={"slug": "bad-privacy", "name": "Bad privacy", "template": "[[INDIAN_WIFE]]"},
        )
        expect(bad_privacy.status_code == 400, "speaking privacy token rejected")

        preview = client.post("/api/prompts/preview", headers=headers, json={"prompt_id": prompt_id})
        expect(preview.status_code == 200, f"preview {preview.text}")
        expect("{{now}}" not in preview.json()["rendered"], "context key resolved")
        expect("[[SELF]]" in preview.json()["rendered"], "privacy token left for gateway")
        expect(preview.json()["llm"] is False, "preview does not call LLM")

        blocked = client.post(
            "/api/prompts/execute",
            headers=headers,
            json={"prompt_id": prompt_id, "data_class": "B"},
        )
        expect(blocked.status_code == 503, f"gateway fail-closed {blocked.text}")
        expect(blocked.json()["detail"]["code"] == "no_egress_provider_configured", "gateway reason")

        class_a = client.post(
            "/api/prompts/execute",
            headers=headers,
            json={"prompt_id": prompt_id, "data_class": "A"},
        )
        expect(class_a.status_code == 503, "class A never leaves")
        expect(class_a.json()["detail"]["code"] == "class_a_never_leaves_local_zone", "class A reason")

        created_user = client.post(
            "/api/users",
            headers=headers,
            json={"email": "user@example.test", "name": "Ute", "password": "user-pass", "role": "user"},
        )
        expect(created_user.status_code == 200, f"admin creates user {created_user.text}")
        expect(created_user.json()["role"] == "user", "created role is user")

        users = client.get("/api/users", headers=headers)
        expect(len(users.json()) == 2, "two profiles")

        user_login = client.post(
            "/api/auth/login",
            json={"email": "user@example.test", "password": "user-pass"},
        )
        expect(user_login.status_code == 200, "user login")
        user_headers = auth_header(user_login.json()["token"])

        forbidden_users = client.get("/api/users", headers=user_headers)
        expect(forbidden_users.status_code == 403, "user cannot list users")

        sub = client.get("/api/subscription", headers=user_headers)
        expect(sub.status_code == 200, "subscription payload")
        expect(sub.json()["tier"]["id"] == "local", "local tier")
        expect(sub.json()["billing"] is None, "no billing provider")
        expect(any(item["feature_id"] == "ai_calls" and item["allowed"] for item in sub.json()["features"]), "ai_calls allowed")

        placeholders = client.get("/api/placeholders", headers=user_headers)
        expect(placeholders.status_code == 200, "placeholder catalog")
        keys = {item["key"] for item in placeholders.json()["context"]}
        expect(
            keys
            == {
                "now",
                "dialogue_context",
                "existing_text",
                "known_labels",
                "register_hint",
                "opening_hint",
                "reconstruction",
                "review_package",
                "source_text",
                "space_title",
                "writing_profile",
                "style_examples",
                "transformation_instructions",
                "detail_instructions",
                "voice_instructions",
                "narrative_instructions",
                "interaction_hint",
            },
            "system and mvp context keys",
        )

        last_admin = client.patch(
            f"/api/users/{me.json()['id']}",
            headers=headers,
            json={"role": "user"},
        )
        expect(last_admin.status_code == 400, "last admin cannot be demoted")

        disabled = client.patch(
            f"/api/users/{created_user.json()['id']}",
            headers=headers,
            json={"status": "disabled"},
        )
        expect(disabled.status_code == 200, "disable user")
        disabled_login = client.post(
            "/api/auth/login",
            json={"email": "user@example.test", "password": "user-pass"},
        )
        expect(disabled_login.status_code == 401, "disabled user cannot login")

        bad_login = client.post(
            "/api/auth/login",
            json={"email": "lars@example.test", "password": "wrong"},
        )
        expect(bad_login.status_code == 401, "bad password")

    print("All frame tests passed.")


if __name__ == "__main__":
    main()