Kansho/backend/tests/test_model_catalog.py

"""Network-free OpenRouter catalog tests. Run from backend/: python tests/test_model_catalog.py"""
from __future__ import annotations

import os
import sys
import tempfile
from pathlib import Path
from unittest.mock import patch

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

os.environ.setdefault("KANSHO_PROVIDER_KEY", "")
os.environ.setdefault("KANSHO_DETECT_PROVIDER_KEY", "")
Path(tempfile.gettempdir()).mkdir(parents=True, exist_ok=True)

from model_catalog import (
    ModelWindow,
    catalog_ttl,
    list_provider_models,
    reset_catalog,
    resolve_generate_metadata,
)
from prompt_budget import ERROR_MODEL_METADATA_UNKNOWN, JournalBudgetError
from providers import ProviderConfig


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


class FakeResponse:
    def __init__(self, payload, status_code: int = 200, text: str = ""):
        self._payload = payload
        self.status_code = status_code
        self.text = text

    def json(self):
        if isinstance(self._payload, Exception):
            raise self._payload
        return self._payload


def _openrouter_config(model: str = "openai/gpt-4o") -> ProviderConfig:
    return ProviderConfig(
        role="generate",
        name="openrouter",
        mode="http",
        url="https://openrouter.ai/api/v1/chat/completions",
        model=model,
        key="sk-test",
        local=False,
        zdr=True,
        no_train=True,
    )


def _clear_env(*names: str) -> dict[str, str | None]:
    previous = {}
    for name in names:
        previous[name] = os.environ.pop(name, None)
    return previous


def _restore_env(previous: dict[str, str | None]) -> None:
    for name, value in previous.items():
        if value is None:
            os.environ.pop(name, None)
        else:
            os.environ[name] = value


def test_reads_context_and_completion() -> None:
    reset_catalog()
    previous = _clear_env("KANSHO_PROVIDER_CONTEXT_LENGTH", "KANSHO_PROVIDER_MAX_COMPLETION_TOKENS")
    payload = {
        "data": [
            {
                "id": "openai/gpt-4o",
                "context_length": 128000,
                "top_provider": {"max_completion_tokens": 16384},
            }
        ]
    }
    with patch("model_catalog.httpx.get", lambda *args, **kwargs: FakeResponse(payload)):
        window = resolve_generate_metadata(_openrouter_config())
    expect(window.context_length == 128000, "catalog reads context_length")
    expect(window.max_completion_tokens == 16384, "catalog reads max completion")
    expect(window.source == "openrouter_models", "window is from the catalog")
    _restore_env(previous)


def test_model_not_in_catalog() -> None:
    reset_catalog()
    previous = _clear_env("KANSHO_PROVIDER_CONTEXT_LENGTH", "KANSHO_PROVIDER_MAX_COMPLETION_TOKENS")
    payload = {"data": [{"id": "other/model", "context_length": 32000, "top_provider": {"max_completion_tokens": 4096}}]}
    with patch("model_catalog.httpx.get", lambda *args, **kwargs: FakeResponse(payload)):
        try:
            resolve_generate_metadata(_openrouter_config("missing/model"))
            raise SystemExit("FAIL: missing catalog model should fail closed")
        except JournalBudgetError as exc:
            expect(exc.code == ERROR_MODEL_METADATA_UNKNOWN, "model not in catalog is fail-closed")
            expect((exc.diagnostics or {}).get("reason") == "model_not_in_catalog", "reason is model_not_in_catalog")
    _restore_env(previous)


def test_incomplete_metadata() -> None:
    reset_catalog()
    previous = _clear_env("KANSHO_PROVIDER_CONTEXT_LENGTH", "KANSHO_PROVIDER_MAX_COMPLETION_TOKENS")
    payload = {"data": [{"id": "openai/gpt-4o", "top_provider": {"max_completion_tokens": 4096}}]}
    with patch("model_catalog.httpx.get", lambda *args, **kwargs: FakeResponse(payload)):
        try:
            resolve_generate_metadata(_openrouter_config())
            raise SystemExit("FAIL: incomplete metadata should fail closed")
        except JournalBudgetError as exc:
            expect(exc.code == ERROR_MODEL_METADATA_UNKNOWN, "incomplete metadata is fail-closed")
    _restore_env(previous)


