File size: 3,662 Bytes
de745bf | 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 | 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()
|