Phillip Tarrant
f7c51b79da
- call_log._default_write: swallow any exception from the default file/ stdout sink and report to stderr, so a log-write failure (disk full, bad permissions, misconfigured CALL_LOG_PATH) never turns a successful narration into a 500 (charter §13). Injected write= sinks (used by tests) are left to surface their own errors. - ollama_client.chat: catch ValueError alongside httpx.HTTPError in the retry loop so a 200 response with a non-JSON body (JSONDecodeError is a ValueError subclass) counts as a failed attempt and falls through to ModelError after the one retry, instead of escaping chat() uncaught (charter §12 — one retry, then ModelError, nothing else escapes). - main.py: update stale docstrings — /dm/narrate is now fully wired (routing + model call + logging); the other four roles remain stubs. Regression tests added for both fixes (TDD: watched RED, then GREEN). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
96 lines
2.9 KiB
Python
96 lines
2.9 KiB
Python
import json
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from app.ollama_client import ModelError, chat
|
|
|
|
|
|
def _client(handler):
|
|
return httpx.Client(transport=httpx.MockTransport(handler), base_url="http://ollama")
|
|
|
|
|
|
def test_success_returns_content_and_sends_payload():
|
|
seen = {}
|
|
|
|
def handler(request):
|
|
seen["body"] = json.loads(request.content)
|
|
return httpx.Response(200, json={"message": {"content": "You stand on the wharf."}})
|
|
|
|
out = chat("qwen3.5:latest", [{"role": "user", "content": "hi"}], {"seed": 7},
|
|
think=False, client=_client(handler))
|
|
assert out == "You stand on the wharf."
|
|
assert seen["body"]["model"] == "qwen3.5:latest"
|
|
assert seen["body"]["stream"] is False
|
|
assert seen["body"]["options"]["seed"] == 7
|
|
assert seen["body"]["think"] is False
|
|
|
|
|
|
def test_think_omitted_when_none():
|
|
seen = {}
|
|
|
|
def handler(request):
|
|
seen["body"] = json.loads(request.content)
|
|
return httpx.Response(200, json={"message": {"content": "ok"}})
|
|
|
|
chat("m", [{"role": "user", "content": "x"}], {}, client=_client(handler))
|
|
assert "think" not in seen["body"]
|
|
|
|
|
|
def test_retries_once_then_succeeds():
|
|
calls = {"n": 0}
|
|
|
|
def handler(request):
|
|
calls["n"] += 1
|
|
if calls["n"] == 1:
|
|
raise httpx.ConnectError("boom")
|
|
return httpx.Response(200, json={"message": {"content": "recovered"}})
|
|
|
|
out = chat("m", [{"role": "user", "content": "x"}], {}, client=_client(handler))
|
|
assert out == "recovered"
|
|
assert calls["n"] == 2
|
|
|
|
|
|
def test_two_transport_failures_raise_modelerror():
|
|
def handler(request):
|
|
raise httpx.ConnectError("down")
|
|
|
|
with pytest.raises(ModelError):
|
|
chat("m", [{"role": "user", "content": "x"}], {}, client=_client(handler))
|
|
|
|
|
|
def test_timeout_exhausted_raises_modelerror():
|
|
def handler(request):
|
|
raise httpx.TimeoutException("slow")
|
|
|
|
with pytest.raises(ModelError):
|
|
chat("m", [{"role": "user", "content": "x"}], {}, client=_client(handler))
|
|
|
|
|
|
def test_non_2xx_retried_then_modelerror():
|
|
def handler(request):
|
|
return httpx.Response(500, json={"error": "nope"})
|
|
|
|
with pytest.raises(ModelError):
|
|
chat("m", [{"role": "user", "content": "x"}], {}, client=_client(handler))
|
|
|
|
|
|
def test_empty_content_retried_then_modelerror():
|
|
def handler(request):
|
|
return httpx.Response(200, json={"message": {"content": " "}})
|
|
|
|
with pytest.raises(ModelError):
|
|
chat("m", [{"role": "user", "content": "x"}], {}, client=_client(handler))
|
|
|
|
|
|
def test_non_json_200_retried_then_modelerror():
|
|
"""A 200 response with a non-JSON body (e.g. a proxy error page) must
|
|
degrade to ModelError after the one retry, not escape as a bare
|
|
JSONDecodeError/ValueError (charter §12/§13).
|
|
"""
|
|
def handler(request):
|
|
return httpx.Response(200, text="<html>not json</html>")
|
|
|
|
with pytest.raises(ModelError):
|
|
chat("m", [{"role": "user", "content": "x"}], {}, client=_client(handler))
|