File size: 6,028 Bytes
58d0e85 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 | 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):
# Systemic: affects every address at that MX. Retrying per-address would
# flood the queue; the MX-level circuit breaker handles this instead.
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)
# Not due yet.
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)
# Still queued (2nd attempt allowed), and not immediately due.
self.assertEqual(await q.size(), 1)
self.assertEqual(await q.due(), [])
clock["t"] = 601.0 + 1200.0 # exponential: 600 * 2
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() |