import unittest from app.models import DNSResult, SMTPResult from app.validator.validation_cache import ValidationCache class FakeBackend: """Dict-backed stand-in for the Redis/memory cache.""" def __init__(self): self.store: dict[str, str] = {} self.puts = 0 async def get(self, key): return self.store.get(key) async def set(self, key, value, ttl=0): self.puts += 1 self.store[key] = value def make_cache(): backend = FakeBackend() return ValidationCache(get=backend.get, set=backend.set, ttl_seconds=3600), backend class DomainDnsCacheTests(unittest.IsolatedAsyncioTestCase): async def test_definitive_dns_result_round_trips(self): cache, _ = make_cache() dns = DNSResult( domain_exists=True, has_mx_records=True, mx_records=[{"host": "mx.google.com", "priority": 5}], has_spf=True, spf_record="v=spf1 include:_spf.google.com ~all", has_dmarc=True, ) await cache.put_dns("gmail.com", dns) got = await cache.get_dns("gmail.com") self.assertIsNotNone(got) self.assertTrue(got.domain_exists) self.assertTrue(got.has_mx_records) self.assertEqual(got.mx_records[0]["host"], "mx.google.com") self.assertTrue(got.has_spf) async def test_keys_are_case_insensitive(self): cache, _ = make_cache() await cache.put_dns("Gmail.COM", DNSResult(domain_exists=True, has_mx_records=True)) self.assertIsNotNone(await cache.get_dns("gmail.com")) async def test_dns_timeout_is_never_cached(self): # A transient resolver failure must not be remembered as a verdict — # otherwise one timeout would mark a domain dead for a full day. cache, backend = make_cache() await cache.put_dns( "flaky.test", DNSResult(domain_exists=False, has_mx_records=False, error_code="timeout"), ) self.assertIsNone(await cache.get_dns("flaky.test")) self.assertEqual(backend.puts, 0) async def test_resolver_error_is_never_cached(self): cache, _ = make_cache() await cache.put_dns( "flaky.test", DNSResult(domain_exists=False, has_mx_records=False, error_code="resolver_error"), ) self.assertIsNone(await cache.get_dns("flaky.test")) async def test_missing_key_returns_none(self): cache, _ = make_cache() self.assertIsNone(await cache.get_dns("never-seen.test")) class AddressVerdictCacheTests(unittest.IsolatedAsyncioTestCase): async def test_verified_mailbox_is_cached(self): cache, _ = make_cache() await cache.put_verdict( "a@gmail.com", deep=True, smtp=SMTPResult(verified=True, smtp_code=250), catch_all=False, ) got = await cache.get_verdict("a@gmail.com", deep=True) self.assertIsNotNone(got) smtp, catch_all = got self.assertTrue(smtp.verified) self.assertIs(catch_all, False) async def test_rejected_mailbox_is_cached(self): cache, _ = make_cache() await cache.put_verdict( "ghost@gmail.com", deep=True, smtp=SMTPResult(verified=False, smtp_code=550, smtp_message="550 no such user"), catch_all=False, ) smtp, _ = await cache.get_verdict("ghost@gmail.com", deep=True) self.assertFalse(smtp.verified) self.assertEqual(smtp.smtp_code, 550) async def test_inconclusive_verdict_is_never_cached(self): # verified=None means "we could not tell" — caching it would freeze an # unknown for 24h and suppress a later successful probe. cache, backend = make_cache() await cache.put_verdict( "a@gmail.com", deep=True, smtp=SMTPResult(verified=None, error="SMTP verification timed out"), catch_all=None, ) self.assertIsNone(await cache.get_verdict("a@gmail.com", deep=True)) self.assertEqual(backend.puts, 0) async def test_catch_all_flag_is_preserved(self): cache, _ = make_cache() await cache.put_verdict( "x@catchall.test", deep=True, smtp=SMTPResult(verified=None, error="catch_all"), catch_all=True, ) got = await cache.get_verdict("x@catchall.test", deep=True) # A catch-all row IS a definitive domain fact, so it is cacheable. self.assertIsNotNone(got) self.assertIs(got[1], True) async def test_deep_and_fast_verdicts_do_not_collide(self): # The UI flags unsaved demo runs by `deep`; a fast-mode (no mailbox # proof) verdict must never be served to a deep request. cache, _ = make_cache() await cache.put_verdict( "a@gmail.com", deep=False, smtp=SMTPResult(verified=None, error="skipped"), catch_all=None, ) self.assertIsNone(await cache.get_verdict("a@gmail.com", deep=True)) async def test_emails_are_case_insensitive(self): cache, _ = make_cache() await cache.put_verdict( "A@Gmail.com", deep=True, smtp=SMTPResult(verified=True, smtp_code=250), catch_all=False, ) self.assertIsNotNone(await cache.get_verdict("a@gmail.com", deep=True)) class CacheFailureToleranceTests(unittest.IsolatedAsyncioTestCase): """A cache outage must degrade to "no cache", never break validation.""" async def test_get_errors_return_none(self): async def boom(key): raise RuntimeError("redis down") cache = ValidationCache(get=boom, set=boom, ttl_seconds=60) self.assertIsNone(await cache.get_dns("gmail.com")) self.assertIsNone(await cache.get_verdict("a@gmail.com", deep=True)) async def test_set_errors_do_not_raise(self): async def boom(*a, **k): raise RuntimeError("redis down") cache = ValidationCache(get=boom, set=boom, ttl_seconds=60) await cache.put_dns("gmail.com", DNSResult(domain_exists=True, has_mx_records=True)) await cache.put_verdict( "a@gmail.com", deep=True, smtp=SMTPResult(verified=True, smtp_code=250), catch_all=False, ) # must not raise async def test_corrupt_payload_returns_none(self): async def bad_get(key): return "{not json" cache = ValidationCache(get=bad_get, set=bad_get, ttl_seconds=60) self.assertIsNone(await cache.get_dns("gmail.com")) self.assertIsNone(await cache.get_verdict("a@gmail.com", deep=True)) if __name__ == "__main__": unittest.main()