File size: 6,661 Bytes
c94194a | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 | 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() |