178 lines
8.8 KiB
Python
178 lines
8.8 KiB
Python
"""Authorization expiry and quote-state conflicts must not publish private replies."""
|
|
|
|
from hashlib import sha256
|
|
import json
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
from flask import g
|
|
|
|
import test_model_chat as fixtures
|
|
from test_model_tool_loop import evidence, final_facts, quote_arguments, tool_plan
|
|
from test_quotes import frames
|
|
from logistics_chat import db
|
|
from logistics_chat.conversation_state import commit_state, load_state
|
|
from logistics_chat.model_turn_store import latest_message_id, save_exchange
|
|
from logistics_chat.quote_data import load_book
|
|
from logistics_chat.quote_tool import QuoteTurn
|
|
|
|
|
|
class ModelTurnSecurityTests(unittest.TestCase):
|
|
def setUp(self):
|
|
fixtures.ModelChatTests.setUp(self)
|
|
|
|
def messages(self):
|
|
with self.app.app_context():
|
|
connection = db.connect()
|
|
try:
|
|
return connection.execute('SELECT conv_id,role,content FROM messages ORDER BY id').fetchall()
|
|
finally:
|
|
connection.close()
|
|
|
|
def test_stateless_private_results_are_not_sent_after_login_revoked_during_final_model_call(self):
|
|
fixtures.ModelChatTests.login(self)
|
|
before = self.messages()
|
|
requests = []
|
|
|
|
def provider(body, **kwargs):
|
|
requests.append(body)
|
|
if len(requests) == 1:
|
|
return fixtures.plan(('query_orders', '我的异常订单有多少个'))
|
|
self.assertEqual(len(requests), 2)
|
|
facts = evidence(body)
|
|
self.assertTrue(facts)
|
|
self.assertIn('等待资料', facts[0]['result'])
|
|
connection = db.connect()
|
|
connection.execute('DELETE FROM customer_sessions')
|
|
connection.commit()
|
|
connection.close()
|
|
return final_facts('fact_0')
|
|
|
|
with patch('logistics_chat.ai_client.post_completion', side_effect=provider):
|
|
events = frames(self.client.post('/stream_chat', json={
|
|
'msg': '我有多少异常单，帮我看看原因', 'conv_id': None}))
|
|
self.assertEqual(len(requests), 2)
|
|
self.assertTrue(events[-1].get('d'), events)
|
|
self.assertTrue(events[0].get('e') or events[0].get('login_required'), events)
|
|
serialized = json.dumps(events, ensure_ascii=False)
|
|
self.assertNotIn('等待资料', serialized)
|
|
self.assertNotIn('VISIBLE20260001', serialized)
|
|
self.assertNotIn('INBOUND20260001', serialized)
|
|
self.assertFalse(any('card' in event for event in events))
|
|
self.assertEqual(self.messages(), before)
|
|
|
|
def test_bound_private_results_are_not_saved_after_login_revoked_during_final_model_call(self):
|
|
fixtures.ModelChatTests.login(self)
|
|
before = self.messages()
|
|
calls = 0
|
|
|
|
def provider(body, **kwargs):
|
|
nonlocal calls
|
|
calls += 1
|
|
if calls == 1:
|
|
return fixtures.plan(('query_orders', '我的异常订单有多少个'))
|
|
self.assertIn('等待资料', evidence(body)[0]['result'])
|
|
connection = db.connect()
|
|
connection.execute('DELETE FROM customer_sessions')
|
|
connection.commit()
|
|
connection.close()
|
|
return final_facts('fact_0')
|
|
|
|
with patch('logistics_chat.ai_client.post_completion', side_effect=provider):
|
|
events = frames(self.client.post('/stream_chat', json={
|
|
'msg': '我有多少异常单', 'conv_id': self.cid}))
|
|
self.assertEqual(calls, 2)
|
|
self.assertIn('e', events[0])
|
|
self.assertNotIn('等待资料', json.dumps(events, ensure_ascii=False))
|
|
self.assertEqual(self.messages(), before)
|
|
|
|
def test_greeting_tool_only_returns_customer_facing_text(self):
|
|
responses = [tool_plan('read_service_info', {'question': '你好'}, 'greeting'), final_facts('fact_0')]
|
|
with patch('logistics_chat.ai_client.post_completion', side_effect=responses):
|
|
events = frames(self.client.post('/stream_chat', json={'msg': '你好', 'conv_id': self.cid}))
|
|
self.assertTrue(events[-1].get('d'), events)
|
|
self.assertIn('t', events[0])
|
|
self.assertTrue(events[0]['t'].strip())
|
|
for internal in ('可以回应问候', '针对原问题澄清', '本次没有查得其他业务事实', '报价工具', 'query_orders', 'fact_0'):
|
|
self.assertNotIn(internal, events[0]['t'])
|
|
self.assertIn(('assistant', events[0]['t']), [(role, text) for _, role, text in self.messages()])
|
|
|
|
def test_revocation_immediately_before_transaction_saves_neither_reply_nor_quote_state(self):
|
|
fixtures.ModelChatTests.login(self)
|
|
before = self.messages()
|
|
order_call = fixtures.plan(('query_orders', '我的异常订单有多少个')).message['tool_calls'][0]
|
|
quote_call = tool_plan('quote_shipping', quote_arguments(shipment={
|
|
'country': 'US', 'total_weight_kg': '1000', 'origin': '义乌', 'zone': '美西'}),
|
|
'quote_candidate').message['tool_calls'][0]
|
|
plan = fixtures.Completion({'role': 'assistant', 'content': '',
|
|
'tool_calls': [order_call, quote_call]}, 'tool_calls')
|
|
save_attempts = []
|
|
|
|
def revoke_then_save(*args, **kwargs):
|
|
turn = kwargs['quote_turn']
|
|
self.assertIsNotNone(turn.candidate)
|
|
self.assertEqual(turn.candidate['intent']['weight'], '1000')
|
|
connection = db.connect()
|
|
connection.execute('DELETE FROM customer_sessions')
|
|
connection.commit()
|
|
connection.close()
|
|
result = save_exchange(*args, **kwargs)
|
|
save_attempts.append(result)
|
|
return result
|
|
|
|
with patch('logistics_chat.ai_client.post_completion', side_effect=[plan, final_facts('fact_0', 'fact_1')]), \
|
|
patch('logistics_chat.model_chat_service.save_exchange', side_effect=revoke_then_save):
|
|
events = frames(self.client.post('/stream_chat', json={
|
|
'msg': '查异常单，同时核算一吨货美国美西的运费', 'conv_id': self.cid}))
|
|
self.assertEqual(save_attempts, ['deleted'])
|
|
self.assertIn('e', events[0])
|
|
self.assertNotIn('等待资料', json.dumps(events, ensure_ascii=False))
|
|
self.assertEqual(self.messages(), before)
|
|
with self.app.app_context():
|
|
connection = db.connect()
|
|
stored = connection.execute('SELECT conversation_state,order_list_state FROM chat_history WHERE id=?',
|
|
(self.cid,)).fetchone()
|
|
connection.close()
|
|
self.assertEqual(stored, (None, None))
|
|
|
|
def test_state_cas_failure_rolls_back_inserted_exchange(self):
|
|
with self.app.test_request_context():
|
|
g.visitor_id = sha256(('6' * 64).encode()).hexdigest()
|
|
book = load_book()
|
|
stale, competing = (QuoteTurn(self.cid, '查询报价', book) for _ in range(2))
|
|
stale.run(quote_arguments(shipment={'country': 'US', 'total_weight_kg': '1000'}))
|
|
competing.run(quote_arguments(shipment={'country': 'US', 'total_weight_kg': '2000'}))
|
|
previous_message_id = latest_message_id(self.cid)
|
|
connection = db.connect()
|
|
self.assertTrue(commit_state(connection, self.cid, competing.candidate, competing.expected_version))
|
|
connection.commit()
|
|
connection.close()
|
|
before = self.messages()
|
|
result = save_exchange(self.cid, '旧的一吨报价', '旧报价事实',
|
|
previous_message_id=previous_message_id, quote_turn=stale)
|
|
self.assertEqual(result, 'stale')
|
|
self.assertEqual(self.messages(), before)
|
|
self.assertEqual(load_state(self.cid)['intent']['weight'], '2000')
|
|
self.assertEqual(load_state(self.cid)['version'], 1)
|
|
|
|
def test_deleted_conversation_is_not_recreated_by_saving_exchange(self):
|
|
with self.app.test_request_context():
|
|
g.visitor_id = sha256(('6' * 64).encode()).hexdigest()
|
|
turn = QuoteTurn(self.cid, '查询报价', load_book())
|
|
turn.run(quote_arguments(shipment={'country': 'US', 'total_weight_kg': '1000'}))
|
|
previous_message_id = latest_message_id(self.cid)
|
|
connection = db.connect()
|
|
connection.execute('DELETE FROM messages WHERE conv_id=?', (self.cid,))
|
|
connection.execute('DELETE FROM chat_history WHERE id=?', (self.cid,))
|
|
connection.commit()
|
|
connection.close()
|
|
result = save_exchange(self.cid, '迟到请求', '迟到报价',
|
|
previous_message_id=previous_message_id, quote_turn=turn)
|
|
self.assertEqual(result, 'deleted')
|
|
self.assertIsNone(load_state(self.cid))
|
|
self.assertFalse(any(cid == self.cid for cid, _, _ in self.messages()))
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|