92 lines
4.7 KiB
Python
92 lines
4.7 KiB
Python
"""Task removal preserves delivery evidence and never creates another send."""
|
|
import tempfile
|
|
import time
|
|
import unittest
|
|
from pathlib import Path
|
|
from proactive_contact_service import ContactService
|
|
from proactive_contact_scheduler import ContactScheduler
|
|
from proactive_contact_store import ContactStore
|
|
from test_proactive_contact import FakeGateway, FakeTransport
|
|
|
|
|
|
class DeleteTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.temp.cleanup)
|
|
self.gateway, self.transport = FakeGateway(), FakeTransport()
|
|
self.service = ContactService(self.temp.name, self.gateway, self.transport)
|
|
|
|
def task(self, conversation='M:2'):
|
|
key = self.service.add('owner', [dict(account='1', conversation=conversation, binding_id='binding')])['added'][0]
|
|
p = self.service.preview('owner', [key], {}, text='请问下午方便联系吗？')['preview']
|
|
self.service.confirm('owner', p['id'])
|
|
return self.service.store.tasks('owner')[0]
|
|
|
|
def test_delete_unsent_batch_survives_restart_and_cannot_resume(self):
|
|
tasks = [self.task('M:2'), self.task('M:3')]
|
|
result = self.service.delete_tasks('owner', [t['id'] for t in tasks])
|
|
self.assertEqual(len(result['deleted']), 2)
|
|
store = ContactStore(self.temp.name)
|
|
self.assertEqual(store.tasks('owner'), [])
|
|
for task in tasks:
|
|
self.assertIsNone(store.claim('owner', task['id'], time.time()+100))
|
|
self.assertFalse(store.update('owner', task['id'], 'scheduled'))
|
|
ContactScheduler(self.service).tick('owner')
|
|
self.assertEqual(self.transport.sent, [])
|
|
self.assertEqual(len(store.customers('owner')), 2)
|
|
|
|
def test_delete_sent_keeps_receipt_pacing_and_dedup(self):
|
|
task = self.task()
|
|
ContactScheduler(self.service).tick('owner')
|
|
pacing = self.service.store.pacing('1', time.time())
|
|
result = self.service.delete_tasks('owner', [task['id']])
|
|
self.assertEqual(result['deleted'], [task['id']])
|
|
self.assertIsNone(self.service.store.get_task('owner', task['id']))
|
|
saved = self.service.store.get_task('owner', task['id'], include_deleted=True)
|
|
self.assertEqual(saved['status'], 'succeeded')
|
|
self.assertEqual(saved['receipt']['serverId'], 'server-1')
|
|
self.assertEqual(self.service.store.pacing('1', time.time()), pacing)
|
|
p = self.service.preview('owner', [task['customer']], {}, text=task['text'])['preview']
|
|
with self.assertRaisesRegex(ValueError, '相同内容'):
|
|
self.service.confirm('owner', p['id'])
|
|
self.assertEqual(len(self.transport.sent), 1)
|
|
self.assertEqual(self.service.delete_tasks('owner', [task['id']])['deleted'], [])
|
|
|
|
def test_mixed_batch_rejects_sending_and_foreign_ids(self):
|
|
a, b = self.task(), self.task('M:3')
|
|
self.service.store.claim('owner', a['id'], time.time()+1)
|
|
result = self.service.delete_tasks('owner', [a['id'], b['id'], 'missing'])
|
|
self.assertEqual(result['deleted'], [b['id']])
|
|
self.assertEqual(len(result['errors']), 2)
|
|
self.assertEqual(self.service.store.get_task('owner', a['id'])['status'], 'sending')
|
|
deleted, errors = self.service.store.delete_tasks('other', [a['id']])
|
|
self.assertEqual(deleted, [])
|
|
self.assertTrue(errors)
|
|
self.gateway.identity = 'other'
|
|
with self.assertRaisesRegex(ValueError, '账号已切换'):
|
|
self.service.delete_tasks('owner', [a['id']])
|
|
|
|
def test_unknown_deleted_task_can_reconcile_but_never_resend(self):
|
|
task = self.task()
|
|
claimed = self.service.store.claim('owner', task['id'], time.time()+1)
|
|
self.service.store.finish(claimed, 'delivery_unknown', error='回执未到')
|
|
self.service.delete_tasks('owner', [task['id']])
|
|
p = self.service.preview('owner', [task['customer']], {}, text=task['text'])['preview']
|
|
with self.assertRaisesRegex(ValueError, '相同内容'):
|
|
self.service.confirm('owner', p['id'])
|
|
self.transport.receipt = lambda t: dict(status='confirmed', accountId=t['account'], conversationId=t['conversation'], serverId='12', senderVerified=True, textVerified=True)
|
|
self.assertEqual(self.service.reconcile_receipts('owner'), 1)
|
|
self.assertEqual(self.service.snapshot('owner')['tasks'], [])
|
|
self.assertEqual(self.transport.sent, [])
|
|
|
|
def test_invalid_selection_does_not_delete(self):
|
|
task = self.task()
|
|
for ids in ([], None, [None], [task['id']]*101):
|
|
with self.assertRaises(ValueError):
|
|
self.service.delete_tasks('owner', ids)
|
|
self.assertEqual(len(self.service.store.tasks('owner')), 1)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|