conjurer/tests/unit/test_ai_provider_switch.py

"""Unit tests for the GPT/Claude provider switch.

The unit CI job installs only pytest, so the heavy runtime deps that
``ai_functions`` imports unguarded (``openai``, ``tiktoken``, ``other_functions``
-> ``discord``) are stubbed *only when genuinely absent*. Locally, where the
real packages exist, the stubs are skipped and the real modules are used.
"""
import sys
import types


def _stub_if_missing(name: str, build):
    if name in sys.modules:
        return
    try:  # real package present (local dev / bot image) -> use it
        __import__(name)
    except ImportError:
        sys.modules[name] = build()


def _build_tiktoken():
    mod = types.ModuleType("tiktoken")

    class _Enc:
        def encode(self, text):
            return list(text)

    mod.encoding_for_model = lambda _model: _Enc()
    return mod


def _build_other_functions():
    mod = types.ModuleType("other_functions")

    async def _noop(*_a, **_k):
        return None

    mod.discord_friendly_send = _noop
    mod.discord_friendly_reply = _noop
    return mod


def _build_openai():
    mod = types.ModuleType("openai")
    for cls_name in (
        "APITimeoutError",
        "APIConnectionError",
        "BadRequestError",
        "APIResponseValidationError",
        "AuthenticationError",
        "PermissionDeniedError",
        "RateLimitError",
        "UnprocessableEntityError",
        "APIError",
        "OpenAIError",
    ):
        setattr(mod, cls_name, type(cls_name, (Exception,), {}))
    return mod


_stub_if_missing("tiktoken", _build_tiktoken)
_stub_if_missing("other_functions", _build_other_functions)
_stub_if_missing("openai", _build_openai)

import ai_functions  # noqa: E402  (import after stubbing)
import openai  # noqa: E402  (real or stub, same object ai_functions uses)


def _reset_active(name="gpt"):
    ai_functions._ACTIVE_CONFIG_NAME = name


def test_split_extracts_system_and_starts_with_user():
    system, convo = ai_functions._to_anthropic_messages(
        [
            {"role": "system", "content": "SYS1"},
            {"role": "system", "content": "SYS2"},
            {"role": "assistant", "content": "leading-assistant-dropped"},
            {"role": "user", "content": "u1"},
            {"role": "assistant", "content": "a1"},
            {"role": "user", "content": "hi"},
        ]
    )
    assert system == "SYS1\n\nSYS2"
    assert convo[0] == {"role": "user", "content": "u1"}
    assert convo == [
        {"role": "user", "content": "u1"},
        {"role": "assistant", "content": "a1"},
        {"role": "user", "content": "hi"},
    ]


def test_split_synthesises_user_turn_when_only_system():
    _system, convo = ai_functions._to_anthropic_messages(
        [{"role": "system", "content": "only"}]
    )
    assert convo == [{"role": "user", "content": " "}]


def test_select_model_gpt_active():
    _reset_active("gpt")
    # legacy default auto-selects; MUSIC is cheap; explicit ids are honoured
    assert ai_functions.select_model("GENERAL", "gpt-4o") == "gpt-4o"
    assert ai_functions.select_model("MUSIC", "gpt-4o") == "gpt-4o-mini"
    assert ai_functions.select_model("MUSIC", "auto") == "gpt-4o-mini"
    assert ai_functions.select_model("GENERAL", "o1-preview") == "o1-preview"


def test_select_model_claude_active_maps_legacy_default():
    _reset_active("claude")
    try:
        # the old gpt-4o default must not leak to Claude - it auto-maps
        assert ai_functions.select_model("GENERAL", "gpt-4o") == "claude-opus-4-8"
        assert ai_functions.select_model("MUSIC", "gpt-4o") == "claude-haiku-4-5"
        # a real, deliberate model id is still honoured verbatim
        assert ai_functions.select_model("GENERAL", "claude-sonnet-5") == "claude-sonnet-5"
    finally:
        _reset_active("gpt")


def test_list_ai_configs_hides_templates():
    names = ai_functions.list_ai_configs()
    assert "_template" not in names
    assert {"gpt", "claude"}.issubset(set(names))


def _bare(cls):
    # Build an instance without invoking __init__ - the real openai SDK
    # exceptions require response/body kwargs, the CI stubs don't. isinstance
    # (all _map_openai_error cares about) works on __new__-created instances.
    return cls.__new__(cls)


def test_map_openai_error_categories():
    assert ai_functions._map_openai_error(_bare(openai.AuthenticationError)).category == "auth"
    assert ai_functions._map_openai_error(_bare(openai.RateLimitError)).category == "rate_limit"
    assert ai_functions._map_openai_error(_bare(openai.APITimeoutError)).category == "timeout"
    assert ai_functions._map_openai_error(ValueError("x")).category == "api"


# ---------------------------------------------------------------- Ollama
# The self-hosted backend is wired through Ollama's OpenAI-compatible surface,
# so it reuses the message format and the error mapping above. What is new and
# worth pinning: it must appear in the picker even on an upgraded settings file,
# models come from the SERVER, and pinning one must stick.
import asyncio  # noqa: E402


class _FakeModel:
    def __init__(self, ident):
        self.id = ident


