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