Spaces:

sitcod3
/

modul-x

Running

modul-x / tests /test_validation_cache.py

Upload folder using huggingface_hub

c94194a verified 11 days ago

6.66 kB

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()

	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()