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 | """DoH DNS/MX worker tests (spec §3.2 / §7.3) — mocked HTTP, hermetic.""" | |
| import unittest | |
| from unittest.mock import AsyncMock, patch | |
| from app.validator import doh_resolver | |
| from app.validator.doh_resolver import ( | |
| _mx_sets_match, | |
| _parse_mx, | |
| resolve_mx_dual, | |
| resolve_mx_dual_cached, | |
| ) | |
| def _doh_response(status=0, answers=None): | |
| return {"Status": status, "Answer": answers or []} | |
| def _mx_answer(priority, host): | |
| return {"name": "example.com.", "type": 15, "TTL": 300, "data": f"{priority} {host}"} | |
| def _a_answer(ip="93.184.216.34"): | |
| return {"name": "example.com.", "type": 1, "TTL": 300, "data": ip} | |
| class _FakeResp: | |
| def __init__(self, data): | |
| self._data = data | |
| def raise_for_status(self): | |
| return None | |
| def json(self): | |
| return self._data | |
| class FakeDoHClient: | |
| """Routes by (host, type) so MX / A / AAAA queries can differ. | |
| `responses` maps ("google"|"cloudflare", rtype) -> payload-or-Exception. | |
| """ | |
| responses: dict = {} | |
| def __init__(self, responses=None, *a, **kw): | |
| if responses is not None: | |
| self.responses = responses | |
| async def __aenter__(self): | |
| return self | |
| async def __aexit__(self, *a): | |
| return False | |
| async def get(self, url, params=None, **kw): | |
| host = "google" if "dns.google" in url else "cloudflare" | |
| rtype = (params or {}).get("type", "MX") | |
| payload = self.responses.get((host, rtype)) | |
| if payload is None: | |
| payload = {"Status": 0, "Answer": []} | |
| if isinstance(payload, Exception): | |
| raise payload | |
| return _FakeResp(payload) | |
| def _client_patch(responses): | |
| return patch( | |
| "app.validator.doh_resolver.httpx.AsyncClient", | |
| lambda *a, **kw: FakeDoHClient(responses, *a, **kw), | |
| ) | |
| class ParseTests(unittest.TestCase): | |
| def test_parse_mx_sorted_by_priority(self): | |
| data = _doh_response(0, [_mx_answer(20, "mx2.example.com."), _mx_answer(10, "mx1.example.com.")]) | |
| records = _parse_mx(data) | |
| self.assertEqual(records, [ | |
| {"host": "mx1.example.com", "priority": 10}, | |
| {"host": "mx2.example.com", "priority": 20}, | |
| ]) | |
| def test_parse_mx_ignores_non_mx(self): | |
| data = _doh_response(0, [ | |
| {"name": "example.com.", "type": 1, "TTL": 300, "data": "93.184.216.34"}, | |
| _mx_answer(10, "mx1.example.com."), | |
| ]) | |
| records = _parse_mx(data) | |
| self.assertEqual(len(records), 1) | |
| def test_sets_match_order_insensitive(self): | |
| a = [{"host": "mx1", "priority": 10}, {"host": "mx2", "priority": 20}] | |
| b = [{"host": "mx2", "priority": 20}, {"host": "mx1", "priority": 10}] | |
| self.assertTrue(_mx_sets_match(a, b)) | |
| self.assertFalse(_mx_sets_match(a, [{"host": "mx3", "priority": 10}])) | |
| class ResolveDualTests(unittest.IsolatedAsyncioTestCase): | |
| async def test_consistent_resolvers(self): | |
| mx = [_mx_answer(10, "mx1.example.com.")] | |
| responses = {("google", "MX"): _doh_response(0, mx), ("cloudflare", "MX"): _doh_response(0, mx)} | |
| with _client_patch(responses): | |
| out = await resolve_mx_dual("example.com") | |
| self.assertEqual(out["status"], "ok") | |
| self.assertTrue(out["consistent"]) | |
| self.assertEqual(out["mx_records"][0]["host"], "mx1.example.com") | |
| async def test_inconsistent_resolvers_flagged(self): | |
| g = [_mx_answer(10, "mx1.example.com.")] | |
| c = [_mx_answer(10, "mx2.example.com.")] | |
| responses = {("google", "MX"): _doh_response(0, g), ("cloudflare", "MX"): _doh_response(0, c)} | |
| with _client_patch(responses): | |
| out = await resolve_mx_dual("example.com") | |
| self.assertEqual(out["status"], "ok") | |
| self.assertIs(out["consistent"], False) | |
| self.assertEqual(out["mx_records"][0]["host"], "mx1.example.com") | |
| async def test_nxdomain(self): | |
| responses = { | |
| ("google", "MX"): _doh_response(3), | |
| ("cloudflare", "MX"): _doh_response(3), | |
| } | |
| with _client_patch(responses): | |
| out = await resolve_mx_dual("nope.invalid") | |
| self.assertEqual(out["status"], "nxdomain") | |
| async def test_a_fallback_when_no_mx(self): | |
| responses = { | |
| ("google", "MX"): _doh_response(0, []), | |
| ("cloudflare", "MX"): _doh_response(0, []), | |
| ("google", "A"): _doh_response(0, [_a_answer()]), | |
| ("google", "AAAA"): _doh_response(0, []), | |
| } | |
| with _client_patch(responses): | |
| out = await resolve_mx_dual("example.com") | |
| self.assertEqual(out["status"], "ok") | |
| self.assertTrue(out["a_fallback"]) | |
| self.assertEqual(out["mx_records"], [{"host": "example.com", "priority": 0}]) | |
| async def test_no_mail_when_no_mx_no_address(self): | |
| responses = { | |
| ("google", "MX"): _doh_response(0, []), | |
| ("cloudflare", "MX"): _doh_response(0, []), | |
| ("google", "A"): _doh_response(0, []), | |
| ("google", "AAAA"): _doh_response(0, []), | |
| } | |
| with _client_patch(responses): | |
| out = await resolve_mx_dual("example.com") | |
| self.assertEqual(out["status"], "no_mail") | |
| async def test_primary_failure_falls_back_to_cloudflare_answer(self): | |
| mx = [_mx_answer(10, "mx1.example.com.")] | |
| responses = { | |
| ("google", "MX"): RuntimeError("google down"), | |
| ("cloudflare", "MX"): _doh_response(0, mx), | |
| } | |
| with _client_patch(responses): | |
| out = await resolve_mx_dual("example.com") | |
| self.assertEqual(out["status"], "ok") | |
| self.assertEqual(out["mx_records"][0]["host"], "mx1.example.com") | |
| self.assertIsNone(out["consistent"]) | |
| async def test_both_fail_is_error(self): | |
| responses = { | |
| ("google", "MX"): RuntimeError("g down"), | |
| ("cloudflare", "MX"): RuntimeError("c down"), | |
| } | |
| with _client_patch(responses): | |
| out = await resolve_mx_dual("example.com") | |
| self.assertEqual(out["status"], "error") | |
| class CachedResolveTests(unittest.IsolatedAsyncioTestCase): | |
| async def test_cache_hit_short_circuits_network(self): | |
| payload = { | |
| "status": "ok", "mx_records": [{"host": "mx1.example.com", "priority": 10}], | |
| "consistent": True, "a_fallback": False, "error": None, "cache_hit": True, | |
| } | |
| with patch.object( | |
| doh_resolver, "_cache_get", AsyncMock(return_value=payload) | |
| ), patch.object( | |
| doh_resolver, "resolve_mx_dual", | |
| AsyncMock(side_effect=AssertionError("must not resolve")), | |
| ): | |
| out = await resolve_mx_dual_cached("example.com") | |
| self.assertTrue(out["cache_hit"]) | |
| self.assertEqual(out["mx_records"][0]["host"], "mx1.example.com") | |
| async def test_ok_result_is_cached(self): | |
| result = { | |
| "status": "ok", "mx_records": [{"host": "mx1.example.com", "priority": 10}], | |
| "consistent": True, "a_fallback": False, "error": None, | |
| } | |
| with patch.object(doh_resolver, "_cache_get", AsyncMock(return_value=None)), \ | |
| patch.object(doh_resolver, "_cache_set", AsyncMock()) as set_mock, \ | |
| patch.object(doh_resolver, "resolve_mx_dual", AsyncMock(return_value=result)): | |
| out = await resolve_mx_dual_cached("example.com") | |
| set_mock.assert_awaited_once() | |
| args, kwargs = set_mock.call_args | |
| self.assertEqual(kwargs.get("ttl") or args[2], doh_resolver.MX_TTL_SECONDS) | |
| self.assertFalse(out.get("cache_hit", False)) | |
| async def test_error_results_never_cached(self): | |
| calls = {"n": 0} | |
| async def fake_resolve(domain, timeout=None): | |
| calls["n"] += 1 | |
| return {"status": "error", "mx_records": [], "consistent": None, | |
| "a_fallback": False, "error": "doh down"} | |
| with patch.object(doh_resolver, "_cache_get", AsyncMock(return_value=None)), \ | |
| patch.object(doh_resolver, "_cache_set", AsyncMock()) as set_mock, \ | |
| patch.object(doh_resolver, "resolve_mx_dual", fake_resolve): | |
| await resolve_mx_dual_cached("example.com") | |
| await resolve_mx_dual_cached("example.com") | |
| set_mock.assert_not_called() | |
| self.assertEqual(calls["n"], 2) | |
| if __name__ == "__main__": | |
| unittest.main() | |