230 lines
12 KiB
Python
230 lines
12 KiB
Python
"""Real protocol receive/process/dedup with fake history, AI and sender only."""
|
|
from copy import deepcopy
|
|
from pathlib import Path
|
|
import tempfile
|
|
import json
|
|
import time
|
|
import unittest
|
|
from unittest.mock import Mock, patch
|
|
import message_origin
|
|
from proactive_replies import incoming_after_proactive, is_proactive_customer_turn
|
|
from protocol_engine import ProtocolBot, session_key
|
|
|
|
|
|
class ProactiveRepliesTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.temp.cleanup)
|
|
self.root = Path(self.temp.name)
|
|
self.now = time.time()
|
|
self.account, self.conv = '1', 'M:2'
|
|
with patch('message_origin.time.time', return_value=self.now-60):
|
|
message_origin.record_attempt('proactive-contact:test', '1', 'M:2', '下午方便联系吗？', 'system', root=self.root)
|
|
message_origin.finish('proactive-contact:test', 'confirmed', '101', root=self.root)
|
|
self.messages = [self.message(1, True, '下午方便联系吗？', self.now-50),
|
|
self.message(2, False, '下午三点可以，怎么安排？', self.now-10)]
|
|
self.db = Mock()
|
|
self.db.health_check.return_value = True
|
|
self.db.get_new_messages.return_value = [] # Reply predates listener start.
|
|
self.db.get_conversation_context_by_id.side_effect = lambda *a, **kw: self.context()
|
|
self.sender = Mock()
|
|
self.sender.lookup.return_value = None
|
|
self.sender.send.side_effect = lambda *a, **kw: self.send_receipt(kw['guard'])
|
|
self.generate = Mock(return_value='收到，下午三点联系您。')
|
|
assistant = Mock()
|
|
assistant.enabled.return_value = False
|
|
self.assistant = assistant
|
|
self.enterContext(patch('protocol_engine.application_data_dir', return_value=self.root))
|
|
self.enterContext(patch('queue_log.QueueLog'))
|
|
self.enterContext(patch('conversation_blacklist.blocked_reason', return_value=''))
|
|
self.enterContext(patch('ai_config.AI_REVIEW_RULES', []))
|
|
|
|
def message(self, row, own, text, stamp):
|
|
return dict(account='1', conv_id='M:2', sender_id='1' if own else '2', is_self=own,
|
|
server_id=str(100+row), rowid=row, dedup_key=f'1:M:2:server:{100+row}',
|
|
content=text, content_type=0, send_time=stamp, display_name='模拟客户')
|
|
|
|
def context(self):
|
|
return dict(account='1', conv_id='M:2', messages=self.messages, last_message=self.messages[-1],
|
|
display_name='模拟客户', text='我：下午方便联系吗？\n客户：下午三点可以，怎么安排？')
|
|
|
|
def send_receipt(self, guard):
|
|
self.assertTrue(guard())
|
|
return dict(status='confirmed', accountId='1', conversationId='M:2', serverId='200', senderVerified=True, textVerified=True)
|
|
|
|
def bot(self):
|
|
bot = ProtocolBot(db=self.db, sender=self.sender, identity=dict(accountId='1', pid=0),
|
|
path=self.root/'pending_replies.protocol.json', generate=self.generate,
|
|
review_assistant=self.assistant, log=lambda s: None)
|
|
bot.message_batch_window_seconds = bot.send_delay_seconds = 0
|
|
return bot
|
|
|
|
def test_reply_before_listener_start_is_sent_once_even_after_restart(self):
|
|
bot = self.bot()
|
|
self.assertGreater(bot._since, self.messages[-1]['send_time'])
|
|
bot.poll_once()
|
|
self.assertEqual(self.generate.call_count, 1)
|
|
self.assertEqual(self.sender.send.call_count, 1)
|
|
self.assertEqual(self.sender.send.call_args.args[2:5], ('1', 'M:2', '收到，下午三点联系您。'))
|
|
bot._proactive_scan_at = 0
|
|
bot.poll_once()
|
|
self.bot().poll_once()
|
|
self.assertEqual(self.sender.send.call_count, 1)
|
|
|
|
def test_review_and_unknown_send_protection_are_preserved(self):
|
|
bot = self.bot()
|
|
bot.send_mode = 'review'
|
|
bot.poll_once()
|
|
key = session_key('1', 'M:2')
|
|
self.assertTrue(bot._pending[key]['awaiting_review'])
|
|
self.sender.send.assert_not_called()
|
|
bot._pending[key].update(send_state='uncertain', stage='receipt_check')
|
|
old = bot._pending[key]
|
|
self.messages.append(self.message(3, False, '还有个问题？', self.now))
|
|
bot._proactive_scan_at = 0
|
|
bot.poll_once()
|
|
self.assertIs(bot._pending[key], old)
|
|
self.assertEqual(old['send_state'], 'uncertain')
|
|
self.sender.send.assert_not_called()
|
|
|
|
def test_unproven_wrong_identity_or_already_answered_is_not_replayed(self):
|
|
original = deepcopy(self.messages)
|
|
cases = [lambda: self.messages[0].update(server_id='999'),
|
|
lambda: self.messages[0].update(sender_id='other'),
|
|
lambda: self.messages[-1].update(account='other'),
|
|
lambda: self.messages[-1].update(conv_id='M:3'),
|
|
lambda: self.messages[-1].update(send_time=self.now-100),
|
|
lambda: self.messages.append(self.message(3, True, '已回复', self.now))]
|
|
for mutate in cases:
|
|
self.messages = deepcopy(original)
|
|
mutate()
|
|
self.assertEqual(incoming_after_proactive(self.db, '1', self.root, now=self.now), [])
|
|
self.messages = original
|
|
self.assertEqual(incoming_after_proactive(self.db, 'other', self.root, now=self.now), [])
|
|
message_origin.finish('proactive-contact:test', 'confirmed', '999', root=self.root)
|
|
self.assertEqual(incoming_after_proactive(self.db, '1', self.root, now=self.now), [])
|
|
|
|
def test_no_confirmation_no_backfill_and_stopped_listener_does_nothing(self):
|
|
with message_origin.connect(self.root) as db:
|
|
db.execute("UPDATE origins SET status='attempted'")
|
|
self.assertEqual(incoming_after_proactive(self.db, '1', self.root), [])
|
|
message_origin.finish('proactive-contact:test', 'confirmed', '101', root=self.root)
|
|
bot = self.bot()
|
|
bot.stop_event.set()
|
|
bot.poll_once()
|
|
self.sender.send.assert_not_called()
|
|
self.generate.assert_not_called()
|
|
|
|
def dual_role(self):
|
|
from review_assistant import ReviewAssistantCoordinator, SETTINGS_KEY
|
|
self.assistant = ReviewAssistantCoordinator(root=self.root, account='1', db=self.db,
|
|
settings_loader=lambda: {SETTINGS_KEY: {'1': {'enabled': True, 'recipient_conv_id': 'M:2'}}})
|
|
self.addCleanup(self.assistant.close)
|
|
# Notification delivery isn't part of this incoming-message test.
|
|
self.assistant.enqueue_notifications = Mock()
|
|
|
|
def test_assistant_recipient_answering_proactive_question_enters_queue_and_replies(self):
|
|
self.dual_role()
|
|
self.assertTrue(self.assistant.is_recipient('M:2'))
|
|
bot = self.bot()
|
|
bot.message_batch_window_seconds = 60
|
|
bot.poll_once()
|
|
key = session_key('1', 'M:2')
|
|
self.assertEqual(bot._pending[key]['staged_user_text'], self.messages[-1]['content'])
|
|
self.assertEqual(bot._pending[key]['stage'], 'queued')
|
|
bot._pending[key]['ready_at'] = 0
|
|
bot.poll_once()
|
|
self.assertEqual(self.sender.send.call_count, 1)
|
|
self.assertTrue(self.assistant.is_recipient('M:2'), 'must retain assistant configuration')
|
|
|
|
def test_ai_followup_keeps_customer_turn_but_assistant_reminder_ends_it(self):
|
|
self.dual_role()
|
|
message_origin.record_attempt('incoming-question', '1', 'M:2', '收到', 'ai', root=self.root)
|
|
message_origin.finish('incoming-question', 'confirmed', '103', root=self.root)
|
|
self.messages.extend([self.message(3, True, '收到', self.now-5), self.message(4, False, '那明天呢？', self.now-3)])
|
|
self.assertTrue(is_proactive_customer_turn(self.db, '1', 'M:2', self.root))
|
|
message_origin.record_attempt('review-assistant:notice', '1', 'M:2', '审核提醒', 'system', root=self.root)
|
|
message_origin.finish('review-assistant:notice', 'confirmed', '105', root=self.root)
|
|
self.messages.extend([self.message(5, True, '审核提醒', self.now-2), self.message(6, False, '收到提醒', self.now-1)])
|
|
bot = self.bot()
|
|
bot.poll_once()
|
|
self.assertEqual(bot._pending, {})
|
|
self.sender.send.assert_not_called()
|
|
|
|
def test_dual_role_keeps_risk_review_and_unknown_delivery_protection(self):
|
|
self.dual_role()
|
|
bot = self.bot()
|
|
bot.send_mode = 'review'
|
|
bot.poll_once()
|
|
key = session_key('1', 'M:2')
|
|
self.assertTrue(bot._pending[key]['awaiting_review'])
|
|
self.sender.send.assert_not_called()
|
|
bot._pending[key].update(send_state='uncertain', stage='receipt_check')
|
|
self.messages.append(self.message(3, False, '还有个问题', self.now))
|
|
bot._upsert(self.messages[-1])
|
|
self.assertEqual(bot._pending[key]['send_state'], 'uncertain')
|
|
self.sender.send.assert_not_called()
|
|
|
|
def test_no_exact_proactive_evidence_keeps_assistant_exclusion(self):
|
|
self.dual_role()
|
|
for mutation in ('unconfirmed', 'wrong-account', 'wrong-conversation', 'wrong-receipt'):
|
|
with self.subTest(mutation=mutation), message_origin.connect(self.root) as conn:
|
|
conn.execute("UPDATE origins SET status='confirmed',account='1',conversation='M:2',server_id='101'")
|
|
column, value = {'unconfirmed': ('status','attempted'), 'wrong-account': ('account','2'),
|
|
'wrong-conversation': ('conversation','M:3'), 'wrong-receipt': ('server_id','999')}[mutation]
|
|
conn.execute(f'UPDATE origins SET {column}=?', (value,))
|
|
self.assertFalse(is_proactive_customer_turn(self.db, '1', 'M:2', self.root))
|
|
|
|
def repair_ticket(self, **changes):
|
|
message = self.messages[-1]
|
|
row = {k:message[k] for k in ('account','conv_id','dedup_key')}
|
|
row.update(reason='assistant_recipient_after_proactive', **changes)
|
|
(self.root/'protocol_inbound_recovery.json').write_text(json.dumps({'requests':[row]}), encoding='utf-8')
|
|
|
|
def test_exact_filtered_event_can_be_restored_once_including_after_restart(self):
|
|
self.dual_role()
|
|
self.repair_ticket()
|
|
bot = self.bot()
|
|
key = session_key('1', 'M:2')
|
|
bot._seen[key] = self.messages[-1]['dedup_key'] # Prior version silently consumed it.
|
|
bot._save()
|
|
bot.poll_once()
|
|
self.assertEqual(self.sender.send.call_count, 1)
|
|
self.bot().poll_once()
|
|
self.assertEqual(self.sender.send.call_count, 1)
|
|
|
|
def test_seen_messages_without_matching_repair_ticket_are_not_replayed(self):
|
|
self.dual_role()
|
|
bot = self.bot()
|
|
key = session_key('1', 'M:2')
|
|
bot._seen[key] = self.messages[-1]['dedup_key']
|
|
bot.poll_once()
|
|
self.sender.send.assert_not_called()
|
|
self.repair_ticket(account='other')
|
|
bot._proactive_scan_at = 0
|
|
bot.poll_once()
|
|
self.sender.send.assert_not_called()
|
|
|
|
def test_repair_cannot_replay_attempted_or_handed_off_message(self):
|
|
self.dual_role()
|
|
self.repair_ticket()
|
|
bot = self.bot()
|
|
key = session_key('1', 'M:2')
|
|
event = self.messages[-1]['dedup_key']
|
|
bot._seen[key] = event
|
|
for outcome in ('unknown', 'confirmed', 'failed'):
|
|
self.sender.lookup.return_value = {'status': outcome}
|
|
bot._proactive_scan_at = 0
|
|
bot.poll_once()
|
|
self.assertEqual(bot._pending, {})
|
|
self.sender.lookup.return_value = None
|
|
bot._manual_handoffs[key] = {'state': {'dedup_key': event}}
|
|
bot._proactive_scan_at = 0
|
|
bot.poll_once()
|
|
self.assertEqual(bot._pending, {})
|
|
self.sender.send.assert_not_called()
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|