| import unittest |
|
|
| from app.validator.retry_queue import ( |
| MemoryRetryStore, |
| RetryEntry, |
| RetryQueue, |
| is_greylist_result, |
| ) |
|
|
|
|
| class GreylistDetectionTests(unittest.TestCase): |
| """Only a genuine SMTP 4xx (greylisting / temporary mailbox error) is worth |
| retrying. A transport failure is systemic to the MX — re-queuing every |
| address behind it would flood the queue for no gain.""" |
|
|
| def _result(self, **kw): |
| from app.models import SMTPResult |
|
|
| base = dict(verified=None) |
| base.update(kw) |
| return SMTPResult(**base) |
|
|
| def test_4xx_is_retryable(self): |
| self.assertTrue(is_greylist_result(self._result(smtp_code=451))) |
| self.assertTrue(is_greylist_result(self._result(smtp_code=421))) |
|
|
| def test_5xx_is_not_retryable(self): |
| self.assertFalse(is_greylist_result(self._result(verified=False, smtp_code=550))) |
|
|
| def test_250_is_not_retryable(self): |
| self.assertFalse(is_greylist_result(self._result(verified=True, smtp_code=250))) |
|
|
| def test_transport_failure_is_not_retryable(self): |
| |
| |
| self.assertFalse( |
| is_greylist_result(self._result(error="transport_error: connect timed out")) |
| ) |
|
|
| def test_none_result_is_not_retryable(self): |
| self.assertFalse(is_greylist_result(None)) |
|
|
|
|
| class RetryQueueTests(unittest.IsolatedAsyncioTestCase): |
| async def test_enqueue_schedules_a_future_attempt(self): |
| q = RetryQueue(store=MemoryRetryStore(), backoff_seconds=900, now=lambda: 1000.0) |
| ok = await q.enqueue("a@corp.test", "mx.corp.test", code=451) |
| self.assertTrue(ok) |
| self.assertEqual(await q.size(), 1) |
| |
| self.assertEqual(await q.due(), []) |
|
|
| async def test_entry_becomes_due_after_backoff(self): |
| clock = {"t": 1000.0} |
| q = RetryQueue(store=MemoryRetryStore(), backoff_seconds=900, now=lambda: clock["t"]) |
| await q.enqueue("a@corp.test", "mx.corp.test", code=451) |
| clock["t"] = 1901.0 |
| due = await q.due() |
| self.assertEqual(len(due), 1) |
| self.assertEqual(due[0].email, "a@corp.test") |
| self.assertEqual(due[0].mx_host, "mx.corp.test") |
|
|
| async def test_duplicate_enqueue_does_not_create_twice(self): |
| q = RetryQueue(store=MemoryRetryStore(), now=lambda: 1.0) |
| await q.enqueue("a@corp.test", "mx.corp.test", code=451) |
| await q.enqueue("a@corp.test", "mx.corp.test", code=451) |
| self.assertEqual(await q.size(), 1) |
|
|
| async def test_email_matching_is_case_insensitive(self): |
| q = RetryQueue(store=MemoryRetryStore(), now=lambda: 1.0) |
| await q.enqueue("A@Corp.test", "mx.corp.test", code=451) |
| await q.enqueue("a@corp.test", "mx.corp.test", code=451) |
| self.assertEqual(await q.size(), 1) |
|
|
| async def test_attempt_increments_and_reschedules_with_backoff(self): |
| clock = {"t": 0.0} |
| q = RetryQueue( |
| store=MemoryRetryStore(), backoff_seconds=600, max_attempts=3, now=lambda: clock["t"] |
| ) |
| await q.enqueue("a@corp.test", "mx.corp.test", code=451) |
| clock["t"] = 601.0 |
| entry = (await q.due())[0] |
| await q.mark_attempted(entry) |
|
|
| |
| self.assertEqual(await q.size(), 1) |
| self.assertEqual(await q.due(), []) |
| clock["t"] = 601.0 + 1200.0 |
| self.assertEqual(len(await q.due()), 1) |
|
|
| async def test_max_attempts_removes_the_entry(self): |
| clock = {"t": 0.0} |
| q = RetryQueue( |
| store=MemoryRetryStore(), backoff_seconds=10, max_attempts=2, now=lambda: clock["t"] |
| ) |
| await q.enqueue("a@corp.test", "mx.corp.test", code=451) |
| clock["t"] = 11.0 |
| await q.mark_attempted((await q.due())[0]) |
| clock["t"] = 40.0 |
| await q.mark_attempted((await q.due())[0]) |
| self.assertEqual(await q.size(), 0, "must not retry forever") |
|
|
| async def test_resolve_removes_the_entry(self): |
| clock = {"t": 0.0} |
| q = RetryQueue(store=MemoryRetryStore(), backoff_seconds=10, now=lambda: clock["t"]) |
| await q.enqueue("a@corp.test", "mx.corp.test", code=451) |
| clock["t"] = 11.0 |
| await q.resolve((await q.due())[0]) |
| self.assertEqual(await q.size(), 0) |
|
|
| async def test_due_respects_limit(self): |
| clock = {"t": 0.0} |
| q = RetryQueue(store=MemoryRetryStore(), backoff_seconds=1, now=lambda: clock["t"]) |
| for i in range(5): |
| await q.enqueue(f"u{i}@corp.test", "mx.corp.test", code=451) |
| clock["t"] = 5.0 |
| self.assertEqual(len(await q.due(limit=2)), 2) |
|
|
| async def test_store_failure_does_not_raise(self): |
| class BrokenStore: |
| async def add(self, *a, **k): |
| raise RuntimeError("redis down") |
|
|
| async def due(self, *a, **k): |
| raise RuntimeError("redis down") |
|
|
| async def remove(self, *a, **k): |
| raise RuntimeError("redis down") |
|
|
| async def get(self, *a, **k): |
| raise RuntimeError("redis down") |
|
|
| async def count(self, *a, **k): |
| raise RuntimeError("redis down") |
|
|
| q = RetryQueue(store=BrokenStore()) |
| self.assertFalse(await q.enqueue("a@corp.test", "mx.corp.test", code=451)) |
| self.assertEqual(await q.due(), []) |
| self.assertEqual(await q.size(), 0) |
|
|
|
|
| class RetryEntryTests(unittest.TestCase): |
| def test_entry_serialises(self): |
| e = RetryEntry(email="a@corp.test", mx_host="mx.corp.test", attempts=1, due_at=123.0, last_code=451) |
| data = e.as_dict() |
| self.assertEqual(data["email"], "a@corp.test") |
| back = RetryEntry.from_dict(data) |
| self.assertEqual(back.email, e.email) |
| self.assertEqual(back.due_at, e.due_at) |
|
|
|
|
| if __name__ == "__main__": |
| unittest.main() |
