opensquilla--opensquilla/tests/test_onboarding/test_mutations_keep_current.py

"""Keep-current semantics for onboarding mutations.

Omitted parameters (``None``) must never reset stored provider, search, or
router settings — re-saving for a key rotation has to be loss-free.
"""

from __future__ import annotations

from opensquilla.gateway.config import GatewayConfig
from opensquilla.onboarding.mutations import (
    upsert_llm_provider,
    upsert_router,
    upsert_search_provider,
)

_HAND_TIERS = {
    "c0": {"provider": "openrouter", "model": "hand-c0"},
    "c1": {"provider": "openrouter", "model": "hand-c1"},
    "c2": {"provider": "openrouter", "model": "hand-c2"},
    "c3": {"provider": "openrouter", "model": "hand-c3"},
}


def _config_with_hand_tiers() -> GatewayConfig:
    return GatewayConfig(
        llm={"provider": "openrouter", "model": "hand-c1", "api_key": "sk-old"},
        squilla_router={
            "enabled": True,
            "tier_profile": None,
            "tiers": {name: dict(tier) for name, tier in _HAND_TIERS.items()},
        },
    )


def _configured_openrouter() -> GatewayConfig:
    res = upsert_llm_provider(
        GatewayConfig(),
        provider_id="openrouter",
        model="corp-model",
        api_key="sk-old",
        base_url="https://llm.corp.example/v1",
        proxy="http://127.0.0.1:7890",
        provider_routing={"corp-model": "corp-upstream"},
    )
    cfg = res.config
    cfg.llm.max_tokens = 4096
    cfg.llm.thinking = "high"
    return cfg


# --- B1-1: provider re-save keeps every field not explicitly passed ---------


def test_upsert_llm_provider_same_provider_resave_keeps_unspecified_fields():
    cfg = _configured_openrouter()

    res = upsert_llm_provider(cfg, provider_id="openrouter", api_key="sk-rotated")

    llm = res.config.llm
    assert llm.api_key == "sk-rotated"
    assert llm.model == "corp-model"
    assert llm.base_url == "https://llm.corp.example/v1"
    assert llm.proxy == "http://127.0.0.1:7890"
    assert llm.provider_routing == {"corp-model": "corp-upstream"}
    assert llm.max_tokens == 4096
    assert llm.thinking == "high"


def test_upsert_llm_provider_public_payload_reflects_kept_values():
    cfg = _configured_openrouter()

    res = upsert_llm_provider(cfg, provider_id="openrouter", api_key="sk-rotated")

    assert res.public_payload["model"] == "corp-model"
    assert res.public_payload["base_url"] == "https://llm.corp.example/v1"
    assert res.public_payload["proxy"] == "http://127.0.0.1:7890"
    assert res.public_payload["provider_routing"] == {"corp-model": "corp-upstream"}


def test_upsert_llm_provider_explicit_values_still_override_stored():
    cfg = _configured_openrouter()

    res = upsert_llm_provider(
        cfg,
        provider_id="openrouter",
        model="new-model",
        api_key="sk-rotated",
        base_url="https://other.example/v1",
        proxy="",
        provider_routing={},
    )

    llm = res.config.llm
    assert llm.model == "new-model"
    assert llm.base_url == "https://other.example/v1"
    assert llm.proxy == ""
    assert llm.provider_routing == {}


def test_upsert_llm_provider_empty_model_string_keeps_legacy_default_chain():
    # model="" (explicit empty, the legacy RPC spelling) still derives the
    # router-tier default rather than keeping the stored model.
    cfg = _configured_openrouter()

    res = upsert_llm_provider(cfg, provider_id="openrouter", model="", api_key="sk-r")

    assert res.config.llm.model != "corp-model"
    assert res.config.llm.model


def test_upsert_llm_provider_provider_switch_does_not_carry_unrelated_fields():
    cfg = _configured_openrouter()

    res = upsert_llm_provider(
        cfg, provider_id="deepseek", model="deepseek-chat", api_key="sk-ds"
    )

    llm = res.config.llm
    assert llm.provider == "deepseek"
    assert llm.proxy == ""
    assert llm.provider_routing == {}
    assert llm.base_url != "https://llm.corp.example/v1"
    assert llm.max_tokens == 0
    assert llm.thinking is None


def test_upsert_llm_provider_same_provider_resave_keeps_custom_router_tiers():
    cfg = _config_with_hand_tiers()

    res = upsert_llm_provider(cfg, provider_id="openrouter", api_key="sk-rotated")

    router = res.config.squilla_router
    assert router.tier_profile is None
    for tier in ("c0", "c1", "c2", "c3"):
        assert router.tiers[tier]["model"] == f"hand-{tier}"


def test_upsert_llm_provider_same_provider_resave_still_reconciles_untouched_tiers():
    # Guard for the skip rule: default (never hand-edited) tiers still get
    # the packaged compact profile on a same-provider save.
    res = upsert_llm_provider(
        GatewayConfig(), provider_id="openrouter", model="m", api_key="sk-x"
    )

    assert res.config.squilla_router.tier_profile == "openrouter"


