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 asyncio | |
| import time | |
| import unittest | |
| from unittest.mock import patch | |
| from app.config import settings | |
| from app.validator import dns_check | |
| def disable_doh(): | |
| """Concurrency tests target the legacy dnspython lookups; keep DoH off so | |
| the MX gate does not hit the network.""" | |
| return patch.object(settings, "doh_enabled", False) | |
| class FakeAnswer: | |
| def __init__(self, text: str): | |
| self.strings = [text.encode()] | |
| class SlowResolver: | |
| """Every query takes 0.1s — 8 sequential queries cost ~0.8s.""" | |
| timeout = 5 | |
| lifetime = 5 | |
| async def resolve(self, domain, rtype): | |
| await asyncio.sleep(0.1) | |
| if rtype == "MX": | |
| return [type("MX", (), {"exchange": "mx.example.test.", "preference": 10})()] | |
| if rtype == "TXT" and domain.startswith("_dmarc."): | |
| return [FakeAnswer("v=DMARC1; p=none")] | |
| if rtype == "TXT" and domain.startswith("default._domainkey."): | |
| return [FakeAnswer("v=DKIM1; k=rsa; p=abc")] | |
| if rtype == "TXT": | |
| return [FakeAnswer("v=spf1 include:example.test -all")] | |
| raise dns_check.dns.resolver.NoAnswer() | |
| class DnsAuthConcurrencyTests(unittest.IsolatedAsyncioTestCase): | |
| """SPF / DMARC / DKIM are independent DNS queries and must not run serially. | |
| Measured live: one request took 42.4s while the same address took 2.45s | |
| once caches were warm — the shape of ~8 sequential queries each paying a | |
| 5s timeout. The DKIM selector list alone is 6 sequential lookups. | |
| """ | |
| async def test_auth_record_lookups_run_concurrently(self): | |
| with disable_doh(), patch.object(dns_check.dns.asyncresolver, "Resolver", | |
| return_value=SlowResolver()): | |
| t0 = time.perf_counter() | |
| result = await dns_check.check_dns("example.test") | |
| elapsed = time.perf_counter() - t0 | |
| self.assertTrue(result.has_mx_records) | |
| self.assertTrue(result.has_spf) | |
| self.assertTrue(result.has_dmarc) | |
| self.assertTrue(result.has_dkim) | |
| # Sequential = 1 (MX) + 1 (SPF) + 1 (DMARC) + 6 (DKIM) = 0.9s. | |
| # Concurrent should be far below that. | |
| self.assertLess( | |
| elapsed, 0.5, | |
| f"auth lookups ran sequentially ({elapsed:.2f}s for 9 queries)", | |
| ) | |
| async def test_still_returns_a_result_when_auth_records_are_absent(self): | |
| class Bare(SlowResolver): | |
| async def resolve(self, domain, rtype): | |
| await asyncio.sleep(0.01) | |
| if rtype == "MX": | |
| return [type("MX", (), {"exchange": "mx.example.test.", | |
| "preference": 10})()] | |
| raise dns_check.dns.resolver.NoAnswer() | |
| with disable_doh(), patch.object(dns_check.dns.asyncresolver, "Resolver", return_value=Bare()): | |
| result = await dns_check.check_dns("example.test") | |
| self.assertTrue(result.has_mx_records) | |
| self.assertFalse(result.has_spf) | |
| self.assertFalse(result.has_dmarc) | |
| self.assertFalse(result.has_dkim) # absence is a verdict, not an error | |
| if __name__ == "__main__": | |
| unittest.main() |