retoorandClaude Sonnet 5
d9ff99c4a0
gateway_thinking_dialect overrides the URL-sniffed protocol dialect for a reverse-proxied upstream (e.g. Ollama) whose URL carries no identifying token; upstream_capabilities() uses the same effective dialect to stop sending stream_options to upstreams that don't support it. gateway_allow_client_model lets a client-requested model name through even with force-model on, for an upstream that serves many models with no single stable alias. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01XjW4qocnaJxhugUi5ca8Wo
137 lines
4.9 KiB
Python
137 lines
4.9 KiB
Python
# retoor <retoor@molodetz.nl>
|
|
|
|
from devplacepy.services.openai_gateway.thinking import (
|
|
apply_thinking,
|
|
client_thinking_enabled,
|
|
thinking_dialect,
|
|
upstream_capabilities,
|
|
)
|
|
|
|
|
|
def test_dialect_from_upstream_url():
|
|
assert thinking_dialect("https://api.deepseek.com/chat/completions") == "deepseek"
|
|
assert thinking_dialect("https://openrouter.ai/api/v1/chat/completions") == "openrouter"
|
|
assert thinking_dialect("http://127.0.0.1:11434/api/chat") == "ollama"
|
|
assert thinking_dialect("http://ollama.local/v1/chat/completions") == "ollama"
|
|
assert thinking_dialect("https://unknown.example/v1/chat/completions") == "deepseek"
|
|
|
|
|
|
def test_upstream_capabilities_openai_full_schema():
|
|
caps = upstream_capabilities("https://api.deepseek.com/chat/completions")
|
|
assert caps.dialect == "deepseek"
|
|
assert caps.supports_stream_options is True
|
|
assert caps.supports_stream_usage is True
|
|
|
|
|
|
def test_upstream_capabilities_ollama_lacks_stream_options():
|
|
caps = upstream_capabilities("http://127.0.0.1:11434/api/chat")
|
|
assert caps.dialect == "ollama"
|
|
assert caps.supports_stream_options is False
|
|
assert caps.supports_stream_usage is False
|
|
|
|
|
|
def test_upstream_capabilities_explicit_dialect_override():
|
|
caps = upstream_capabilities(
|
|
"https://ai.example.com/v1/chat/completions", dialect="ollama"
|
|
)
|
|
assert caps.dialect == "ollama"
|
|
assert caps.supports_stream_options is False
|
|
|
|
|
|
def test_apply_thinking_respects_explicit_dialect_override():
|
|
without = apply_thinking({}, "https://ai.example.com/v1/chat/completions")
|
|
assert without["thinking"] == {"type": "disabled"}
|
|
with_override = apply_thinking(
|
|
{}, "https://ai.example.com/v1/chat/completions", dialect="ollama"
|
|
)
|
|
assert with_override["think"] is False
|
|
assert "thinking" not in with_override
|
|
|
|
|
|
def test_client_intent_unspecified():
|
|
assert client_thinking_enabled({}) is None
|
|
assert client_thinking_enabled({"messages": []}) is None
|
|
assert client_thinking_enabled(None) is None
|
|
|
|
|
|
def test_client_intent_disable():
|
|
assert client_thinking_enabled({"think": False}) is False
|
|
assert client_thinking_enabled({"think": "false"}) is False
|
|
assert client_thinking_enabled({"thinking": {"type": "disabled"}}) is False
|
|
assert client_thinking_enabled({"reasoning": {"effort": "none"}}) is False
|
|
assert client_thinking_enabled({"reasoning": {"enabled": False}}) is False
|
|
assert client_thinking_enabled({"enable_thinking": False}) is False
|
|
assert client_thinking_enabled(
|
|
{"chat_template_kwargs": {"enable_thinking": False}}
|
|
) is False
|
|
|
|
|
|
def test_client_intent_enable():
|
|
assert client_thinking_enabled({"think": True}) is True
|
|
assert client_thinking_enabled({"think": "high"}) is True
|
|
assert client_thinking_enabled({"thinking": {"type": "enabled"}}) is True
|
|
assert client_thinking_enabled({"reasoning": {"effort": "low"}}) is True
|
|
assert client_thinking_enabled({"reasoning_effort": "max"}) is True
|
|
|
|
|
|
def test_default_disables_deepseek_thinking():
|
|
payload = apply_thinking(
|
|
{"messages": [{"role": "user", "content": "hi"}]},
|
|
"https://api.deepseek.com/chat/completions",
|
|
)
|
|
assert payload["thinking"] == {"type": "disabled"}
|
|
assert "think" not in payload
|
|
assert "reasoning" not in payload
|
|
assert "reasoning_effort" not in payload
|
|
|
|
|
|
def test_default_disables_openrouter_and_ollama():
|
|
openrouter = apply_thinking({}, "https://openrouter.ai/api/v1/chat/completions")
|
|
assert openrouter["reasoning"] == {"effort": "none"}
|
|
assert "thinking" not in openrouter
|
|
ollama = apply_thinking({}, "http://127.0.0.1:11434/api/chat")
|
|
assert ollama["think"] is False
|
|
assert "thinking" not in ollama
|
|
|
|
|
|
def test_client_can_enable_thinking():
|
|
payload = apply_thinking(
|
|
{"think": True, "messages": []},
|
|
"https://api.deepseek.com/chat/completions",
|
|
)
|
|
assert payload["thinking"] == {"type": "enabled"}
|
|
assert "think" not in payload
|
|
|
|
|
|
def test_client_disable_beats_admin_default_on():
|
|
payload = apply_thinking(
|
|
{"thinking": {"type": "disabled"}},
|
|
"https://api.deepseek.com/chat/completions",
|
|
default_enabled=True,
|
|
)
|
|
assert payload["thinking"] == {"type": "disabled"}
|
|
|
|
|
|
def test_admin_default_on_when_client_silent():
|
|
payload = apply_thinking(
|
|
{"messages": []},
|
|
"https://api.deepseek.com/chat/completions",
|
|
default_enabled=True,
|
|
)
|
|
assert payload["thinking"] == {"type": "enabled"}
|
|
|
|
|
|
def test_strips_conflicting_client_fields():
|
|
payload = apply_thinking(
|
|
{
|
|
"think": False,
|
|
"reasoning_effort": "high",
|
|
"chat_template_kwargs": {"enable_thinking": True, "other": 1},
|
|
},
|
|
"https://api.deepseek.com/chat/completions",
|
|
)
|
|
assert payload["thinking"] == {"type": "disabled"}
|
|
assert "think" not in payload
|
|
assert "reasoning_effort" not in payload
|
|
assert payload["chat_template_kwargs"] == {"other": 1}
|