def test_upsert_llm_provider_same_provider_model_change_reseeds_preset_seeded_tiers():
    # Tiers a previous save seeded from the provider preset are not a
    # hand-edited ladder: a model change must keep refreshing them.
    first = upsert_llm_provider(
        GatewayConfig(), provider_id="groq", model="model-a", api_key="sk-g"
    )
    second = upsert_llm_provider(
        first.config, provider_id="groq", model="model-b", api_key=""
    )

    router = second.config.squilla_router
    for tier in ("c0", "c1", "c2", "c3"):
        assert router.tiers[tier]["model"] == "model-b"


# --- B1-2: search re-save keeps omitted settings ------------------------------


def test_upsert_search_provider_omitted_fields_keep_current():
    first = upsert_search_provider(
        GatewayConfig(),
        provider_id="brave",
        api_key="brave-key",
        max_results=9,
        proxy="http://127.0.0.1:7890",
        use_env_proxy=True,
        fallback_policy="network",
        diagnostics=True,
    )

    second = upsert_search_provider(
        first.config, provider_id="brave", api_key="brave-rotated"
    )

    cfg = second.config
    assert cfg.search_api_key == "brave-rotated"
    assert cfg.search_max_results == 9
    assert cfg.search_proxy == "http://127.0.0.1:7890"
    assert cfg.search_use_env_proxy is True
    assert cfg.search_fallback_policy == "network"
    assert cfg.search_diagnostics is True


def test_upsert_search_provider_explicit_values_override_stored():
    first = upsert_search_provider(
        GatewayConfig(),
        provider_id="brave",
        api_key="brave-key",
        max_results=9,
        proxy="http://127.0.0.1:7890",
        use_env_proxy=True,
        fallback_policy="network",
        diagnostics=True,
    )

    second = upsert_search_provider(
        first.config,
        provider_id="brave",
        max_results=3,
        proxy="",
        use_env_proxy=False,
        fallback_policy="off",
        diagnostics=False,
    )

    cfg = second.config
    assert cfg.search_api_key == "brave-key"  # blank key keeps current
    assert cfg.search_max_results == 3
    assert cfg.search_proxy == ""
    assert cfg.search_use_env_proxy is False
    assert cfg.search_fallback_policy == "off"
    assert cfg.search_diagnostics is False


def test_upsert_search_provider_keep_current_applies_across_provider_switch():
    # max_results/proxy/... are global search settings, so they survive a
    # provider switch when omitted.
    first = upsert_search_provider(
        GatewayConfig(),
        provider_id="brave",
        api_key="brave-key",
        max_results=9,
        diagnostics=True,
    )

    second = upsert_search_provider(first.config, provider_id="duckduckgo")

    assert second.config.search_provider == "duckduckgo"
    assert second.config.search_max_results == 9
    assert second.config.search_diagnostics is True


# --- B1-3: disable preserves the ladder; custom re-enable restores it --------


def test_upsert_router_disable_preserves_inline_custom_tiers():
    cfg = _config_with_hand_tiers()

    res = upsert_router(cfg, mode="disabled")

    router = res.config.squilla_router
    assert router.enabled is False
    assert router.tier_profile is None
    for tier in ("c0", "c1", "c2", "c3"):
        assert router.tiers[tier]["model"] == f"hand-{tier}"


def test_upsert_router_disable_then_custom_reenable_restores_hand_built_tiers():
    cfg = _config_with_hand_tiers()

    disabled = upsert_router(cfg, mode="disabled")
    reenabled = upsert_router(disabled.config, mode="custom")

    router = reenabled.config.squilla_router
    assert router.enabled is True
    assert router.tier_profile is None
    for tier in ("c0", "c1", "c2", "c3"):
        assert router.tiers[tier]["model"] == f"hand-{tier}"


def test_upsert_router_custom_keeps_inline_hand_tiers_when_tiers_omitted():
    # Even without a disable/enable round trip, a custom-mode save that
    # passes no tiers must not reset an inline hand-built ladder.
    cfg = _config_with_hand_tiers()

    res = upsert_router(cfg, mode="custom")

    router = res.config.squilla_router
    for tier in ("c0", "c1", "c2", "c3"):
        assert router.tiers[tier]["model"] == f"hand-{tier}"


def test_upsert_router_recommended_reenable_still_resets_to_packaged_profile():
    # "recommended" remains an explicit reset to the packaged profile.
    cfg = _config_with_hand_tiers()

    disabled = upsert_router(cfg, mode="disabled")
    reenabled = upsert_router(disabled.config, mode="recommended")

    router = reenabled.config.squilla_router
    assert router.enabled is True
    assert router.tier_profile == "openrouter"
    assert router.tiers["c0"]["model"] != "hand-c0"


def test_upsert_router_disabled_ladder_survives_toml_round_trip():
    cfg = _config_with_hand_tiers()

    disabled = upsert_router(cfg, mode="disabled")
    persisted = disabled.config.to_toml_dict()["squilla_router"]
    reloaded = GatewayConfig(squilla_router=persisted, llm={"provider": "openrouter"})

    assert reloaded.squilla_router.enabled is False
    assert reloaded.squilla_router.tiers["c2"]["model"] == "hand-c2"