| import unittest | |
| from unittest.mock import AsyncMock, patch | |
| from app.models import ( | |
| DNSResult, | |
| DisposableResult, | |
| RiskLevel, | |
| ScoringResult, | |
| SMTPResult, | |
| SyntaxResult, | |
| ValidationStatus, | |
| ) | |
| from app.validator.engine import EmailValidationEngine, _mask_email_for_log | |
| class EngineDNSFailureTests(unittest.IsolatedAsyncioTestCase): | |
| def test_status_thresholds_match_risk_score_bands(self): | |
| engine = EmailValidationEngine() | |
| self.assertEqual( | |
| engine._determine_status( | |
| ScoringResult(score=40, risk_level=RiskLevel.MEDIUM) | |
| ), | |
| ValidationStatus.RISKY, | |
| ) | |
| self.assertEqual( | |
| engine._determine_status( | |
| ScoringResult(score=39, risk_level=RiskLevel.HIGH) | |
| ), | |
| ValidationStatus.INVALID, | |
| ) | |
| def test_logs_mask_local_part(self): | |
| self.assertEqual( | |
| _mask_email_for_log("person@example.test"), "p***@example.test" | |
| ) | |
| self.assertEqual(_mask_email_for_log("not-an-address"), "[invalid-email]") | |
| async def test_transient_dns_failure_is_risky_without_smtp(self): | |
| dns_result = DNSResult( | |
| domain_exists=False, | |
| has_mx_records=False, | |
| error="DNS timeout", | |
| error_code="timeout", | |
| ) | |
| engine = EmailValidationEngine() | |
| with ( | |
| patch( | |
| "app.validator.engine.validate_syntax", | |
| return_value=SyntaxResult( | |
| passed=True, | |
| normalized_email="person@example.test", | |
| domain="example.test", | |
| local_part="person", | |
| ), | |
| ), | |
| patch( | |
| "app.validator.engine.check_dns", new=AsyncMock(return_value=dns_result) | |
| ), | |
| patch.object(engine, "use_proxies", False), | |
| patch( | |
| "app.validator.mailbox_provider.smtp_verifier.verify", new=AsyncMock() | |
| ) as smtp_verify, | |
| ): | |
| result = await engine.validate("person@example.test", deep=True) | |
| self.assertEqual(result.status, ValidationStatus.RISKY) | |
| self.assertEqual(result.dns.error_code, "timeout") | |
| self.assertEqual(result.scoring.score, 50) | |
| self.assertEqual(result.scoring.risk_level.value, "medium") | |
| self.assertEqual(result.scoring.breakdown["DNS_TRANSIENT_FAILURE"], -50) | |
| smtp_verify.assert_not_awaited() | |
| async def test_a_record_fallback_is_risky_even_if_smtp_accepts(self): | |
| dns_result = DNSResult( | |
| domain_exists=True, | |
| has_mx_records=False, | |
| mx_records=[{"host": "example.test", "priority": 0}], | |
| a_fallback=True, | |
| ) | |
| class FakeMailbox: | |
| async def verify(self, email, mx_host): | |
| return SMTPResult(verified=True) | |
| async def catch_all(self, domain, mx_host): | |
| return (False, 0.9) | |
| engine = EmailValidationEngine(mailbox=FakeMailbox()) | |
| with ( | |
| patch( | |
| "app.validator.engine.validate_syntax", | |
| return_value=SyntaxResult( | |
| passed=True, | |
| normalized_email="person@example.test", | |
| domain="example.test", | |
| local_part="person", | |
| ), | |
| ), | |
| patch( | |
| "app.validator.engine.check_dns", new=AsyncMock(return_value=dns_result) | |
| ), | |
| patch("app.validator.engine.check_disposable", return_value=None), | |
| ): | |
| result = await engine.validate("person@example.test", deep=True) | |
| self.assertEqual(result.status, ValidationStatus.RISKY) | |
| self.assertTrue(result.dns.a_fallback) | |
| async def test_fast_mode_does_not_claim_mailbox_validity(self): | |
| engine = EmailValidationEngine() | |
| with ( | |
| patch( | |
| "app.validator.engine.validate_syntax", | |
| return_value=SyntaxResult( | |
| passed=True, | |
| normalized_email="person@example.test", | |
| domain="example.test", | |
| local_part="person", | |
| ), | |
| ), | |
| 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 engine.validate("person@example.test", deep=False) | |
| self.assertEqual(result.status, ValidationStatus.RISKY) | |
| self.assertIsNone(result.smtp) | |
| self.assertEqual(result.scoring.score, 70) | |
| self.assertEqual(result.scoring.risk_level.value, "medium") | |
| self.assertEqual( | |
| result.scoring.breakdown["FAST_MODE_SKIPPED_MAILBOX_CHECK"], -30 | |
| ) | |
| self.assertIn("Mailbox verification was not performed", result.scoring.warnings) | |
| async def test_domain_without_mail_routes_is_invalid(self): | |
| dns_result = DNSResult( | |
| domain_exists=True, | |
| has_mx_records=False, | |
| error="Domain has no MX, A, or AAAA records", | |
| error_code="no_mail_records", | |
| ) | |
| engine = EmailValidationEngine() | |
| with ( | |
| patch( | |
| "app.validator.engine.validate_syntax", | |
| return_value=SyntaxResult( | |
| passed=True, | |
| normalized_email="person@example.test", | |
| domain="example.test", | |
| local_part="person", | |
| ), | |
| ), | |
| patch( | |
| "app.validator.engine.check_dns", new=AsyncMock(return_value=dns_result) | |
| ), | |
| patch( | |
| "app.validator.mailbox_provider.smtp_verifier.verify", new=AsyncMock() | |
| ) as smtp_verify, | |
| ): | |
| result = await engine.validate("person@example.test", deep=True) | |
| self.assertEqual(result.status, ValidationStatus.INVALID) | |
| self.assertEqual(result.scoring.score, 0) | |
| smtp_verify.assert_not_awaited() | |
| if __name__ == "__main__": | |
| unittest.main() | |