FortyDN Dev
v3.0.0: fully autonomous engine — remove MillionVerifier, add DoH DNS worker, AI risk scorer (HF PRO, circuit breaker), i18n, normalized verdicts, remote SMTP worker
de745bf | import unittest | |
| from unittest.mock import patch | |
| import dns.exception | |
| import dns.resolver | |
| from app.config import settings | |
| from app.validator.dns_check import check_dns | |
| def disable_doh(): | |
| """These tests exercise the LEGACY dnspython path — keep them hermetic by | |
| turning the DoH layer off (its own behaviour is covered by | |
| test_doh_resolver.py).""" | |
| return patch.object(settings, "doh_enabled", False) | |
| class ResolverStub: | |
| timeout = 0 | |
| lifetime = 0 | |
| def __init__(self, responses): | |
| self.responses = responses | |
| async def resolve(self, name, record_type): | |
| response = self.responses.get((name, record_type)) | |
| if isinstance(response, Exception): | |
| raise response | |
| if response is None: | |
| raise dns.resolver.NoAnswer() | |
| return response | |
| class DNSCheckTests(unittest.IsolatedAsyncioTestCase): | |
| async def test_timeout_is_reported_as_transient(self): | |
| resolver = ResolverStub({("example.test", "MX"): dns.exception.Timeout()}) | |
| with disable_doh(), patch( | |
| "app.validator.dns_check.dns.asyncresolver.Resolver", return_value=resolver | |
| ): | |
| result = await check_dns("example.test") | |
| self.assertEqual(result.error_code, "timeout") | |
| self.assertFalse(result.domain_exists) | |
| async def test_no_nameservers_is_reported_as_transient(self): | |
| resolver = ResolverStub( | |
| {("unavailable.test", "MX"): dns.resolver.NoNameservers()} | |
| ) | |
| with disable_doh(), patch( | |
| "app.validator.dns_check.dns.asyncresolver.Resolver", return_value=resolver | |
| ): | |
| result = await check_dns("unavailable.test") | |
| self.assertEqual(result.error_code, "resolver_error") | |
| async def test_nxdomain_is_reported_as_definitive(self): | |
| resolver = ResolverStub({("missing.test", "MX"): dns.resolver.NXDOMAIN()}) | |
| with disable_doh(), patch( | |
| "app.validator.dns_check.dns.asyncresolver.Resolver", return_value=resolver | |
| ): | |
| result = await check_dns("missing.test") | |
| self.assertEqual(result.error_code, "nxdomain") | |
| self.assertFalse(result.domain_exists) | |
| async def test_a_record_fallback_is_preserved(self): | |
| resolver = ResolverStub( | |
| { | |
| ("mail.example.test", "MX"): dns.resolver.NoAnswer(), | |
| ("mail.example.test", "A"): ["192.0.2.10"], | |
| } | |
| ) | |
| with disable_doh(), patch( | |
| "app.validator.dns_check.dns.asyncresolver.Resolver", return_value=resolver | |
| ): | |
| result = await check_dns("mail.example.test") | |
| self.assertTrue(result.domain_exists) | |
| self.assertFalse(result.has_mx_records) | |
| self.assertTrue(result.a_fallback) | |
| self.assertEqual(result.mx_records[0]["host"], "mail.example.test") | |
| async def test_no_mx_or_address_is_distinct_from_transient_dns_error(self): | |
| resolver = ResolverStub( | |
| { | |
| ("no-mail.example.test", "MX"): dns.resolver.NoAnswer(), | |
| ("no-mail.example.test", "A"): dns.resolver.NoAnswer(), | |
| ("no-mail.example.test", "AAAA"): dns.resolver.NoAnswer(), | |
| } | |
| ) | |
| with disable_doh(), patch( | |
| "app.validator.dns_check.dns.asyncresolver.Resolver", return_value=resolver | |
| ): | |
| result = await check_dns("no-mail.example.test") | |
| self.assertEqual(result.error_code, "no_mail_records") | |
| self.assertTrue(result.domain_exists) | |
| self.assertFalse(result.has_mx_records) | |
| self.assertFalse(result.a_fallback) | |
| if __name__ == "__main__": | |
| unittest.main() | |