File size: 2,368 Bytes
0f6b827 | 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 | import asyncio
import unittest
from unittest.mock import AsyncMock, MagicMock
from app.models import SMTPResult
from app.tasks.drain_loop import DrainLoop
class DrainLoopTests(unittest.IsolatedAsyncioTestCase):
"""The retry queue fills as addresses greylist. Without a periodic drain
the queue is a leak: greylisted mailboxes stay permanently risky. The
Dockerfile CMD runs uvicorn only (no Celery beat), so the drain must run
in-process."""
async def test_drain_loop_calls_the_prober_periodically(self):
prober = AsyncMock(return_value=SMTPResult(verified=True, smtp_code=250))
loop = DrainLoop(prober=prober, interval_seconds=0.05, limit=10)
task = asyncio.create_task(loop.start())
await asyncio.sleep(0.2)
await loop.stop()
await task
self.assertGreaterEqual(loop.drain_count, 1)
async def test_stop_terminates_cleanly(self):
prober = AsyncMock(return_value=SMTPResult(verified=True, smtp_code=250))
loop = DrainLoop(prober=prober, interval_seconds=3600)
task = asyncio.create_task(loop.start())
await asyncio.sleep(0.02)
await loop.stop()
await asyncio.wait_for(task, timeout=2) # must not hang
async def test_drain_errors_do_not_kill_the_loop(self):
from app.validator.retry_queue import MemoryRetryStore, RetryQueue
calls = {"n": 0}
async def failing_prober(email, mx_host):
calls["n"] += 1
raise RuntimeError("mx down")
queue = RetryQueue(store=MemoryRetryStore(), backoff_seconds=0,
max_attempts=99)
await queue.enqueue("a@corp.test", "mx.corp.test", code=451)
loop = DrainLoop(prober=failing_prober, interval_seconds=0.05, queue=queue)
task = asyncio.create_task(loop.start())
await asyncio.sleep(0.2)
await loop.stop()
await task
self.assertGreaterEqual(calls["n"], 1, "loop must survive prober errors")
async def test_drain_uses_the_shared_queue(self):
from app.validator.retry_queue import retry_queue as shared_queue
prober = AsyncMock(return_value=SMTPResult(verified=True, smtp_code=250))
loop = DrainLoop(prober=prober, interval_seconds=3600)
self.assertIs(loop.queue, shared_queue)
if __name__ == "__main__":
unittest.main() |