188 lines
10 KiB
Python
188 lines
10 KiB
Python
"""HTTP replay keeps relative edits idempotent and identity scoped."""
|
|
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
import test_model_tool_loop as loop_fixtures
|
|
from test_model_tool_loop import tool_plan, quote_arguments, final_facts
|
|
from test_model_chat import plan
|
|
from test_quotes import frames
|
|
|
|
|
|
class ChatRequestTests(unittest.TestCase):
|
|
setUp = loop_fixtures.ModelToolLoopTests.setUp
|
|
saved = loop_fixtures.ModelToolLoopTests.saved
|
|
login = loop_fixtures.ModelToolLoopTests.login
|
|
|
|
def send(self, question, arguments, request_id):
|
|
with patch('logistics_chat.ai_client.post_completion', side_effect=[
|
|
tool_plan('quote_shipping', quote_arguments(**arguments), 'quote_a'), final_facts('fact_0')]):
|
|
return frames(self.client.post('/stream_chat', json={
|
|
'msg': question, 'conv_id': self.cid, 'request_id': request_id}))
|
|
|
|
def start_quote(self):
|
|
return self.send('美国1000KG义乌美东普船海派报价', {'services': ['普船海派'], 'shipment': {
|
|
'country': 'US', 'total_weight_kg': '1000', 'zone': '美东', 'origin': '义乌', 'kinds': ['parcel']}}, 'initial-request-0001')
|
|
|
|
def add_weight(self, request_id='weight-request-0001'):
|
|
return self.send('再加100KG', {'changes': [{'field': 'total_weight_kg', 'operation': 'add', 'value': '100'}]}, request_id)
|
|
|
|
def test_http_replay_is_read_only_and_an_identical_new_operation_still_adds(self):
|
|
self.start_quote()
|
|
original = self.add_weight()
|
|
before = self.saved()
|
|
with patch('logistics_chat.ai_client.post_completion') as model:
|
|
repeated = frames(self.client.post('/stream_chat', json={
|
|
'msg': '再加100KG', 'conv_id': self.cid, 'request_id': 'weight-request-0001'}))
|
|
model.assert_not_called()
|
|
self.assertEqual(repeated, original)
|
|
self.assertEqual(self.saved(), before)
|
|
self.assertIn('1100KG', repeated[0]['t'])
|
|
self.assertIn('1200KG', self.add_weight('weight-request-0002')[0]['t'])
|
|
with patch('logistics_chat.ai_client.post_completion') as model:
|
|
old_reply = frames(self.client.post('/stream_chat', json={
|
|
'msg': '再加100KG', 'conv_id': self.cid, 'request_id': 'weight-request-0001'}))
|
|
self.assertIn('1100KG', old_reply[0]['t'])
|
|
self.assertEqual(self.saved()[1]['intent']['weight'], '1200')
|
|
model.assert_not_called()
|
|
|
|
def test_duplicate_commits_return_the_original_result_before_state_cas(self):
|
|
self.start_quote()
|
|
original = self.add_weight()
|
|
before = self.saved()
|
|
with patch('logistics_chat.chat_requests.replay_result', return_value=None):
|
|
raced = self.add_weight()
|
|
self.assertEqual(raced, original)
|
|
self.assertEqual(self.saved(), before)
|
|
|
|
def test_request_id_cannot_be_reused_for_different_text(self):
|
|
self.start_quote()
|
|
with patch('logistics_chat.ai_client.post_completion') as model:
|
|
events = frames(self.client.post('/stream_chat', json={
|
|
'msg': '改为2000KG', 'conv_id': self.cid, 'request_id': 'initial-request-0001'}))
|
|
self.assertIn('e', events[0])
|
|
model.assert_not_called()
|
|
|
|
def test_invalid_ids_are_rejected_before_calling_a_model(self):
|
|
for value in ('short', 'x' * 81, ['request'], True):
|
|
with self.subTest(value=value), patch('logistics_chat.ai_client.post_completion') as model:
|
|
events = frames(self.client.post('/stream_chat', json={'msg': '报价', 'conv_id': self.cid, 'request_id': value}))
|
|
self.assertIn('e', events[0])
|
|
model.assert_not_called()
|
|
|
|
def test_deleting_a_conversation_removes_its_cached_results(self):
|
|
from logistics_chat import db
|
|
self.start_quote()
|
|
self.assertTrue(self.client.delete('/api/del_history', query_string={'id': self.cid}).json['ok'])
|
|
with self.app.app_context():
|
|
connection = db.connect()
|
|
self.assertEqual(connection.execute('SELECT COUNT(*) FROM chat_request_results WHERE conv_id=?', (self.cid,)).fetchone()[0], 0)
|
|
connection.close()
|
|
with patch('logistics_chat.ai_client.post_completion') as model:
|
|
events = frames(self.client.post('/stream_chat', json={
|
|
'msg': '美国1000KG义乌美东普船海派报价', 'conv_id': self.cid, 'request_id': 'initial-request-0001'}))
|
|
self.assertIn('e', events[0])
|
|
model.assert_not_called()
|
|
|
|
def test_guest_login_refreshes_orders_without_applying_the_quote_edit_again(self):
|
|
self.start_quote()
|
|
quote = tool_plan('quote_shipping', quote_arguments(changes=[
|
|
{'field': 'total_weight_kg', 'operation': 'add', 'value': '100'}]), 'quote_a').message['tool_calls'][0]
|
|
order = plan(('query_orders', '我的订单有多少个')).message['tool_calls'][0]
|
|
order['id'] = 'order_b'
|
|
from test_model_chat import Completion
|
|
question = '再加100KG，顺便查我的订单有多少个'
|
|
request_id = 'login-compound-0001'
|
|
def send():
|
|
with patch('logistics_chat.ai_client.post_completion', side_effect=[
|
|
Completion({'role': 'assistant', 'content': '', 'tool_calls': [quote, order]}, 'tool_calls'),
|
|
final_facts('fact_0', 'fact_1')]):
|
|
return frames(self.client.post('/stream_chat', json={
|
|
'msg': question, 'conv_id': self.cid, 'request_id': request_id}))
|
|
guest = send()
|
|
self.assertTrue(guest[0]['login_required'])
|
|
self.assertEqual(self.saved()[1]['intent']['weight'], '1100')
|
|
self.login()
|
|
authenticated = send()
|
|
self.assertNotIn('login_required', authenticated[0])
|
|
self.assertIn('1100KG', authenticated[0]['t'])
|
|
self.assertIn('1 票', authenticated[0]['t'])
|
|
self.assertEqual(self.saved()[1]['intent']['weight'], '1100')
|
|
self.client.post('/api/customer/logout', json={})
|
|
with patch('logistics_chat.ai_client.post_completion') as model:
|
|
denied = frames(self.client.post('/stream_chat', json={
|
|
'msg': question, 'conv_id': self.cid, 'request_id': request_id}))
|
|
self.assertTrue(denied[0]['login_required'])
|
|
model.assert_not_called()
|
|
|
|
def test_quote_replay_rejects_an_outdated_price_version(self):
|
|
from logistics_chat import db
|
|
from logistics_chat.quote_data import import_snapshot
|
|
self.start_quote()
|
|
newer = dict(self.snapshot, source_sha256='9' * 64)
|
|
with self.app.app_context():
|
|
connection = db.connect()
|
|
import_snapshot(connection, newer, activate=True)
|
|
connection.commit()
|
|
connection.close()
|
|
with patch('logistics_chat.ai_client.post_completion') as model:
|
|
events = frames(self.client.post('/stream_chat', json={
|
|
'msg': '美国1000KG义乌美东普船海派报价', 'conv_id': self.cid, 'request_id': 'initial-request-0001'}))
|
|
self.assertIn('报价资料已更新', events[0]['e'])
|
|
model.assert_not_called()
|
|
|
|
def test_login_model_timeout_reuses_confirmed_quote_in_local_fallback(self):
|
|
from logistics_chat import ai_client, db
|
|
from logistics_chat.quote_data import import_snapshot
|
|
from test_model_chat import Completion
|
|
self.start_quote()
|
|
question = '再加100KG报价，顺便查我的订单有多少个'
|
|
request_id = 'login-fallback-0001'
|
|
quote = tool_plan('quote_shipping', quote_arguments(changes=[
|
|
{'field': 'total_weight_kg', 'operation': 'add', 'value': '100'}]), 'quote_a').message['tool_calls'][0]
|
|
order = plan(('query_orders', '我的订单有多少个')).message['tool_calls'][0]
|
|
order['id'] = 'order_b'
|
|
with patch('logistics_chat.ai_client.post_completion', side_effect=[
|
|
Completion({'role': 'assistant', 'content': '', 'tool_calls': [quote, order]}, 'tool_calls'),
|
|
final_facts('fact_0', 'fact_1')]):
|
|
guest = frames(self.client.post('/stream_chat', json={
|
|
'msg': question, 'conv_id': self.cid, 'request_id': request_id}))
|
|
self.assertTrue(guest[0]['login_required'])
|
|
self.login()
|
|
with patch('logistics_chat.ai_client.post_completion', side_effect=ai_client.Timeout('test timeout')):
|
|
result = frames(self.client.post('/stream_chat', json={
|
|
'msg': question, 'conv_id': self.cid, 'request_id': request_id}))
|
|
self.assertIn('1100KG', result[0]['t'])
|
|
self.assertIn('1 票', result[0]['t'])
|
|
self.assertEqual(self.saved()[1]['intent']['weight'], '1100')
|
|
with self.app.app_context():
|
|
connection = db.connect()
|
|
import_snapshot(connection, dict(self.snapshot, source_sha256='8' * 64), activate=True)
|
|
connection.commit()
|
|
connection.close()
|
|
with patch('logistics_chat.ai_client.post_completion') as model:
|
|
expired = frames(self.client.post('/stream_chat', json={
|
|
'msg': question, 'conv_id': self.cid, 'request_id': request_id}))
|
|
self.assertIn('报价资料已更新', expired[0]['e'])
|
|
model.assert_not_called()
|
|
|
|
def test_legacy_rule_path_uses_the_same_replay_contract(self):
|
|
self.app.config['CHAT_USE_MODEL_TOOLS'] = False
|
|
self.start_quote()
|
|
original = self.add_weight()
|
|
before = self.saved()
|
|
repeated = self.add_weight()
|
|
self.assertEqual(repeated, original)
|
|
self.assertEqual(self.saved(), before)
|
|
|
|
def test_failed_state_commit_does_not_cache_a_partial_result(self):
|
|
from logistics_chat import db
|
|
with patch('logistics_chat.conversation_state.commit_state', return_value=False):
|
|
failed = self.start_quote()
|
|
self.assertIn('e', failed[0])
|
|
with self.app.app_context():
|
|
connection = db.connect()
|
|
self.assertEqual(connection.execute('SELECT COUNT(*) FROM chat_request_results').fetchone()[0], 0)
|
|
connection.close()
|
|
self.assertIn('1000KG', self.start_quote()[0]['t'])
|