Z User
v3.1.0: Part B architecture migration — eliminate Hugging Face inference credit dependency
6c516b6 | """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() | |