178 lines
8.6 KiB
Python
178 lines
8.6 KiB
Python
"""Order references in a tool turn resolve against its pending list, never an older page."""
|
|
|
|
import json
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
import test_model_chat as fixtures
|
|
from test_model_tool_loop import evidence, final_facts, tool_plan
|
|
from test_quotes import frames
|
|
from logistics_chat import db, order_data
|
|
|
|
|
|
def order_arguments(question, action='followup', condition='all'):
|
|
return {'question': question, 'action': action, 'condition': condition,
|
|
'stage': '', 'status': '', 'reference': ''}
|
|
|
|
|
|
class ModelOrderLoopTests(unittest.TestCase):
|
|
def setUp(self):
|
|
fixtures.ModelChatTests.setUp(self)
|
|
fixtures.ModelChatTests.login(self)
|
|
|
|
def ask(self, question):
|
|
events = frames(self.client.post('/stream_chat', json={'msg': question, 'conv_id': self.cid}))
|
|
self.assertTrue(events[-1].get('d'), events)
|
|
self.assertNotIn('SECRET', json.dumps(events, ensure_ascii=False))
|
|
return events
|
|
|
|
def saved(self):
|
|
with self.app.app_context():
|
|
connection = db.connect()
|
|
try:
|
|
rows = connection.execute('SELECT role,content FROM messages WHERE conv_id=? ORDER BY id',
|
|
(self.cid,)).fetchall()
|
|
state = connection.execute('SELECT order_list_state FROM chat_history WHERE id=?',
|
|
(self.cid,)).fetchone()[0]
|
|
return rows, json.loads(state) if state else None
|
|
finally:
|
|
connection.close()
|
|
|
|
def add_orders(self, count):
|
|
with self.app.app_context():
|
|
connection = db.connect()
|
|
try:
|
|
connection.execute("UPDATE order_test_data SET source_updated_at='2026-11-01 12:00:00' WHERE tenant_id='a'")
|
|
order_data.import_records(connection, [
|
|
{'tenant_id': 'a', 'order_no': f'NEWNORMAL{index:08d}',
|
|
'inbound_code': f'NEWINBOUND{index:08d}', 'status': 'COMPLETED',
|
|
'source_updated_at': '2026-11-06 12:00:00'}
|
|
for index in range(1, count + 1)])
|
|
connection.commit()
|
|
finally:
|
|
connection.close()
|
|
|
|
def test_new_list_is_available_to_followup_before_any_state_is_saved(self):
|
|
before = self.saved()
|
|
calls = 0
|
|
|
|
def provider(body, **kwargs):
|
|
nonlocal calls
|
|
calls += 1
|
|
self.assertEqual(self.saved(), before)
|
|
if calls == 1:
|
|
return tool_plan('query_orders', order_arguments('列出我的异常订单', 'list', 'exception'), 'new_list')
|
|
if calls == 2:
|
|
self.assertIn('INBOUND20260001', evidence(body)[0]['result'])
|
|
return tool_plan('query_orders', order_arguments('第一个订单的轨迹'), 'first_order')
|
|
facts = evidence(body)
|
|
self.assertIn('VISIBLE20260001', facts[1]['result'])
|
|
self.assertNotIn('请先查看', facts[1]['result'])
|
|
return final_facts('fact_0', 'fact_1')
|
|
|
|
with patch('logistics_chat.ai_client.post_completion', side_effect=provider):
|
|
events = self.ask('列出我的异常订单，再看第一票的轨迹')
|
|
self.assertEqual(calls, 3)
|
|
self.assertIn('VISIBLE20260001', events[0]['t'])
|
|
rows, state = self.saved()
|
|
self.assertEqual(len(rows), len(before[0]) + 2)
|
|
self.assertTrue(state['query']['exception'])
|
|
self.assertEqual(state['selected_index'], 1)
|
|
self.assertEqual(state['shown'][0]['order_no'], 'VISIBLE20260001')
|
|
|
|
def test_current_filter_replaces_an_older_persisted_list_for_ordinal_lookup(self):
|
|
self.add_orders(2)
|
|
with patch('logistics_chat.ai_client.post_completion', side_effect=[
|
|
tool_plan('query_orders', order_arguments('列出我的订单', 'list'), 'old_list'), final_facts('fact_0')]):
|
|
self.ask('先列出全部订单')
|
|
old_state = self.saved()[1]
|
|
self.assertEqual(old_state['shown'][0]['order_no'], 'NEWNORMAL00000002')
|
|
calls = 0
|
|
|
|
def provider(body, **kwargs):
|
|
nonlocal calls
|
|
calls += 1
|
|
if calls == 1:
|
|
return tool_plan('query_orders', order_arguments('列出我的异常订单', 'list', 'exception'), 'exceptions')
|
|
if calls == 2:
|
|
self.assertEqual(self.saved()[1], old_state)
|
|
return tool_plan('query_orders', order_arguments('第一个订单的轨迹'), 'current_first')
|
|
current = evidence(body)[1]['result']
|
|
self.assertIn('VISIBLE20260001', current)
|
|
self.assertNotIn('NEWNORMAL00000002', current)
|
|
return final_facts('fact_0', 'fact_1')
|
|
|
|
with patch('logistics_chat.ai_client.post_completion', side_effect=provider):
|
|
self.ask('改看异常订单列表，再查其中第一票')
|
|
self.assertEqual(self.saved()[1]['shown'][0]['order_no'], 'VISIBLE20260001')
|
|
|
|
def test_repeated_next_page_calls_advance_the_pending_page_instead_of_reusing_cache(self):
|
|
self.add_orders(45)
|
|
before = self.saved()
|
|
self.app.config['MODEL_MAX_TOOL_ROUNDS'] = 3
|
|
self.app.config['MODEL_MAX_EVIDENCE_CHARS'] = 50000
|
|
calls = 0
|
|
|
|
def provider(body, **kwargs):
|
|
nonlocal calls
|
|
calls += 1
|
|
self.assertEqual(self.saved(), before)
|
|
if calls == 1:
|
|
return tool_plan('query_orders', order_arguments('列出我的订单', 'list'), 'page_one')
|
|
if calls == 2:
|
|
self.assertIn('第1页', evidence(body)[-1]['result'])
|
|
return tool_plan('query_orders', order_arguments('下一页'), 'page_two')
|
|
if calls == 3:
|
|
self.assertIn('第2页', evidence(body)[-1]['result'])
|
|
return tool_plan('query_orders', order_arguments('下一页'), 'page_three')
|
|
self.assertIn('第3页', evidence(body)[-1]['result'])
|
|
return final_facts('fact_0', 'fact_1', 'fact_2')
|
|
|
|
with patch('logistics_chat.ai_client.post_completion', side_effect=provider):
|
|
self.ask('列出订单，接着看第二页和第三页')
|
|
self.assertEqual(calls, 4)
|
|
state = self.saved()[1]
|
|
self.assertEqual(state['page'], 3)
|
|
self.assertEqual(len(state['shown']), 6)
|
|
|
|
def test_selected_index_advances_within_the_turn_and_remains_available_next_turn(self):
|
|
self.add_orders(2)
|
|
calls = 0
|
|
|
|
def provider(body, **kwargs):
|
|
nonlocal calls
|
|
calls += 1
|
|
if calls == 1:
|
|
return tool_plan('query_orders', order_arguments('列出我的订单', 'list'), 'orders')
|
|
if calls == 2:
|
|
return tool_plan('query_orders', order_arguments('第一个订单的轨迹'), 'first')
|
|
if calls == 3:
|
|
self.assertIn('NEWNORMAL00000002', evidence(body)[-1]['result'])
|
|
return tool_plan('query_orders', order_arguments('下一票的轨迹'), 'next')
|
|
self.assertIn('NEWNORMAL00000001', evidence(body)[-1]['result'])
|
|
return final_facts('fact_0', 'fact_1', 'fact_2')
|
|
|
|
with patch('logistics_chat.ai_client.post_completion', side_effect=provider):
|
|
self.ask('列出全部订单，依次查第一票和下一票的轨迹')
|
|
self.assertEqual(self.saved()[1]['selected_index'], 2)
|
|
with patch('logistics_chat.ai_client.post_completion', side_effect=[
|
|
tool_plan('query_orders', order_arguments('下一票的轨迹'), 'third'), final_facts('fact_0')]):
|
|
events = self.ask('再看下一票')
|
|
serialized = json.dumps(events, ensure_ascii=False)
|
|
self.assertIn('VISIBLE20260001', serialized)
|
|
self.assertEqual(self.saved()[1]['selected_index'], 3)
|
|
|
|
def test_list_then_followup_in_one_batch_uses_the_new_candidate(self):
|
|
calls = [tool_plan('query_orders', order_arguments('列出我的异常订单', 'list', 'exception'), 'batch_list').message['tool_calls'][0],
|
|
tool_plan('query_orders', order_arguments('第一个订单的轨迹'), 'batch_followup').message['tool_calls'][0]]
|
|
batch = fixtures.Completion({'role': 'assistant', 'content': '', 'tool_calls': calls}, 'tool_calls')
|
|
with patch('logistics_chat.ai_client.post_completion', side_effect=[batch, final_facts('fact_0', 'fact_1')]) as model:
|
|
events = self.ask('列异常订单并查询第一票的轨迹')
|
|
second_fact = evidence(model.call_args_list[1].args[0])[1]
|
|
self.assertIn('VISIBLE20260001', second_fact['result'])
|
|
self.assertNotIn('请先查看', events[0]['t'])
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|