stubborn-otter/otter-boards

Otter Boards, a team task board you can run from the web or from Claude and ChatGPT. FastAPI + DataK3 + appid + Models + MCP.

• boards
• tasks
• fastapi
• mcp
• dodil-challenge
• dodil-challenge-boards

A remix template · 1 remixes · Remix stubborn-otter/otter-boards

tests/test_models_client.py

"""The Dodil Models client against a stand-in transport shaped like the brief describes.
This proves our handling of those shapes. Only the real service can prove the shapes are right."""
import json

import httpx
import pytest

from boards.application.ports import AssistantUnavailable
from boards.infrastructure.dodil import ModelsAssistant


def assistant(handler):
    seen = []

    def record(request):
        seen.append(request)
        return handler(request)

    client = httpx.Client(transport=httpx.MockTransport(record))
    return ModelsAssistant(get_token=lambda: "tok", base="https://models.test/v1", http=client), seen


def chat_reply(text):
    return {"choices": [{"message": {"content": text}}]}


def wrapped(body):
    return httpx.Response(200, json={"status": "ok", "data": body})


def test_chat_unwraps_the_gateway_envelope_and_sends_the_token():
    ai, seen = assistant(lambda r: wrapped(chat_reply("Hello")))
    assert ai.chat("hi") == "Hello"
    assert seen[0].headers["authorization"] == "Bearer tok"
    assert json.loads(seen[0].content)["model"] == "kimi-k2.6"


def test_chat_also_reads_a_bare_openai_body():
    ai, _ = assistant(lambda r: httpx.Response(200, json=chat_reply("Hello")))
    assert ai.chat("hi") == "Hello"


def test_checklist_cleans_up_how_models_write_lists():
    text = "- One\n* Two\n3. Three\n4) Four\n[ ] Five\n\n• Six\n"
    ai, _ = assistant(lambda r: wrapped(chat_reply(text)))
    assert ai.checklist("t", "d") == ["One", "Two", "Three", "Four", "Five", "Six"]


def test_embeddings_inside_the_envelope_keep_their_order():
    body = {"object": "list", "data": [{"index": 1, "embedding": [0.0, 1.0]},
                                       {"index": 0, "embedding": [1.0, 0.0]}]}
    ai, seen = assistant(lambda r: wrapped(body))
    assert ai.embed(["a", "b"]) == [[1.0, 0.0], [0.0, 1.0]]
    assert json.loads(seen[0].content)["model"] == "jina-embeddings-v4"


def test_embeddings_in_a_bare_openai_body():
    body = {"object": "list", "data": [{"index": 0, "embedding": [1.0]}]}
    ai, _ = assistant(lambda r: httpx.Response(200, json=body))
    assert ai.embed(["a"]) == [[1.0]]


def test_many_texts_go_in_batches():
    def reply(request):
        count = len(json.loads(request.content)["input"])
        return wrapped({"data": [{"index": i, "embedding": [i]} for i in range(count)]})

    ai, seen = assistant(reply)
    assert len(ai.embed(["t"] * 70)) == 70
    assert len(seen) == 3


@pytest.mark.parametrize("response", [
    httpx.Response(500, text="boom"), httpx.Response(401, json={}),
    httpx.Response(200, text="<html>"), httpx.Response(200, json={"status": "ok"}),
    httpx.Response(200, json=["odd"]),
])
def test_every_failure_becomes_unavailable(response):
    ai, _ = assistant(lambda r: response)
    with pytest.raises(AssistantUnavailable):
        ai.chat("hi")
    with pytest.raises(AssistantUnavailable):
        ai.embed(["hi"])


def test_a_network_failure_or_missing_credentials_becomes_unavailable():
    def down(request):
        raise httpx.ConnectError("no route")

    ai, _ = assistant(down)
    with pytest.raises(AssistantUnavailable):
        ai.summary("facts")

    def no_secret():
        raise KeyError("DODIL_SERVICE_ACCOUNT_ID")

    ai = ModelsAssistant(get_token=no_secret, base="https://models.test/v1")
    with pytest.raises(AssistantUnavailable):
        ai.chat("hi")