| import unittest |
| from unittest.mock import AsyncMock, patch |
|
|
| from app.models import ( |
| CatchAllResult, |
| DNSResult, |
| DisposableResult, |
| ScoringResult, |
| SMTPResult, |
| SyntaxResult, |
| ValidationStatus, |
| ) |
| from app.config import settings |
| from app.validator.engine import EmailValidationEngine |
|
|
|
|
| def _syntax(**kw): |
| base = dict( |
| passed=True, |
| normalized_email="person@example.test", |
| domain="example.test", |
| local_part="person", |
| ) |
| base.update(kw) |
| return SyntaxResult(**base) |
|
|
|
|
| def _dns(**kw): |
| base = dict( |
| domain_exists=True, |
| has_mx_records=True, |
| mx_records=[{"host": "mx.example.test", "priority": 10}], |
| ) |
| base.update(kw) |
| return DNSResult(**base) |
|
|
|
|
| def _disposable(**kw): |
| base = dict(is_disposable=False, is_free_provider=False, is_role_based=False) |
| base.update(kw) |
| return DisposableResult(**base) |
|
|
|
|
| class FakeMailbox: |
| """Injectable mailbox provider — controls the Layer 5/6 verdicts exactly. |
| |
| Injecting the provider (instead of patching module internals) keeps these |
| tests independent of which concrete provider the engine resolves by default. |
| """ |
|
|
| def __init__(self, smtp=None, catch_all=(False, 0.9), raises=None): |
| self._smtp = smtp if smtp is not None else SMTPResult(verified=True, smtp_code=250) |
| self._catch_all = catch_all |
| self._raises = raises |
|
|
| async def verify(self, email, mx_host): |
| if self._raises: |
| raise self._raises |
| return self._smtp |
|
|
| async def catch_all(self, domain, mx_host): |
| if self._raises: |
| raise self._raises |
| return self._catch_all |
|
|
|
|
| def _engine_with(smtp=None, catch_all=(False, 0.9)): |
| return EmailValidationEngine( |
| mailbox=FakeMailbox(smtp=smtp, catch_all=catch_all) |
| ) |
|
|
|
|
| class EngineMetadataTests(unittest.IsolatedAsyncioTestCase): |
| """The engine must emit the provenance fields the frontend contract expects: |
| checks_run, reason_code, validator_version, run_id (RC5).""" |
|
|
| async def test_deep_result_emits_provenance_fields(self): |
| engine = _engine_with() |
| with ( |
| patch("app.validator.engine.validate_syntax", return_value=_syntax()), |
| patch("app.validator.engine.check_dns", new=AsyncMock(return_value=_dns())), |
| patch("app.validator.engine.check_disposable", return_value=_disposable()), |
| ): |
| result = await engine.validate("person@example.test", deep=True) |
|
|
| self.assertEqual( |
| result.checks_run, |
| ["syntax", "dns", "mx", "disposable", "catch_all", "smtp", "scoring"], |
| ) |
| self.assertEqual(result.validator_version, settings.app_version) |
| self.assertTrue(result.run_id) |
| self.assertGreater(len(result.run_id), 5) |
|
|
| async def test_fast_mode_checks_run_excludes_mailbox_layers(self): |
| engine = _engine_with() |
| with ( |
| patch("app.validator.engine.validate_syntax", return_value=_syntax()), |
| patch("app.validator.engine.check_dns", new=AsyncMock(return_value=_dns())), |
| patch("app.validator.engine.check_disposable", return_value=_disposable()), |
| ): |
| result = await engine.validate("person@example.test", deep=False) |
|
|
| self.assertEqual( |
| result.checks_run, ["syntax", "dns", "mx", "disposable", "scoring"] |
| ) |
| self.assertIsNone(result.smtp) |
| self.assertIsNone(result.catch_all) |
|
|
| async def test_reason_code_dns_timeout(self): |
| engine = _engine_with() |
| with ( |
| patch("app.validator.engine.validate_syntax", return_value=_syntax()), |
| patch( |
| "app.validator.engine.check_dns", |
| new=AsyncMock( |
| return_value=_dns( |
| domain_exists=False, |
| has_mx_records=False, |
| error="DNS timeout", |
| error_code="timeout", |
| ) |
| ), |
| ), |
| ): |
| result = await engine.validate("person@example.test", deep=True) |
|
|
| self.assertEqual(result.reason_code, "dns_timeout") |
|
|
| async def test_reason_code_nxdomain(self): |
| engine = _engine_with() |
| with ( |
| patch("app.validator.engine.validate_syntax", return_value=_syntax()), |
| patch( |
| "app.validator.engine.check_dns", |
| new=AsyncMock( |
| return_value=_dns( |
| domain_exists=False, |
| has_mx_records=False, |
| error="Domain not found", |
| error_code="nxdomain", |
| ) |
| ), |
| ), |
| ): |
| result = await engine.validate("person@example.test", deep=True) |
|
|
| self.assertEqual(result.reason_code, "nxdomain") |
|
|
| async def test_reason_code_catch_all(self): |
| engine = _engine_with( |
| smtp=SMTPResult(verified=True, smtp_code=250), catch_all=(True, 0.95) |
| ) |
| with ( |
| patch("app.validator.engine.validate_syntax", return_value=_syntax()), |
| patch("app.validator.engine.check_dns", new=AsyncMock(return_value=_dns())), |
| patch("app.validator.engine.check_disposable", return_value=_disposable()), |
| ): |
| result = await engine.validate("person@example.test", deep=True) |
|
|
| self.assertEqual(result.reason_code, "catch_all_detected") |
| self.assertEqual(result.status, ValidationStatus.RISKY) |
|
|
| async def test_explicit_5xx_rejection_outranks_catch_all(self): |
| |
| |
| engine = _engine_with( |
| smtp=SMTPResult( |
| verified=False, |
| smtp_code=550, |
| smtp_message="550 mailbox does not exist (provider confirmed)", |
| ), |
| catch_all=(True, 0.95), |
| ) |
| with ( |
| patch("app.validator.engine.validate_syntax", return_value=_syntax()), |
| patch("app.validator.engine.check_dns", new=AsyncMock(return_value=_dns())), |
| patch("app.validator.engine.check_disposable", return_value=_disposable()), |
| ): |
| result = await engine.validate("person@example.test", deep=True) |
|
|
| self.assertEqual(result.reason_code, "rcpt_5xx") |
| self.assertEqual(result.scoring.score, 0) |
| self.assertEqual(result.status, ValidationStatus.INVALID) |
|
|
| async def test_reason_code_smtp_timeout(self): |
| engine = _engine_with( |
| smtp=SMTPResult(verified=None, error="SMTP verification timed out"), |
| catch_all=(None, 0.0), |
| ) |
| with ( |
| patch("app.validator.engine.validate_syntax", return_value=_syntax()), |
| patch("app.validator.engine.check_dns", new=AsyncMock(return_value=_dns())), |
| patch("app.validator.engine.check_disposable", return_value=_disposable()), |
| ): |
| result = await engine.validate("person@example.test", deep=True) |
|
|
| self.assertEqual(result.reason_code, "smtp_timeout") |
|
|
|
|
| class CatchAllUnknownTests(unittest.IsolatedAsyncioTestCase): |
| """An unreachable catch-all probe must read as UNKNOWN (None), never as |
| False ("domain rejects unknown recipients"). RC6.""" |
|
|
| async def test_catch_all_probe_unavailable_is_unknown_not_false(self): |
| engine = _engine_with( |
| smtp=SMTPResult(verified=None, error="SMTP verification timed out"), |
| catch_all=(None, 0.0), |
| ) |
| with ( |
| patch("app.validator.engine.validate_syntax", return_value=_syntax()), |
| patch("app.validator.engine.check_dns", new=AsyncMock(return_value=_dns())), |
| patch("app.validator.engine.check_disposable", return_value=_disposable()), |
| ): |
| result = await engine.validate("person@example.test", deep=True) |
|
|
| self.assertIsNone(result.catch_all.is_catch_all) |
| self.assertEqual(result.catch_all.confidence, 0.0) |
|
|
| async def test_catch_all_exception_is_unknown_not_false(self): |
| engine = EmailValidationEngine( |
| mailbox=FakeMailbox(raises=RuntimeError("port 25 unreachable")) |
| ) |
| with ( |
| patch("app.validator.engine.validate_syntax", return_value=_syntax()), |
| patch("app.validator.engine.check_dns", new=AsyncMock(return_value=_dns())), |
| patch("app.validator.engine.check_disposable", return_value=_disposable()), |
| ): |
| result = await engine.validate("person@example.test", deep=True) |
|
|
| self.assertIsNone(result.catch_all.is_catch_all) |
|
|
|
|
| class ScoreCoherenceTests(unittest.IsolatedAsyncioTestCase): |
| """A perfect score (100/low) must only appear when a mailbox verdict exists.""" |
|
|
| async def test_smtp_verified_can_reach_high_score(self): |
| engine = _engine_with( |
| smtp=SMTPResult(verified=True, smtp_code=250), catch_all=(False, 0.9) |
| ) |
| with ( |
| patch("app.validator.engine.validate_syntax", return_value=_syntax()), |
| patch("app.validator.engine.check_dns", new=AsyncMock(return_value=_dns())), |
| patch( |
| "app.validator.engine.check_disposable", |
| return_value=_disposable(is_free_provider=False), |
| ), |
| ): |
| result = await engine.validate("person@example.test", deep=True) |
|
|
| self.assertGreaterEqual(result.scoring.score, 75) |
| self.assertEqual(result.scoring.risk_level.value, "low") |
| self.assertEqual(result.status, ValidationStatus.VALID) |
|
|
| async def test_unverified_mailbox_never_scores_100(self): |
| engine = _engine_with( |
| smtp=SMTPResult(verified=None, error="SMTP verification timed out"), |
| catch_all=(None, 0.0), |
| ) |
| with ( |
| patch("app.validator.engine.validate_syntax", return_value=_syntax()), |
| patch("app.validator.engine.check_dns", new=AsyncMock(return_value=_dns())), |
| patch("app.validator.engine.check_disposable", return_value=_disposable()), |
| ): |
| result = await engine.validate("person@example.test", deep=True) |
|
|
| self.assertLess(result.scoring.score, 75) |
| self.assertNotEqual(result.scoring.risk_level.value, "low") |
|
|
|
|
| if __name__ == "__main__": |
| unittest.main() |
