clanker-saddle/test/bundles/modelconfig.py

"""The model configuration bundle (Admin → Model page): the OpenAI
endpoint every saddle assignment is sent to lives in the model_config
row — environment variables no longer configure it (the harness's
SADDLE_LLM_BASE_URL seeds the row once, on the app's first read; the
page's save action owns it afterwards).
"""
from harness import *  # noqa: F401,F403 — the shared wire helpers


def run(ctx):
    auth, fake, fake_id, fake_token = ctx.auth, ctx.fake, ctx.fake_id, ctx.fake_token

    print("== model configuration: the Admin → Model page ==")
    fake.ask_on_register = False
    fake.hold_state = True

    # The bootstrap seed: the app's first configuration read imported
    # the harness's SADDLE_LLM_BASE_URL environment into the row.
    model = slot_public(auth, "/models", "model")
    check(model.get("dispatch_ready") is True,
          "the page reports dispatch ready (the env bootstrap seeded the row)")
    check(model.get("base_url", "").startswith("http://"),
          "the seeded base URL is visible on the page")

    # The save action rewrites the row: the next assignment carries the
    # DATABASE's endpoint, not the environment's.
    task_id = create_queued(auth, {
        "project_id": "1", "title": "Model config round trip",
        "priority": "2"})
    set_model_config(base_url="http://127.0.0.1:9", model="db-model",
                     api_key="db-key-12345678")
    ask_status, assignment = fake.ask_now()
    check(ask_status == 200 and int(assignment["task"]["id"]) == task_id,
          "the ask claims the task under the database configuration")
    llm = assignment.get("llm_config") or {}
    check(llm.get("base_url") == "http://127.0.0.1:9"
          and llm.get("model") == "db-model"
          and llm.get("api_key") == "db-key-12345678",
          "the assignment's llm_config carries the database endpoint")
    # close the round so the fake saddle is not left busy (a busy
    # saddle's ask 204s for its OWN open session, which would mask the
    # configuration gate the later checks isolate)
    fake.state = "done"
    fake.push_results("done", summary="# done",
                      session_id=assignment.get("session_id"))
    wait_for(lambda: task_state(auth, task_id).get("status") == "done"
             or False, timeout=30, what="model config task done")

    # The page's own state reflects the row (the key is never echoed —
    # only that one is set plus a masked hint).
    model = slot_public(auth, "/models", "model")
    check(model.get("base_url") == "http://127.0.0.1:9"
          and model.get("model") == "db-model",
          "the page shows the saved endpoint fields")
    check(model.get("api_key_set") is True
          and "db-key-12345678" not in json.dumps(model),
          "the page never echoes the stored key (set flag + masked hint only)")
    check(model.get("api_key_hint", "").endswith("5678")
          and len(model.get("api_key_hint", "")) <= 5,
          "the masked hint carries only the key's tail")

    # A save with a BLANK api key keeps the stored one (the page's
    # write-only contract); a new base URL and model still land.
    model = slot_public(auth, "/models", "model")
    act(auth, model["save"], {
        "base_url": "http://127.0.0.1:10", "model": "db-model-2",
        "api_key": "", "advisor_base_url": "", "advisor_model": "",
        "advisor_api_key": ""})
    conn = db()
    row = conn.execute(
        "select base_url, model, api_key from model_config where id = 1").fetchone()
    conn.close()
    check(row is not None and row[0] == "http://127.0.0.1:10" and row[1] == "db-model-2",
          "the save action persists the new endpoint fields")
    check(row is not None and row[2] == "db-key-12345678",
          "a blank key field KEEPS the stored key")

    # The advisor override resolves independently: an advisor base URL
    # set through the page overrides, empty falls back to the main
    # fields (observable through the advisor-configured flags).
    set_model_config(advisor_base_url="http://127.0.0.1:11", advisor_model="adv")
    model = slot_public(auth, "/models", "model")
    check(model.get("advisor_base_url") == "http://127.0.0.1:11"
          and model.get("advisor_model") == "adv",
          "the advisor override shows on the page")
    set_model_config(advisor_base_url="")
    model = slot_public(auth, "/models", "model")
    check(model.get("advisor_base_url") == "",
          "clearing the advisor override falls back to the main endpoint")

    # Validation: a non-http base URL is refused and keeps the row.
    model = slot_public(auth, "/models", "model")
    directives = act(auth, model["save"], {
        "base_url": "ftp://nope", "model": "", "api_key": "",
        "advisor_base_url": "", "advisor_model": "", "advisor_api_key": ""})
    notified = json.dumps(directives)
    check("http://" in notified or "error" in notified,
          "a non-http base URL is rejected with a notice")
    conn = db()
    row = conn.execute("select base_url from model_config where id = 1").fetchone()
    conn.close()
    check(row is not None and row[0] == "http://127.0.0.1:10",
          "the rejected save left the stored configuration untouched")

    # Emptying the base URL pauses dispatch: a fresh queued task finds
    # no taker while no model endpoint is configured.
    set_model_config(base_url="")
    model = slot_public(auth, "/models", "model")
    check(model.get("dispatch_ready") is False,
          "the page reports dispatch paused with no base URL")
    paused_task = create_queued(auth, {
        "project_id": "1", "title": "Paused until an endpoint exists",
        "priority": "1"})
    ask_status, assignment = fake.ask_now()
    check(ask_status == 204,
          "the ask returns 204 while no model endpoint is configured")
    # ...and saving one through the PAGE lifts the pause.
    model = slot_public(auth, "/models", "model")
    act(auth, model["save"], {
        "base_url": "http://127.0.0.1:12", "model": "db-model-3",
        "api_key": "", "advisor_base_url": "", "advisor_model": "",
        "advisor_api_key": ""})
    ask_status, assignment = fake.ask_now()
    check(ask_status == 200 and int(assignment["task"]["id"]) == paused_task,
          "saving an endpoint through the page lifts the dispatch pause")