| import json |
| import unittest |
|
|
| from app.models import SMTPResult |
| from app.validator.retry_queue import ( |
| MemoryRetryStore, |
| RedisRetryStore, |
| RetryEntry, |
| RetryQueue, |
| drain_retry_queue, |
| ) |
|
|
|
|
| class FakeRedis: |
| """Minimal async Redis subset (hash + zset) used by RedisRetryStore.""" |
|
|
| def __init__(self): |
| self.hashes: dict[str, dict[str, str]] = {} |
| self.zsets: dict[str, dict[str, float]] = {} |
|
|
| async def hset(self, key, field, value): |
| self.hashes.setdefault(key, {})[field] = value |
|
|
| async def hget(self, key, field): |
| return self.hashes.get(key, {}).get(field) |
|
|
| async def hdel(self, key, field): |
| self.hashes.get(key, {}).pop(field, None) |
|
|
| async def zadd(self, key, mapping): |
| self.zsets.setdefault(key, {}).update(mapping) |
|
|
| async def zrem(self, key, member): |
| self.zsets.get(key, {}).pop(member, None) |
|
|
| async def zrangebyscore(self, key, min_, max_, start=0, num=None): |
| items = [ |
| (m, s) for m, s in self.zsets.get(key, {}).items() if min_ <= s <= max_ |
| ] |
| items.sort(key=lambda kv: kv[1]) |
| members = [m for m, _ in items] |
| return members[start : (start + num) if num else None] |
|
|
| async def zcard(self, key): |
| return len(self.zsets.get(key, {})) |
|
|
| async def hlen(self, key): |
| return len(self.hashes.get(key, {})) |
|
|
|
|
| class RedisRetryStoreTests(unittest.IsolatedAsyncioTestCase): |
| """Production needs the queue to survive a restart and be shared across |
| workers — an in-process dict does neither.""" |
|
|
| async def test_add_persists_entry(self): |
| store = RedisRetryStore(client=FakeRedis()) |
| entry = RetryEntry(email="a@corp.test", mx_host="mx.corp.test", due_at=500.0) |
| await store.add(entry) |
| got = await store.get("a@corp.test") |
| self.assertIsNotNone(got) |
| self.assertEqual(got.mx_host, "mx.corp.test") |
| self.assertEqual(got.due_at, 500.0) |
|
|
| async def test_due_returns_only_matured_entries(self): |
| store = RedisRetryStore(client=FakeRedis()) |
| await store.add(RetryEntry(email="early@corp.test", mx_host="mx", due_at=100.0)) |
| await store.add(RetryEntry(email="late@corp.test", mx_host="mx", due_at=9999.0)) |
| due = await store.due(now=200.0, limit=10) |
| self.assertEqual([e.email for e in due], ["early@corp.test"]) |
|
|
| async def test_due_is_ordered_by_time(self): |
| store = RedisRetryStore(client=FakeRedis()) |
| await store.add(RetryEntry(email="b@corp.test", mx_host="mx", due_at=300.0)) |
| await store.add(RetryEntry(email="a@corp.test", mx_host="mx", due_at=200.0)) |
| due = await store.due(now=400.0, limit=10) |
| self.assertEqual([e.email for e in due], ["a@corp.test", "b@corp.test"]) |
|
|
| async def test_remove_clears_both_indexes(self): |
| redis = FakeRedis() |
| store = RedisRetryStore(client=redis) |
| await store.add(RetryEntry(email="a@corp.test", mx_host="mx", due_at=1.0)) |
| await store.remove("a@corp.test") |
| self.assertIsNone(await store.get("a@corp.test")) |
| self.assertEqual(await store.count(), 0) |
|
|
| async def test_readd_updates_rather_than_duplicates(self): |
| store = RedisRetryStore(client=FakeRedis()) |
| await store.add(RetryEntry(email="a@corp.test", mx_host="mx", due_at=1.0)) |
| await store.add(RetryEntry(email="a@corp.test", mx_host="mx", due_at=99.0, attempts=1)) |
| self.assertEqual(await store.count(), 1) |
| self.assertEqual((await store.get("a@corp.test")).attempts, 1) |
|
|
| async def test_corrupt_entry_is_skipped_not_raised(self): |
| redis = FakeRedis() |
| store = RedisRetryStore(client=redis) |
| await store.add(RetryEntry(email="ok@corp.test", mx_host="mx", due_at=1.0)) |
| |
| redis.hashes.setdefault("vx:retry:entries", {})["bad@corp.test"] = "{not json" |
| redis.zsets.setdefault("vx:retry:due", {})["bad@corp.test"] = 1.0 |
| due = await store.due(now=5.0, limit=10) |
| self.assertEqual([e.email for e in due], ["ok@corp.test"]) |
|
|
|
|
| class DrainWorkerTests(unittest.IsolatedAsyncioTestCase): |
| """A queue that only fills is a leak. Due entries must actually be retried, |
| and a definitive answer must clear the entry.""" |
|
|
| async def test_drain_reprobes_due_entries_and_resolves_them(self): |
| clock = {"t": 0.0} |
| queue = RetryQueue( |
| store=MemoryRetryStore(), backoff_seconds=10, now=lambda: clock["t"] |
| ) |
| await queue.enqueue("a@corp.test", "mx.corp.test", code=451) |
| clock["t"] = 100.0 |
|
|
| probed: list[tuple[str, str]] = [] |
|
|
| async def prober(email, mx_host): |
| probed.append((email, mx_host)) |
| return SMTPResult(verified=True, smtp_code=250) |
|
|
| stats = await drain_retry_queue(prober, queue=queue, limit=10) |
|
|
| self.assertEqual(probed, [("a@corp.test", "mx.corp.test")]) |
| self.assertEqual(stats["resolved"], 1) |
| self.assertEqual(stats["requeued"], 0) |
| self.assertEqual(await queue.size(), 0) |
|
|
| async def test_still_greylisted_entry_is_requeued_with_backoff(self): |
| clock = {"t": 0.0} |
| queue = RetryQueue( |
| store=MemoryRetryStore(), backoff_seconds=10, max_attempts=3, |
| now=lambda: clock["t"], |
| ) |
| await queue.enqueue("a@corp.test", "mx.corp.test", code=451) |
| clock["t"] = 100.0 |
|
|
| async def prober(email, mx_host): |
| return SMTPResult(verified=None, smtp_code=451, smtp_message="still greylisted") |
|
|
| stats = await drain_retry_queue(prober, queue=queue, limit=10) |
|
|
| self.assertEqual(stats["requeued"], 1) |
| self.assertEqual(await queue.size(), 1) |
| |
| self.assertEqual(await queue.due(), []) |
|
|
| async def test_rejected_entry_is_resolved_and_removed(self): |
| clock = {"t": 0.0} |
| queue = RetryQueue(store=MemoryRetryStore(), backoff_seconds=10, now=lambda: clock["t"]) |
| await queue.enqueue("a@corp.test", "mx.corp.test", code=451) |
| clock["t"] = 100.0 |
|
|
| async def prober(email, mx_host): |
| return SMTPResult(verified=False, smtp_code=550, smtp_message="550 no such user") |
|
|
| stats = await drain_retry_queue(prober, queue=queue, limit=10) |
|
|
| self.assertEqual(stats["resolved"], 1) |
| self.assertEqual(await queue.size(), 0) |
|
|
| async def test_prober_error_requeues_without_crashing(self): |
| clock = {"t": 0.0} |
| queue = RetryQueue(store=MemoryRetryStore(), backoff_seconds=10, now=lambda: clock["t"]) |
| await queue.enqueue("a@corp.test", "mx.corp.test", code=451) |
| clock["t"] = 100.0 |
|
|
| async def prober(email, mx_host): |
| raise RuntimeError("mx exploded") |
|
|
| stats = await drain_retry_queue(prober, queue=queue, limit=10) |
|
|
| self.assertEqual(stats["errors"], 1) |
| self.assertEqual(await queue.size(), 1, "must stay queued for another try") |
|
|
| async def test_empty_queue_is_a_noop(self): |
| queue = RetryQueue(store=MemoryRetryStore(), now=lambda: 0.0) |
|
|
| async def prober(email, mx_host): |
| raise AssertionError("should not be called") |
|
|
| stats = await drain_retry_queue(prober, queue=queue, limit=10) |
| self.assertEqual( |
| stats, {"checked": 0, "resolved": 0, "requeued": 0, "errors": 0, "results": []} |
| ) |
|
|
| async def test_results_are_returned_for_persistence(self): |
| clock = {"t": 0.0} |
| queue = RetryQueue(store=MemoryRetryStore(), backoff_seconds=10, now=lambda: clock["t"]) |
| await queue.enqueue("a@corp.test", "mx.corp.test", code=451) |
| clock["t"] = 100.0 |
|
|
| async def prober(email, mx_host): |
| return SMTPResult(verified=True, smtp_code=250) |
|
|
| stats = await drain_retry_queue(prober, queue=queue, limit=10) |
| self.assertEqual(stats["results"][0][0], "a@corp.test") |
| self.assertTrue(stats["results"][0][1].verified) |
|
|
|
|
| if __name__ == "__main__": |
| unittest.main() |
