nisse/nisse-core/tests/test_catalog.py

"""Runtime model discovery (architecture L9).

The point of these tests is that nothing here reads the model's *name*. A
semantic alias like "remote-architecture" is free to point anywhere, so the
route has to come from what LiteLLM says the model is.
"""

import httpx
import pytest

from nisse_core.models.catalog import (
    CatalogUnavailable,
    ModelCatalog,
    UnknownModel,
    select_client,
)


def entry(alias: str, backing: str, provider: str | None, tools: bool | None = None):
    """One /model/info row, shaped as the live proxy returns it."""
    info: dict = {"supports_function_calling": tools}
    if provider is not None:
        info["litellm_provider"] = provider
    return {
        "model_name": alias,
        "litellm_params": {"model": backing, "api_base": "http://proxy.test/v1"},
        "model_info": info,
    }


def catalog(*entries) -> ModelCatalog:
    return ModelCatalog.from_payload({"data": list(entries)})


# --- parsing -------------------------------------------------------------


def test_facts_are_read_from_the_payload() -> None:
    cat = catalog(
        entry("claude-haiku-4-5", "anthropic/claude-haiku-4-5", "anthropic", True)
    )
    facts = cat.facts("claude-haiku-4-5")
    assert facts.backing == "anthropic/claude-haiku-4-5"
    assert facts.provider == "anthropic"
    assert facts.supports_tools is True
    assert facts.native_anthropic


def test_provider_falls_back_to_the_backing_prefix() -> None:
    # litellm_provider is absent on some deployments; the backing model's
    # own "vendor/model" prefix says the same thing.
    cat = catalog(entry("mystery", "anthropic/claude-opus-5", None))
    assert cat.facts("mystery").provider == "anthropic"


def test_unnamed_entries_are_skipped() -> None:
    cat = ModelCatalog.from_payload({"data": [{"litellm_params": {"model": "x"}}]})
    assert cat.aliases() == []


def test_missing_tool_support_is_unknown_not_false() -> None:
    # LiteLLM reports null for models it has no capability data on. That is
    # not the same as declaring the model tool-less, and the startup warning
    # depends on the difference.
    cat = catalog(entry("local-fast", "openai/qwen3-8b-awq", "openai", None))
    assert cat.facts("local-fast").supports_tools is None


# --- route selection by provider, never by name --------------------------


def test_a_semantic_alias_routes_by_its_provider() -> None:
    # The whole feature: nothing in "remote-architecture" says Anthropic.
    cat = catalog(entry("remote-architecture", "anthropic/claude-opus-5", "anthropic"))
    normalized, passthrough = object(), object()
    chosen = select_client("remote-architecture", normalized, passthrough, cat)
    assert chosen is passthrough


def test_the_same_alias_reroutes_when_it_is_repointed() -> None:
    # L9: provisionally accept LiteLLM's normalization for a vendor with no
    # native adapter, rather than sending it Anthropic-shaped bodies.
    cat = catalog(entry("remote-architecture", "openai/gpt-6-astra", "openai"))
    normalized, passthrough = object(), object()
    assert (
        select_client("remote-architecture", normalized, passthrough, cat) is normalized
    )


def test_a_name_that_merely_mentions_a_vendor_is_not_that_vendor() -> None:
    cat = catalog(entry("claude-clone-9", "openai/some-clone", "openai"))
    normalized, passthrough = object(), object()
    assert select_client("claude-clone-9", normalized, passthrough, cat) is normalized


def test_local_models_take_the_normalized_route() -> None:
    cat = catalog(
        entry("local-coder", "openai/qwen3-14b", "openai"),
        entry("local-fast", "openai/qwen3-8b-awq", "openai"),
    )
    normalized, passthrough = object(), object()
    for alias in ("local-coder", "local-fast"):
        assert select_client(alias, normalized, passthrough, cat) is normalized


# --- misconfiguration is loud --------------------------------------------


def test_an_unknown_model_raises_rather_than_guessing_a_route() -> None:
    cat = catalog(entry("local-fast", "openai/qwen3-8b-awq", "openai"))
    with pytest.raises(UnknownModel, match="local-fsat"):
        select_client("local-fsat", object(), object(), cat)


def test_the_error_lists_what_is_available() -> None:
    cat = catalog(
        entry("local-fast", "openai/qwen3-8b-awq", "openai"),
        entry("claude-opus-5", "anthropic/claude-opus-5", "anthropic"),
    )
    with pytest.raises(UnknownModel, match="claude-opus-5, local-fast"):
        cat.facts("typo")


def test_validate_names_every_bad_role_at_once() -> None:
    # One boot should surface the whole list, not make the operator
    # rediscover the next typo on the next start.
    cat = catalog(entry("local-fast", "openai/qwen3-8b-awq", "openai"))
    with pytest.raises(UnknownModel) as caught:
        cat.validate(
            {
                "executor": "local-fast",
                "planner": "clause-opus-5",
                "helper": "remote-architecture",
                "expert": "local-fast",
            }
        )
    message = str(caught.value)
    assert "planner='clause-opus-5'" in message
    assert "helper='remote-architecture'" in message
    assert "executor" not in message  # the good ones are not noise


def test_validate_passes_a_whole_good_mapping() -> None:
    cat = catalog(
        entry("local-fast", "openai/qwen3-8b-awq", "openai"),
        entry("claude-opus-5", "anthropic/claude-opus-5", "anthropic"),
    )
    cat.validate({"executor": "local-fast", "planner": "claude-opus-5"})


# --- an outage is not a misconfiguration ---------------------------------


def test_an_unreachable_proxy_is_reported_as_an_outage() -> None:
    def handler(request: httpx.Request) -> httpx.Response:
        raise httpx.ConnectError("no route to host")

    with pytest.raises(CatalogUnavailable, match="could not reach"):
        ModelCatalog.discover(
            "http://litellm.test", "sk-test", transport=httpx.MockTransport(handler)
        )


def test_a_refused_listing_is_reported_as_an_outage() -> None:
    cat = httpx.MockTransport(lambda request: httpx.Response(401, text="bad key"))
    with pytest.raises(CatalogUnavailable, match="401"):
        ModelCatalog.discover("http://litellm.test", "sk-test", transport=cat)


def test_discovery_asks_the_proxy_with_the_token() -> None:
    seen: dict = {}

    def handler(request: httpx.Request) -> httpx.Response:
        seen["url"] = str(request.url)
        seen["auth"] = request.headers.get("Authorization")
        return httpx.Response(
            200, json={"data": [entry("local-fast", "openai/qwen3-8b-awq", "openai")]}
        )

    cat = ModelCatalog.discover(
        "http://litellm.test/", "sk-test", transport=httpx.MockTransport(handler)
    )
    assert seen["url"] == "http://litellm.test/model/info"
    assert seen["auth"] == "Bearer sk-test"
    assert cat.aliases() == ["local-fast"]


# --- what gets pinned ----------------------------------------------------


def test_pinning_records_what_the_alias_resolved_to() -> None:
    # An alias may point somewhere else tomorrow; eval history (L9) keyed on
    # the alias alone would average across whatever it pointed at.
    cat = catalog(entry("remote-architecture", "anthropic/claude-opus-5", "anthropic"))
    pinned = cat.pinned({"planner": "remote-architecture"})
    assert pinned == {
        "planner": {
            "alias": "remote-architecture",
            "backing": "anthropic/claude-opus-5",
            "provider": "anthropic",
        }
    }