FortyDN Dev
v3.0.0: fully autonomous engine — remove MillionVerifier, add DoH DNS worker, AI risk scorer (HF PRO, circuit breaker), i18n, normalized verdicts, remote SMTP worker
de745bf | """Mailbox-layer tests after the v3.0.0 autonomy refactor. | |
| The MillionVerifier delegation was removed BY DESIGN: the engine must run with | |
| no vendor account, key, or per-check fee, and a stray key must never be able | |
| to re-route the recipient list through a third party. These tests pin that | |
| decision: no MV symbol is importable, no settings knob re-enables it, and the | |
| explicit 5xx mailbox-rejection path (previously exercised through the MV | |
| adapter) still hard-fails to score 0 through a plain provider. | |
| """ | |
| import unittest | |
| from unittest.mock import AsyncMock, patch | |
| from app.config import settings | |
| from app.models import SMTPResult | |
| from app.validator import mailbox_provider as mp | |
| from app.validator.mailbox_provider import ( | |
| MailboxProvider, | |
| LocalSmtpProvider, | |
| AutonomousSmtpProvider, | |
| RemoteSmtpWorkerProvider, | |
| resolve_provider, | |
| ) | |
| class AutonomyInvariantTests(unittest.TestCase): | |
| """The engine is autonomous: no third-party verification API exists.""" | |
| def test_millionverifier_is_gone(self): | |
| self.assertFalse(hasattr(mp, "MillionVerifierProvider")) | |
| self.assertFalse(hasattr(mp, "map_mv_result")) | |
| def test_config_has_no_mv_knobs(self): | |
| from app.config import Settings | |
| for field in ("mv_api_key", "mv_timeout", "prefer_api_provider"): | |
| self.assertFalse( | |
| hasattr(settings, field), | |
| f"settings.{field} must not exist after the autonomy refactor", | |
| ) | |
| self.assertFalse( | |
| field in Settings.model_fields, | |
| f"Settings.{field} must not be declared", | |
| ) | |
| def test_api_mode_is_no_longer_a_valid_choice(self): | |
| # Any leftover "api" configuration degrades to the built-in engine | |
| # instead of raising or silently selecting a vendor. | |
| with patch.object(settings, "mailbox_provider", "api"), patch.object( | |
| settings, "smtp_worker_url", "" | |
| ): | |
| provider = resolve_provider() | |
| self.assertIsInstance(provider, (AutonomousSmtpProvider, RemoteSmtpWorkerProvider)) | |
| class ResolveProviderTests(unittest.TestCase): | |
| def test_local_forced_is_autonomous(self): | |
| with patch.object(settings, "mailbox_provider", "local"), patch.object( | |
| settings, "smtp_worker_url", "https://worker.example.com" | |
| ): | |
| self.assertIsInstance(resolve_provider(), AutonomousSmtpProvider) | |
| def test_auto_without_worker_is_autonomous(self): | |
| with patch.object(settings, "mailbox_provider", "auto"), patch.object( | |
| settings, "smtp_worker_url", "" | |
| ): | |
| self.assertIsInstance(resolve_provider(), AutonomousSmtpProvider) | |
| def test_auto_with_worker_resolves_remote(self): | |
| with patch.object(settings, "mailbox_provider", "auto"), patch.object( | |
| settings, "smtp_worker_url", "https://worker.example.com" | |
| ): | |
| self.assertIsInstance(resolve_provider(), RemoteSmtpWorkerProvider) | |
| def test_remote_forced_without_url_degrades_to_autonomous(self): | |
| with patch.object(settings, "mailbox_provider", "remote"), patch.object( | |
| settings, "smtp_worker_url", "" | |
| ): | |
| self.assertIsInstance(resolve_provider(), AutonomousSmtpProvider) | |
| def test_providers_satisfy_protocol(self): | |
| for cls in (AutonomousSmtpProvider, RemoteSmtpWorkerProvider, LocalSmtpProvider): | |
| self.assertTrue(hasattr(cls, "verify")) | |
| self.assertTrue(hasattr(cls, "catch_all")) | |
| self.assertTrue(issubclass(cls, MailboxProvider)) | |
| class EngineProviderVerdictTests(unittest.IsolatedAsyncioTestCase): | |
| """End-to-end: an explicit 5xx mailbox rejection must hard-fail to score 0 | |
| (Invalid) through the whole engine — the invariant the MV adapter used to | |
| guarantee, now pinned against a plain provider.""" | |
| async def test_explicit_rejection_scores_zero(self): | |
| from app.models import DNSResult, DisposableResult, SyntaxResult | |
| from app.validator.engine import EmailValidationEngine | |
| class DeadMailboxProvider: | |
| async def verify(self, email, mx_host): | |
| return SMTPResult( | |
| verified=False, | |
| smtp_code=550, | |
| smtp_message="550 mailbox does not exist", | |
| ) | |
| async def catch_all(self, domain, mx_host): | |
| return (False, 0.9) | |
| eng = EmailValidationEngine(mailbox=DeadMailboxProvider()) | |
| with ( | |
| patch("app.validator.engine.validate_syntax", return_value=SyntaxResult( | |
| passed=True, normalized_email="a@example.test", | |
| domain="example.test", local_part="a", | |
| )), | |
| patch("app.validator.engine.check_dns", new=AsyncMock(return_value=DNSResult( | |
| domain_exists=True, has_mx_records=True, | |
| mx_records=[{"host": "mx.example.test", "priority": 10}], | |
| ))), | |
| patch("app.validator.engine.check_disposable", return_value=DisposableResult( | |
| is_disposable=False, is_free_provider=False, is_role_based=False, | |
| )), | |
| ): | |
| result = await eng.validate("a@example.test", deep=True) | |
| self.assertEqual(result.scoring.score, 0) | |
| self.assertEqual(result.status.value, "invalid") | |
| self.assertEqual(result.reason_code, "rcpt_5xx") | |
| # The normalized verdict agrees with the deterministic gauntlet. | |
| self.assertEqual(result.deterministic_verdict, "invalid") | |
| class RemoteSmtpWorkerProviderTests(unittest.IsolatedAsyncioTestCase): | |
| """The remote worker path (spec §3.3): orchestrator delegates port-25.""" | |
| def _provider(self): | |
| return RemoteSmtpWorkerProvider( | |
| url="https://worker.example.com", token="s3cret", timeout=5, | |
| fail_threshold=2, cooldown_seconds=60, | |
| ) | |
| async def test_verify_posts_to_worker(self): | |
| p = self._provider() | |
| payload = {"verified": True, "smtp_code": 250, "smtp_message": "ok"} | |
| with patch("app.validator.mailbox_provider.httpx.AsyncClient") as client_cls: | |
| resp = AsyncMock() | |
| resp.raise_for_status = lambda: None | |
| resp.json = lambda: payload | |
| client = AsyncMock() | |
| client.post = AsyncMock(return_value=resp) | |
| client_cls.return_value.__aenter__ = AsyncMock(return_value=client) | |
| client_cls.return_value.__aexit__ = AsyncMock(return_value=False) | |
| result = await p.verify("a@corp.test", "mx.corp.test") | |
| self.assertTrue(result.verified) | |
| args, kwargs = client.post.call_args | |
| self.assertEqual(args[0], "https://worker.example.com/smtp/probe") | |
| self.assertEqual(kwargs["headers"]["X-Worker-Token"], "s3cret") | |
| async def test_worker_unreachable_is_inconclusive_and_opens_circuit(self): | |
| p = self._provider() | |
| with patch("app.validator.mailbox_provider.httpx.AsyncClient") as client_cls: | |
| client_cls.return_value.__aenter__ = AsyncMock( | |
| side_effect=RuntimeError("boom") | |
| ) | |
| client_cls.return_value.__aexit__ = AsyncMock(return_value=False) | |
| r1 = await p.verify("a@corp.test", "mx.corp.test") | |
| r2 = await p.verify("b@corp.test", "mx.corp.test") | |
| self.assertIsNone(r1.verified) | |
| self.assertIn("unreachable", (r1.error or "")) | |
| # Threshold 2 reached → circuit open, next call fails fast. | |
| self.assertTrue(p.circuit_open) | |
| r3 = await p.verify("c@corp.test", "mx.corp.test") | |
| self.assertIn("circuit open", (r3.error or "").lower()) | |
| async def test_catch_all_roundtrip(self): | |
| p = self._provider() | |
| payload = {"is_catch_all": True, "confidence": 0.95} | |
| with patch("app.validator.mailbox_provider.httpx.AsyncClient") as client_cls: | |
| resp = AsyncMock() | |
| resp.raise_for_status = lambda: None | |
| resp.json = lambda: payload | |
| client = AsyncMock() | |
| client.post = AsyncMock(return_value=resp) | |
| client_cls.return_value.__aenter__ = AsyncMock(return_value=client) | |
| client_cls.return_value.__aexit__ = AsyncMock(return_value=False) | |
| is_ca, confidence = await p.catch_all("corp.test", "mx.corp.test") | |
| self.assertTrue(is_ca) | |
| self.assertGreater(confidence, 0.5) | |
| async def test_verify_many_batch_roundtrip(self): | |
| p = self._provider() | |
| payload = { | |
| "results": { | |
| "a@corp.test": {"verified": True, "smtp_code": 250}, | |
| "b@corp.test": {"verified": False, "smtp_code": 550}, | |
| }, | |
| "catch_all": {"corp.test": False}, | |
| } | |
| with patch("app.validator.mailbox_provider.httpx.AsyncClient") as client_cls: | |
| resp = AsyncMock() | |
| resp.raise_for_status = lambda: None | |
| resp.json = lambda: payload | |
| client = AsyncMock() | |
| client.post = AsyncMock(return_value=resp) | |
| client_cls.return_value.__aenter__ = AsyncMock(return_value=client) | |
| client_cls.return_value.__aexit__ = AsyncMock(return_value=False) | |
| results, catch_all = await p.verify_many( | |
| [("a@corp.test", "mx"), ("b@corp.test", "mx")] | |
| ) | |
| self.assertTrue(results["a@corp.test"].verified) | |
| self.assertFalse(results["b@corp.test"].verified) | |
| self.assertEqual(catch_all, {"corp.test": False}) | |
| class CircuitBreakerTests(unittest.TestCase): | |
| """Blocked egress must fail fast instead of burning smtp_timeout per row.""" | |
| def test_local_provider_opens_circuit_after_threshold(self): | |
| p = LocalSmtpProvider(fail_threshold=2, cooldown_seconds=60) | |
| p.record_failure() | |
| self.assertFalse(p.circuit_open) | |
| p.record_failure() | |
| self.assertTrue(p.circuit_open) | |
| def test_open_circuit_returns_immediately_unavailable(self): | |
| import asyncio | |
| p = LocalSmtpProvider(fail_threshold=1, cooldown_seconds=60) | |
| p.record_failure() | |
| result = asyncio.run(p.verify("a@example.com", "mx.example.com")) | |
| self.assertIsNone(result.verified) | |
| self.assertIn("unavailable", (result.error or "").lower()) | |
| def test_circuit_closes_after_cooldown(self): | |
| import time as _t | |
| p = LocalSmtpProvider(fail_threshold=1, cooldown_seconds=60) | |
| p.record_failure() | |
| self.assertTrue(p.circuit_open) | |
| # Simulate the cooldown elapsing: the open-window is now in the past. | |
| p._open_until = _t.monotonic() - 1 | |
| self.assertFalse(p.circuit_open) | |
| self.assertFalse(p.circuit_open_after_cooldown()) | |
| def test_zero_cooldown_never_stays_open(self): | |
| # cooldown_seconds=0 opens and closes in the same instant — there is no | |
| # window, so it must not read as open on the next call. | |
| p = LocalSmtpProvider(fail_threshold=1, cooldown_seconds=0) | |
| p.record_failure() | |
| self.assertFalse(p.circuit_open) | |
| if __name__ == "__main__": | |
| unittest.main() | |