def test_cache_hit_and_ttl_expiry() -> None:
    reset_catalog()
    previous = _clear_env("KANSHO_PROVIDER_CONTEXT_LENGTH", "KANSHO_PROVIDER_MAX_COMPLETION_TOKENS")
    calls = {"n": 0}

    def fake_get(*args, **kwargs):
        calls["n"] += 1
        return FakeResponse(
            {
                "data": [
                    {
                        "id": "openai/gpt-4o",
                        "context_length": 64000 + calls["n"],
                        "top_provider": {"max_completion_tokens": 4096},
                    }
                ]
            }
        )

    with patch("model_catalog.httpx.get", fake_get):
        first = resolve_generate_metadata(_openrouter_config(), now=1000.0)
        second = resolve_generate_metadata(_openrouter_config(), now=1000.0 + catalog_ttl() - 1)
        expect(calls["n"] == 1, "catalog is cached within TTL")
        expect(first.context_length == second.context_length, "cache hit returns the same window")
        expect(second.cached is True, "cached window is marked cached")
        third = resolve_generate_metadata(_openrouter_config(), now=1000.0 + catalog_ttl() + 1)
        expect(calls["n"] == 2, "catalog is refetched after TTL")
        expect(third.context_length != first.context_length, "refetch reads a new catalog row")
    _restore_env(previous)


def test_env_fallback_and_fail_closed() -> None:
    reset_catalog()
    previous = _clear_env("KANSHO_PROVIDER_CONTEXT_LENGTH", "KANSHO_PROVIDER_MAX_COMPLETION_TOKENS")
    payload = {"data": [{"id": "other/model", "context_length": 32000, "top_provider": {"max_completion_tokens": 2048}}]}
    os.environ["KANSHO_PROVIDER_CONTEXT_LENGTH"] = "65536"
    os.environ["KANSHO_PROVIDER_MAX_COMPLETION_TOKENS"] = "8192"
    with patch("model_catalog.httpx.get", lambda *args, **kwargs: FakeResponse(payload)):
        window = resolve_generate_metadata(_openrouter_config("missing/model"))
    expect(window.context_length == 65536, "documented env fallback supplies the window")
    expect(window.max_completion_tokens == 8192, "env fallback supplies completion limit")
    expect(window.source == "env_fallback", "fallback is labeled as env_fallback")

    reset_catalog()
    os.environ.pop("KANSHO_PROVIDER_CONTEXT_LENGTH", None)
    os.environ.pop("KANSHO_PROVIDER_MAX_COMPLETION_TOKENS", None)
    with patch("model_catalog.httpx.get", lambda *args, **kwargs: FakeResponse(payload)):
        try:
            resolve_generate_metadata(_openrouter_config("missing/model"))
            raise SystemExit("FAIL: no window should fail closed")
        except JournalBudgetError as exc:
            expect(exc.code == ERROR_MODEL_METADATA_UNKNOWN, "no window is fail-closed")

    reset_catalog()
    with patch("model_catalog.httpx.get", lambda *args, **kwargs: FakeResponse({"nope": True})):
        try:
            resolve_generate_metadata(_openrouter_config())
            raise SystemExit("FAIL: bad catalog shape should fail closed")
        except JournalBudgetError as exc:
            expect(exc.code == ERROR_MODEL_METADATA_UNKNOWN, "bad metadata shape is fail-closed")
    _restore_env(previous)


def test_list_provider_models() -> None:
    reset_catalog()
    payload = {
        "data": [
            {"id": "openai/gpt-4.1-nano", "name": "GPT 4.1 Nano"},
            {"id": "openai/gpt-5.4", "context_length": 128000},
            {"id": "openai/gpt-4.1-nano"},
        ]
    }
    with patch("model_catalog.httpx.get", lambda *args, **kwargs: FakeResponse(payload)):
        items = list_provider_models(_openrouter_config())
    expect([item["id"] for item in items] == ["openai/gpt-4.1-nano", "openai/gpt-5.4"], "picker lists unique ids")
    with patch("model_catalog.httpx.get", lambda *args, **kwargs: FakeResponse({"nope": True})):
        expect(list_provider_models(_openrouter_config()) == [], "picker soft-fails on bad catalog")
    fake = ProviderConfig(
        role="generate",
        name="fake",
        mode="fake",
        url="",
        model="fake",
        key="",
        local=True,
        zdr=True,
        no_train=True,
    )
    expect(list_provider_models(fake) == [], "fake provider has no catalog")


def main() -> None:
    test_reads_context_and_completion()
    test_model_not_in_catalog()
    test_incomplete_metadata()
    test_cache_hit_and_ttl_expiry()
    test_env_fallback_and_fail_closed()
    test_list_provider_models()
    print("model catalog tests passed.")


if __name__ == "__main__":
    main()