"""Autonomous-engine engine-level tests (spec Phase 3 gate + Phase 5 verdict). Verifies the deterministic-first routing, the normalized verdict fields on both single and batch paths, the EAI deterministic refusal, and the catch-all-never-valid invariant. """ import unittest from unittest.mock import AsyncMock, patch from app.config import settings from app.models import DNSResult, DisposableResult, SMTPResult, SyntaxResult from app.validator.engine import ( EmailValidationEngine, _deterministic_verdict, _smtp_class, ) GOOD_SYNTAX = SyntaxResult( passed=True, normalized_email="john@example.test", domain="example.test", local_part="john", ) GOOD_DNS = DNSResult( domain_exists=True, has_mx_records=True, mx_records=[{"host": "mx.example.test", "priority": 10}], ) GOOD_DISPOSABLE = DisposableResult( is_disposable=False, is_free_provider=False, is_role_based=False, ) class FakeScorer: """Records feature strings; returns a fixed score.""" def __init__(self, score=0.2, label="low_risk", fallback_used=False): self.fixed_score = score self.label = label self.fallback_used = fallback_used self.calls = [] async def score(self, features): self.calls.append(features) from app.ml.risk_scorer import AIScore return AIScore(score=self.fixed_score, label=self.label, source="heuristic" if self.fallback_used else "inference", fallback_used=self.fallback_used) def _build_engine(mailbox, scorer): from app.validator.validation_cache import ValidationCache async def _no_get(key): return None async def _no_set(key, value, ttl=None): return None eng = EmailValidationEngine( mailbox=mailbox, cache=ValidationCache(get=_no_get, set=_no_set), scorer=scorer, ) return eng def _patch_pipeline(syntax=GOOD_SYNTAX, dns=None, disposable=GOOD_DISPOSABLE): dns = dns or GOOD_DNS return ( 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), ) class DeterministicVerdictUnitTests(unittest.TestCase): def test_explicit_5xx_outranks_catch_all(self): smtp = SMTPResult(verified=False, smtp_code=550, smtp_message="no such user") verdict = _deterministic_verdict(None, None, smtp) self.assertEqual(verdict, "invalid") def test_catch_all_never_valid(self): from app.models import CatchAllResult ca = CatchAllResult(is_catch_all=True, confidence=0.95) smtp = SMTPResult(verified=True, smtp_code=250) self.assertEqual(_deterministic_verdict(None, ca, smtp), "catch_all") def test_definitive_accept_is_valid(self): smtp = SMTPResult(verified=True, smtp_code=250) self.assertEqual(_deterministic_verdict(None, None, smtp), "valid") def test_disposable_outranks_everything(self): from app.models import DisposableResult d = DisposableResult(is_disposable=True, is_free_provider=False, is_role_based=False) smtp = SMTPResult(verified=True, smtp_code=250) self.assertEqual(_deterministic_verdict(d, None, smtp), "disposable") def test_inconclusive_is_unknown(self): smtp = SMTPResult(verified=None, error="smtp_unreachable: egress blocked") self.assertEqual(_deterministic_verdict(None, None, smtp), "unknown") def test_smtp_class_mapping(self): self.assertEqual(_smtp_class(SMTPResult(verified=True, smtp_code=250)), "accept") self.assertEqual(_smtp_class(SMTPResult(verified=False, smtp_code=550)), "5xx") self.assertEqual(_smtp_class(SMTPResult(verified=False, smtp_code=451)), "4xx") self.assertEqual( _smtp_class(SMTPResult(verified=None, error="smtp_unreachable: blocked")), "transport_error", ) self.assertEqual(_smtp_class(None), "skipped") class SinglePathTests(unittest.IsolatedAsyncioTestCase): async def test_ambiguous_rows_reach_scorer_and_get_fields(self): mailbox_accepts = AsyncMock() mailbox_accepts.name = "fake" mailbox_accepts.verify = AsyncMock( return_value=SMTPResult(verified=None, error="smtp_unreachable: egress") ) mailbox_accepts.catch_all = AsyncMock(return_value=(True, 0.9)) scorer = FakeScorer(score=0.75, label="high_risk", fallback_used=True) eng = _build_engine(mailbox_accepts, scorer) patches = _patch_pipeline() with patches[0], patches[1], patches[2]: result = await eng.validate("john@example.test", deep=True) self.assertEqual(result.deterministic_verdict, "catch_all") self.assertEqual(result.ai_risk_score, 0.75) self.assertEqual(result.ai_label, "high_risk") self.assertTrue(result.fallback_used) # catch-all NEVER collapses into valid regardless of the AI score. self.assertNotEqual(result.status.value, "valid") self.assertIn("ai_high_risk", result.scoring.warnings) self.assertEqual(len(scorer.calls), 1) async def test_deterministic_rows_never_reach_scorer(self): mailbox = AsyncMock() mailbox.name = "fake" mailbox.verify = AsyncMock( return_value=SMTPResult(verified=False, smtp_code=550, smtp_message="550 no such user") ) mailbox.catch_all = AsyncMock(return_value=(False, 0.9)) scorer = FakeScorer() eng = _build_engine(mailbox, scorer) patches = _patch_pipeline() with patches[0], patches[1], patches[2]: result = await eng.validate("john@example.test", deep=True) self.assertEqual(result.deterministic_verdict, "invalid") self.assertEqual(scorer.calls, []) self.assertIsNone(result.ai_risk_score) self.assertFalse(result.fallback_used) async def test_eai_address_refused_deterministically(self): unicode_syntax = SyntaxResult( passed=True, normalized_email="用户@example.test", domain="example.test", local_part="用户", ) mailbox = AsyncMock() mailbox.name = "fake" mailbox.verify = AsyncMock( side_effect=AssertionError("must not probe an EAI address") ) mailbox.catch_all = AsyncMock(return_value=(None, 0.0)) scorer = FakeScorer() eng = _build_engine(mailbox, scorer) patches = _patch_pipeline(syntax=unicode_syntax) with patches[0], patches[1], patches[2]: result = await eng.validate("用户@example.test", deep=True) self.assertEqual(result.reason_code, "unsupported_eai") mailbox.verify.assert_not_awaited() # The domain-level catch-all probe still ran (independent of local part). mailbox.catch_all.assert_awaited_once() async def test_dns_inconsistent_adds_warning(self): bad_dns = DNSResult( domain_exists=True, has_mx_records=True, mx_records=[{"host": "mx.example.test", "priority": 10}], dns_inconsistent=True, mx_source="doh:dual", ) mailbox = AsyncMock() mailbox.name = "fake" mailbox.verify = AsyncMock( return_value=SMTPResult(verified=None, error="smtp_unreachable: egress") ) mailbox.catch_all = AsyncMock(return_value=(True, 0.9)) scorer = FakeScorer(score=0.3, label="low_risk") eng = _build_engine(mailbox, scorer) patches = _patch_pipeline(dns=bad_dns) with patches[0], patches[1], patches[2]: result = await eng.validate("john@example.test", deep=True) self.assertIn("dns_inconsistent: resolvers disagree on MX", result.scoring.warnings) # The inconsistency is visible to the scorer feature string. self.assertIn("dns_consistent=no", scorer.calls[0]) async def test_fast_path_has_no_deterministic_verdict(self): scorer = FakeScorer() eng = _build_engine(AsyncMock(), scorer) patches = _patch_pipeline() with patches[0], patches[1], patches[2]: result = await eng.validate("john@example.test", deep=False) self.assertIsNone(result.deterministic_verdict) self.assertEqual(scorer.calls, []) class BatchPathTests(unittest.IsolatedAsyncioTestCase): async def test_batch_rows_carry_normalized_fields(self): class BatchMailbox: name = "fake-batch" async def verify_many(self, pairs): results = { e: SMTPResult(verified=None, error="smtp_unreachable: egress") for e, _ in pairs } return results, {"example.test": True} scorer = FakeScorer(score=0.55, label="medium_risk", fallback_used=True) eng = _build_engine(BatchMailbox(), scorer) patches = _patch_pipeline() with patches[0], patches[1], patches[2]: results = await eng.validate_batch( ["john@example.test", "jane@example.test"], deep=True ) self.assertEqual(len(results), 2) for row in results: self.assertEqual(row.deterministic_verdict, "catch_all") self.assertEqual(row.ai_label, "medium_risk") self.assertEqual(row.ai_risk_score, 0.55) self.assertTrue(row.fallback_used) self.assertFalse(row.cache_hit) # Feature dedupe: identical rows share one scorer call. self.assertEqual(len(scorer.calls), 1) async def test_batch_eai_refused_without_probe(self): class BatchMailbox: name = "fake-batch" async def verify_many(self, pairs): # Fires only if the EAI row wrongly reaches the mailbox stage. assert all( all(ord(c) < 128 for c in e.rsplit("@", 1)[0]) for e, _ in pairs ), "EAI rows must not reach the mailbox stage" results = { e: SMTPResult(verified=None, error="smtp_unreachable: egress") for e, _ in pairs } return results, {"example.test": True} ascii_syntax = GOOD_SYNTAX unicode_syntax = SyntaxResult( passed=True, normalized_email="用户@example.test", domain="example.test", local_part="用户", ) def syntax_side(email): return unicode_syntax if "用户" in email else ascii_syntax scorer = FakeScorer(score=0.5, label="medium_risk") eng = _build_engine(BatchMailbox(), scorer) syntax_patch, dns_patch, disposable_patch = _patch_pipeline() with patch("app.validator.engine.validate_syntax", side_effect=syntax_side), \ dns_patch, disposable_patch: results = await eng.validate_batch( ["john@example.test", "用户@example.test"], deep=True ) by_email = {r.email: r for r in results} self.assertEqual(by_email["用户@example.test"].reason_code, "unsupported_eai") self.assertEqual(by_email["john@example.test"].deterministic_verdict, "catch_all") if __name__ == "__main__": unittest.main()