92 lines
3.4 KiB
Python
92 lines
3.4 KiB
Python
"""Thinking switch: the LLM request body carries the thinking-off parameter only when off."""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import tempfile
|
|
import unittest
|
|
from unittest import mock
|
|
|
|
os.environ.setdefault("CONFIG_DIR", tempfile.mkdtemp(prefix="ttvc-think-test-config-"))
|
|
|
|
from fastapi import FastAPI
|
|
from fastapi.testclient import TestClient
|
|
|
|
import routes.conversation as conversation
|
|
|
|
|
|
class _Resp:
|
|
status_code = 200
|
|
|
|
def __init__(self, lines=(), body=None):
|
|
self._lines, self._body = lines, body or {}
|
|
|
|
def raise_for_status(self):
|
|
pass
|
|
|
|
def iter_lines(self):
|
|
return iter(self._lines)
|
|
|
|
def json(self):
|
|
return self._body
|
|
|
|
|
|
class ThinkingParams(unittest.TestCase):
|
|
def test_off_by_default_openai_compatible(self):
|
|
self.assertEqual(conversation._thinking_off_params({}, "http://h:17861/v1"),
|
|
{"chat_template_kwargs": {"enable_thinking": False}})
|
|
|
|
def test_off_ollama_adds_think_false(self):
|
|
p = conversation._thinking_off_params({"conversation_thinking": False}, "http://h:11434/v1")
|
|
self.assertIs(p["think"], False)
|
|
self.assertIn("chat_template_kwargs", p)
|
|
|
|
def test_on_sends_nothing(self):
|
|
for url in ("http://h:17861/v1", "http://h:11434/v1"):
|
|
self.assertEqual(conversation._thinking_off_params({"conversation_thinking": True}, url), {})
|
|
|
|
|
|
class TurnRequest(unittest.TestCase):
|
|
def _post_turn(self, setting, url="http://h:17861/v1"):
|
|
app = FastAPI()
|
|
app.include_router(conversation.router)
|
|
sent = []
|
|
|
|
def fake_post(u, json=None, **kw):
|
|
sent.append(json)
|
|
return _Resp(lines=[b"data: [DONE]"]) # empty reply -> error event, no TTS
|
|
|
|
settings = {"llm_url": url, "conversation_thinking": setting}
|
|
with mock.patch.object(conversation, "_load_settings", return_value=settings), \
|
|
mock.patch.object(conversation.requests, "post", side_effect=fake_post):
|
|
r = TestClient(app).post("/api/conversation/turn", data={"text": "Please tell me a short story about the sea"})
|
|
return sent, r.text
|
|
|
|
def test_stream_off_sends_param(self):
|
|
sent, body = self._post_turn(False)
|
|
self.assertEqual(sent[0]["chat_template_kwargs"], {"enable_thinking": False})
|
|
self.assertIn("turn Thinking off", body)
|
|
|
|
def test_stream_on_omits_param(self):
|
|
sent, _ = self._post_turn(True)
|
|
self.assertNotIn("chat_template_kwargs", sent[0])
|
|
self.assertNotIn("think", sent[0])
|
|
|
|
def test_stream_ollama_off(self):
|
|
sent, _ = self._post_turn(False, "http://h:11434/v1")
|
|
self.assertIs(sent[0]["think"], False)
|
|
|
|
def test_persona_rewrite_off_on_and_strips_think(self):
|
|
body = {"choices": [{"message": {"content": "<think>x</think>Aye"}}]}
|
|
for setting, expect in ((False, True), (True, False)):
|
|
sent = []
|
|
with mock.patch.object(conversation, "_load_settings", return_value={"conversation_thinking": setting}), \
|
|
mock.patch.object(conversation.requests, "post",
|
|
side_effect=lambda u, json=None, **k: (sent.append(json), _Resp(body=body))[1]):
|
|
out = conversation._rewrite_with_persona_sync("hi", "pirate", "http://h/v1")
|
|
self.assertEqual(out, "Aye")
|
|
self.assertEqual("chat_template_kwargs" in sent[0], expect)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|