249 lines
12 KiB
Python
249 lines
12 KiB
Python
"""Provider protocol boundaries without real credentials or network calls."""
|
|
|
|
from copy import deepcopy
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import Mock, patch
|
|
|
|
from logistics_chat import config, create_app
|
|
from logistics_chat.model_provider import completion_body, prepare_history
|
|
|
|
|
|
class ModelProviderTests(unittest.TestCase):
|
|
def setUp(self):
|
|
directory = tempfile.TemporaryDirectory()
|
|
self.addCleanup(directory.cleanup)
|
|
self.root = Path(directory.name)
|
|
root_patch = patch.object(config, 'ROOT_DIR', self.root)
|
|
root_patch.start()
|
|
self.addCleanup(root_patch.stop)
|
|
environment = patch.dict(os.environ, {}, clear=True)
|
|
environment.start()
|
|
self.addCleanup(environment.stop)
|
|
network = patch('requests.sessions.Session.request',
|
|
side_effect=AssertionError('Provider tests must not access the network'))
|
|
network.start()
|
|
self.addCleanup(network.stop)
|
|
self.app = create_app({
|
|
'TESTING': True,
|
|
'INITIALIZE_RUNTIME': False,
|
|
'DATABASE': self.root / 'history.db',
|
|
'UPLOAD_DIR': self.root / 'uploads',
|
|
'BACKUP_DIR': self.root / 'backups',
|
|
'LOG_DIR': self.root / 'logs',
|
|
'DEEPSEEK_API_KEY': '',
|
|
'CHAT_API_KEY': '',
|
|
'MODEL_NAME': 'synthetic-model',
|
|
'MODEL_URL': 'https://api.deepseek.com/v1/chat/completions',
|
|
'MODEL_PROVIDER': 'auto',
|
|
'MODEL_THINKING': 'disabled',
|
|
'MODEL_TOOL_CHOICE': 'required',
|
|
})
|
|
self.tools = [{
|
|
'type': 'function',
|
|
'function': {
|
|
'name': 'read_service_info',
|
|
'description': 'Read synthetic service information.',
|
|
'parameters': {'type': 'object', 'properties': {}},
|
|
},
|
|
}]
|
|
self.messages = [{'role': 'user', 'content': '请查询收货要求'}]
|
|
|
|
def body(self, *, first=True, **settings):
|
|
with self.app.app_context(), patch.dict(self.app.config, settings):
|
|
return completion_body(self.messages, self.tools, first=first)
|
|
|
|
def test_auto_detects_the_official_hostname_for_versioned_and_beta_paths(self):
|
|
urls = (
|
|
'https://api.deepseek.com/chat/completions',
|
|
'https://api.deepseek.com/v1/chat/completions',
|
|
'https://api.deepseek.com/beta/chat/completions',
|
|
'https://API.DEEPSEEK.COM:443/v1/chat/completions',
|
|
)
|
|
for url in urls:
|
|
with self.subTest(url=url):
|
|
body = self.body(MODEL_URL=url)
|
|
self.assertEqual(body['thinking'], {'type': 'disabled'})
|
|
self.assertEqual(body['tool_choice'], 'required')
|
|
|
|
def test_auto_does_not_treat_a_name_in_a_path_or_other_hostname_as_official(self):
|
|
urls = (
|
|
'https://proxy.example.invalid/api.deepseek.com/chat/completions',
|
|
'https://api.deepseek.com.example.invalid/chat/completions',
|
|
'https://api.deepseek.com@proxy.example.invalid/chat/completions',
|
|
)
|
|
for url in urls:
|
|
with self.subTest(url=url):
|
|
body = self.body(MODEL_URL=url, MODEL_THINKING='enabled')
|
|
self.assertNotIn('thinking', body)
|
|
self.assertEqual(body['temperature'], 0)
|
|
self.assertEqual(body['tool_choice'], 'required')
|
|
|
|
def test_explicit_compatible_provider_does_not_send_deepseek_thinking(self):
|
|
for thinking in ('enabled', 'disabled'):
|
|
with self.subTest(thinking=thinking):
|
|
body = self.body(MODEL_PROVIDER='compatible', MODEL_THINKING=thinking)
|
|
self.assertNotIn('thinking', body)
|
|
self.assertEqual(body['temperature'], 0)
|
|
self.assertEqual(body['tool_choice'], 'required')
|
|
|
|
def test_explicit_deepseek_provider_can_configure_a_separate_endpoint(self):
|
|
body = self.body(MODEL_PROVIDER='deepseek',
|
|
MODEL_URL='https://proxy.example.invalid/chat/completions',
|
|
MODEL_THINKING='enabled')
|
|
self.assertEqual(body['thinking'], {'type': 'enabled'})
|
|
self.assertEqual(body['tool_choice'], 'auto')
|
|
self.assertNotIn('temperature', body)
|
|
|
|
def test_thinking_changes_required_to_auto_on_every_tool_round(self):
|
|
for first in (True, False):
|
|
with self.subTest(first=first):
|
|
body = self.body(first=first, MODEL_THINKING='enabled')
|
|
self.assertEqual(body['thinking'], {'type': 'enabled'})
|
|
self.assertEqual(body['tool_choice'], 'auto')
|
|
self.assertNotIn('temperature', body)
|
|
self.assertEqual(body['tools'], self.tools)
|
|
|
|
def test_disabled_thinking_requires_a_tool_only_on_the_first_round(self):
|
|
first = self.body(first=True)
|
|
following = self.body(first=False)
|
|
self.assertEqual(first['tool_choice'], 'required')
|
|
self.assertEqual(following['tool_choice'], 'auto')
|
|
for body in (first, following):
|
|
self.assertEqual(body['thinking'], {'type': 'disabled'})
|
|
self.assertEqual(body['temperature'], 0)
|
|
|
|
def test_auto_choice_remains_available_for_the_first_round(self):
|
|
for provider in ('deepseek', 'compatible'):
|
|
with self.subTest(provider=provider):
|
|
body = self.body(MODEL_PROVIDER=provider, MODEL_TOOL_CHOICE='auto')
|
|
self.assertEqual(body['tool_choice'], 'auto')
|
|
|
|
def test_rejects_unknown_provider_instead_of_silently_guessing_compatibility(self):
|
|
for provider in ('unknown', '', None):
|
|
with self.subTest(provider=provider):
|
|
with self.assertRaisesRegex(ValueError, 'Unsupported model provider'):
|
|
self.body(MODEL_PROVIDER=provider)
|
|
|
|
def test_rejects_unsupported_first_round_tool_choice(self):
|
|
for choice in ('unknown', 'none', '', {'type': 'function'}):
|
|
with self.subTest(choice=choice):
|
|
with self.assertRaisesRegex(ValueError, 'Invalid tool choice'):
|
|
self.body(MODEL_TOOL_CHOICE=choice)
|
|
|
|
def test_rejects_invalid_thinking_configuration(self):
|
|
for thinking in ('auto', '', True):
|
|
with self.subTest(thinking=thinking):
|
|
with self.assertRaisesRegex(ValueError, 'Invalid thinking configuration'):
|
|
self.body(MODEL_THINKING=thinking)
|
|
|
|
def test_preserves_tool_history_and_reasoning_for_the_next_request(self):
|
|
history = self.messages + [{
|
|
'role': 'assistant',
|
|
'content': None,
|
|
'reasoning_content': 'Synthetic protocol state.',
|
|
'tool_calls': [{
|
|
'id': 'call_synthetic',
|
|
'type': 'function',
|
|
'function': {'name': 'read_service_info', 'arguments': '{}'},
|
|
}],
|
|
}, {
|
|
'role': 'tool',
|
|
'tool_call_id': 'call_synthetic',
|
|
'content': '{"status":"ok"}',
|
|
}]
|
|
with self.app.app_context(), patch.dict(self.app.config, {'MODEL_THINKING': 'enabled'}):
|
|
body = completion_body(iter(history), self.tools)
|
|
self.assertEqual(body['messages'], history)
|
|
self.assertEqual(body['model'], 'synthetic-model')
|
|
self.assertFalse(body['stream'])
|
|
|
|
def test_thinking_serializes_visible_history_without_inventing_assistant_trace(self):
|
|
history = [
|
|
{'role': 'user', 'content': '美国一吨货多少钱'},
|
|
{'role': 'assistant', 'content': '此前展示的报价，仅用于理解条件。',
|
|
'reasoning_content': 'This private field must not enter the background.',
|
|
'tool_calls': [{'id': 'old_call'}]},
|
|
{'role': 'user', 'content': '改从义乌交仓'},
|
|
]
|
|
original = deepcopy(history)
|
|
with self.app.app_context(), patch.dict(self.app.config, {'MODEL_THINKING': 'enabled'}):
|
|
prepared = prepare_history(history)
|
|
self.assertEqual([item['role'] for item in prepared], ['user', 'user'])
|
|
self.assertEqual(prepared[-1], original[-1])
|
|
background = json.loads(prepared[0]['content'].split('\n', 1)[1])
|
|
self.assertEqual(background, [{'role': item['role'], 'content': item['content']}
|
|
for item in original[:-1]])
|
|
self.assertNotIn('reasoning_content', prepared[0]['content'])
|
|
self.assertNotIn('old_call', prepared[0]['content'])
|
|
self.assertEqual(history, original)
|
|
|
|
def test_disabled_or_compatible_provider_keeps_the_visible_history_format(self):
|
|
history = self.messages + [{'role': 'assistant', 'content': '已确认公开条件。'},
|
|
{'role': 'user', 'content': '继续查询'}]
|
|
cases = (
|
|
{'MODEL_PROVIDER': 'auto', 'MODEL_THINKING': 'disabled'},
|
|
{'MODEL_PROVIDER': 'compatible', 'MODEL_THINKING': 'enabled'},
|
|
{'MODEL_PROVIDER': 'auto', 'MODEL_THINKING': 'enabled',
|
|
'MODEL_URL': 'https://proxy.example.invalid/chat/completions'},
|
|
)
|
|
for settings in cases:
|
|
with self.subTest(settings=settings), self.app.app_context(), patch.dict(self.app.config, settings):
|
|
self.assertEqual(prepare_history(history), history)
|
|
|
|
def test_explicit_deepseek_thinking_also_prepares_history_for_a_proxy(self):
|
|
history = self.messages + [{'role': 'assistant', 'content': '此前公开回复。'},
|
|
{'role': 'user', 'content': '继续查询'}]
|
|
settings = {'MODEL_PROVIDER': 'deepseek', 'MODEL_THINKING': 'enabled',
|
|
'MODEL_URL': 'https://proxy.example.invalid/chat/completions'}
|
|
with self.app.app_context(), patch.dict(self.app.config, settings):
|
|
prepared = prepare_history(history)
|
|
self.assertEqual([item['role'] for item in prepared], ['user', 'user'])
|
|
self.assertEqual(prepared[-1], history[-1])
|
|
|
|
def test_thinking_with_no_prior_messages_needs_no_background(self):
|
|
for history in ([], self.messages):
|
|
with self.subTest(history=history), self.app.app_context(), patch.dict(self.app.config, {'MODEL_THINKING': 'enabled'}):
|
|
self.assertEqual(prepare_history(history), history)
|
|
|
|
def test_mock_request_starts_fresh_but_preserves_the_current_tool_trace(self):
|
|
from logistics_chat.model_chat_service import completion
|
|
|
|
history = self.messages + [{'role': 'assistant', 'content': '此前公开回复。'},
|
|
{'role': 'user', 'content': '再查询标签要求'}]
|
|
trace = {'role': 'assistant', 'content': None,
|
|
'reasoning_content': 'Synthetic current-turn reasoning.',
|
|
'tool_calls': [{'id': 'current_call', 'type': 'function',
|
|
'function': {'name': 'read_service_info', 'arguments': '{}'}}]}
|
|
response = Mock(status_code=200)
|
|
response.json.return_value = {'choices': [{'finish_reason': 'tool_calls', 'message': trace}]}
|
|
with self.app.app_context(), patch.dict(self.app.config, {'MODEL_THINKING': 'enabled'}), \
|
|
patch('logistics_chat.ai_client.post_completion', return_value=response) as post:
|
|
messages = prepare_history(history)
|
|
returned = completion(messages, tools=True, first=True)
|
|
first_body = post.call_args.args[0]
|
|
self.assertFalse(any(item['role'] == 'assistant' for item in first_body['messages']))
|
|
self.assertEqual(first_body['messages'][-1], history[-1])
|
|
messages.extend([returned, {'role': 'tool', 'tool_call_id': 'current_call',
|
|
'content': '{"status":"ok"}'}])
|
|
completion(messages, tools=True)
|
|
next_body = post.call_args.args[0]
|
|
self.assertEqual(next_body['messages'][-2], trace)
|
|
self.assertEqual(next_body['messages'][-1]['tool_call_id'], 'current_call')
|
|
self.assertEqual(post.call_count, 2)
|
|
self.assertEqual(response.close.call_count, 2)
|
|
|
|
def test_no_tools_omits_tool_choice_and_creates_no_runtime_files(self):
|
|
with self.app.app_context():
|
|
body = completion_body(self.messages, [])
|
|
self.assertNotIn('tools', body)
|
|
self.assertNotIn('tool_choice', body)
|
|
self.assertEqual(list(self.root.iterdir()), [])
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|