| 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) |
|
|
| 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() |