class _FakeModels:
    def __init__(self, ids, raises=None):
        self._ids = ids
        self._raises = raises

    async def list(self):
        if self._raises:
            raise self._raises
        return types.SimpleNamespace(data=[_FakeModel(i) for i in self._ids])


class _FakeOllamaClient:
    def __init__(self, ids=(), raises=None):
        self.models = _FakeModels(list(ids), raises)


def test_ollama_config_is_offered_in_the_picker():
    assert "ollama" in ai_functions.list_ai_configs()
    cfg = ai_functions.AI_CONFIGS["ollama"]
    assert cfg["provider"] == "ollama"


def test_select_model_uses_ollama_models_when_active(monkeypatch):
    monkeypatch.setitem(
        ai_functions.AI_CONFIGS,
        "ollama",
        {"provider": "ollama", "latest_model": "llama3.1:8b", "cheap_model": "qwen2.5:3b"},
    )
    _reset_active("ollama")
    try:
        # the legacy gpt-4o default must auto-map, not leak to Ollama
        assert ai_functions.select_model("GENERAL", "gpt-4o") == "llama3.1:8b"
        assert ai_functions.select_model("MUSIC", "gpt-4o") == "qwen2.5:3b"
        # an explicit id is still honoured verbatim
        assert ai_functions.select_model("GENERAL", "mistral:7b") == "mistral:7b"
    finally:
        _reset_active("gpt")


def test_list_provider_models_queries_the_ollama_server(monkeypatch):
    monkeypatch.setattr(
        ai_functions, "OLLAMACLIENT", _FakeOllamaClient(["b:2", "a:1", "a:1"])
    )
    models = asyncio.run(ai_functions.list_provider_models("ollama"))
    assert models == ["a:1", "b:2"]  # sorted + de-duplicated


def test_list_provider_models_for_hosted_provider_reports_configured_ids():
    models = asyncio.run(ai_functions.list_provider_models("gpt"))
    assert models == [
        ai_functions.AI_CONFIGS["gpt"]["latest_model"],
        ai_functions.AI_CONFIGS["gpt"]["cheap_model"],
    ]


def test_list_provider_models_wraps_server_failure(monkeypatch):
    monkeypatch.setattr(
        ai_functions, "OLLAMACLIENT", _FakeOllamaClient(raises=ValueError("boom"))
    )
    try:
        asyncio.run(ai_functions.list_provider_models("ollama"))
    except ai_functions.AIError as exc:
        assert exc.category == "api"
    else:
        raise AssertionError("a server failure must surface as AIError")


def test_list_provider_models_without_endpoint_is_an_auth_error(monkeypatch):
    monkeypatch.setattr(ai_functions, "OLLAMACLIENT", None)
    try:
        asyncio.run(ai_functions.list_provider_models("ollama"))
    except ai_functions.AIError as exc:
        assert exc.category == "auth"
    else:
        raise AssertionError("an unconfigured Ollama must surface as AIError")


def test_set_active_model_pins_latest_and_keeps_cheap(monkeypatch):
    monkeypatch.setitem(
        ai_functions.AI_CONFIGS,
        "ollama",
        {"provider": "ollama", "latest_model": "old:1", "cheap_model": "cheap:1"},
    )
    written = {}
    monkeypatch.setattr(
        ai_functions, "_persist_active_ai_config", lambda name: written.update(name=name)
    )
    cfg = ai_functions.set_active_model("mistral:7b", "ollama")
    assert cfg["latest_model"] == "mistral:7b"
    assert cfg["cheap_model"] == "cheap:1"   # MUSIC path untouched
    assert written  # the choice was persisted


def test_set_active_model_rejects_blank_and_unknown_config(monkeypatch):
    monkeypatch.setattr(ai_functions, "_persist_active_ai_config", lambda _n: None)
    for bad in ("", "   "):
        try:
            ai_functions.set_active_model(bad, "gpt")
        except ValueError:
            pass
        else:
            raise AssertionError("a blank model id must be rejected")
    try:
        ai_functions.set_active_model("x", "nie-ma-takiego")
    except KeyError:
        pass
    else:
        raise AssertionError("an unknown config must be rejected")


def test_switching_to_ollama_without_endpoint_explains_itself(monkeypatch):
    monkeypatch.setattr(ai_functions, "OLLAMACLIENT", None)
    try:
        ai_functions.set_active_ai_config("ollama")
    except RuntimeError as exc:
        assert "CONJURER_OLLAMA_URL" in str(exc)
    else:
        raise AssertionError("switching to an unconfigured Ollama must raise")
    finally:
        _reset_active("gpt")


def test_provider_generate_routes_to_ollama(monkeypatch):
    monkeypatch.setitem(
        ai_functions.AI_CONFIGS,
        "ollama",
        {"provider": "ollama", "latest_model": "m:1", "cheap_model": "m:1"},
    )
    _reset_active("ollama")
    seen = {}

    async def _fake_ollama_call(messages, model, cfg):
        seen.update(model=model, messages=messages)
        return "odpowiedź z domu"

    monkeypatch.setattr(ai_functions, "_ollama_call", _fake_ollama_call)
    try:
        out = asyncio.run(
            ai_functions.provider_generate([{"role": "user", "content": "hej"}], "m:1")
        )
    finally:
        _reset_active("gpt")
    assert out == "odpowiedź z domu"
    assert seen["model"] == "m:1"