gsadmin/haproxy-openmanager
1
0
Fork 0
mirror of https://github.com/taylanbakircioglu/haproxy-openmanager.git synced 2026-09-20 17:43:29 +00:00
Code Issues Packages Projects Releases Wiki Activity
Files
main
haproxy-openmanager/backend/tests/test_acme_diagnostics.py
T
Mustafa ULUKAYA 9e6e4dd03b fix(acme): repair the diagnostics contract the GET probe broke 
Running the suite properly — the previous rounds could not, pytest was not
installed locally — surfaced four failures, all in the code this branch
touches. main is clean at 1410 passed, so these were mine.

Two were real defects, not stale expectations:

check_port80 turned an unreadable body into a hard failure. The GET probe read
the body outside any guard, so a connection reset mid-response, or a server
that hangs after headers, made a healthy 404 fail. The body is EVIDENCE, not a
precondition: when it cannot be read the check now falls back to the
status-only semantics it has always had, and records body_class 'unread'. The
stricter rule applies only when there is something to judge.

check_routing raised KeyError on a row without `acme_enabled`. The column gates
a warning, so a row shape lacking it should not take down the whole diagnostic.

The rest were expectations that had to change, because the contract did:

- The probe is a GET now, so the test doubles needed a body. `_FakeHEADResp`
  became `_FakeGETResp` with headers and a readable content stream.
- "A port-80 frontend row exists" no longer means ok. That assertion is exactly
  the bug: it describes what the database wants, while the nodes run whatever
  was last applied — which is how this check reported success throughout an
  incident where the live config had no usable challenge route.

New coverage for the branches that had none, including the case that started
all of this: a proxy that has lost its /.well-known/acme-challenge/ location
serves its SPA with HTTP 200, which the old `status in (200, 404)` rule
accepted as healthy. Also the tcp-only cluster, which must warn rather than
fail — SiteWizard blocks submit on any failing check, so failing there would
lock those installs on upgrade day.

Verified against real dependencies rather than a stub harness: the app boots
with the challenge route registered; the field validator fires through pydantic
on both cluster models, normalising whitespace and rejecting scheme-less and
loopback values; `model_fields_set` really does distinguish an explicit null
from an omitted key, which is what makes clearing the field work; the settings
validator rejects both raw and jsonb-encoded bad values; and a legacy
scheme-less value is skipped so the next source renders.

Suite: 1477 passed on the branch, 1410 on main, 0 failed on either.
2026-08-11 20:13:06 +03:00

886 lines
31 KiB
Python
Raw Permalink Blame History

"""
v1.5.0 Feature A — services/acme_diagnostics.py per-check unit tests.
Each check is exercised with a mocked asyncpg connection (or no conn at all
for stdlib-only checks). DNS / port-80 are exercised through monkeypatched
asyncio primitives so the tests run hermetically — no real network.
"""
import asyncio
import json
import socket
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from services.acme_diagnostics import (
CHECK_IDS,
_check_result,
check_account,
check_agents,
check_dns,
check_port80,
check_routing,
run_checks,
)
# ----------------------------------------------------------------------------
# _check_result schema invariants
# ----------------------------------------------------------------------------
def test_check_result_default_shape():
r = _check_result("dns", "DNS resolution", "ok", "all good")
assert set(r.keys()) >= {"id", "label", "status", "severity", "message", "details", "duration_ms"}
assert r["id"] == "dns"
assert r["status"] == "ok"
assert r["severity"] == "info"
assert r["details"] == {}
assert r["duration_ms"] is None
def test_check_ids_constant_order():
assert CHECK_IDS == ("dns", "port80", "routing", "account", "agents")
# ----------------------------------------------------------------------------
# DNS check
# ----------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_check_dns_all_resolve(monkeypatch):
def fake_gethostbyname_ex(domain):
return (domain, [], ["10.0.0.1"])
monkeypatch.setattr(socket, "gethostbyname_ex", fake_gethostbyname_ex)
out = await check_dns(["a.example.com", "b.example.com"])
assert out["status"] == "ok"
assert out["details"]["resolved"]["a.example.com"] == ["10.0.0.1"]
assert out["duration_ms"] is not None and out["duration_ms"] >= 0
@pytest.mark.asyncio
async def test_check_dns_failure_marks_fail(monkeypatch):
def fake_gethostbyname_ex(domain):
raise socket.gaierror("Name or service not known")
monkeypatch.setattr(socket, "gethostbyname_ex", fake_gethostbyname_ex)
out = await check_dns(["nope.example.com"])
assert out["status"] == "fail"
assert out["severity"] == "error"
assert len(out["details"]["failed"]) == 1
assert out["details"]["failed"][0]["domain"] == "nope.example.com"
@pytest.mark.asyncio
async def test_check_dns_wildcard_skipped(monkeypatch):
"""*.example.com cannot be HTTP-01 validated — must NOT be resolved."""
called = []
def fake_gethostbyname_ex(domain):
called.append(domain)
return (domain, [], ["10.0.0.1"])
monkeypatch.setattr(socket, "gethostbyname_ex", fake_gethostbyname_ex)
out = await check_dns(["*.example.com"])
assert out["status"] == "ok"
assert called == [] # wildcard never reached the resolver
assert out["details"]["resolved"]["*.example.com"] == []
@pytest.mark.asyncio
async def test_check_dns_empty_ips_marks_failure(monkeypatch):
def fake_gethostbyname_ex(domain):
return (domain, [], [])
monkeypatch.setattr(socket, "gethostbyname_ex", fake_gethostbyname_ex)
out = await check_dns(["a.example.com"])
assert out["status"] == "fail"
assert "no A records" in out["details"]["failed"][0]["reason"]
# ----------------------------------------------------------------------------
# Port-80 check (GET probe)
#
# The probe is a GET, not a HEAD: a reverse proxy that has lost its
# /.well-known/acme-challenge/ location falls through to its catch-all and serves
# an SPA with HTTP 200, which a status-code-only check accepts as healthy while
# every real validation fails. The fakes below therefore carry a body.
# ----------------------------------------------------------------------------
class _FakeContent:
def __init__(self, body):
self._body = body
async def read(self, n=-1):
if self._body is None:
raise ConnectionResetError("reset mid-body")
return self._body if n is None or n < 0 else self._body[:n]
class _FakeGETResp:
def __init__(self, status, body=b"", content_type="text/plain"):
self.status = status
self.headers = {"content-type": content_type}
self.content = _FakeContent(body)
async def __aenter__(self):
return self
async def __aexit__(self, *args):
return False
class _FakeSession:
def __init__(self, *, statuses=None, raise_timeout=False, raise_client_error=False,
bodies=None, content_types=None):
self._statuses = list(statuses or [])
self._raise_timeout = raise_timeout
self._raise_client_error = raise_client_error
self._bodies = list(bodies or [])
self._content_types = list(content_types or [])
async def __aenter__(self):
return self
async def __aexit__(self, *args):
return False
def get(self, url, allow_redirects=False):
if self._raise_timeout:
raise asyncio.TimeoutError()
if self._raise_client_error:
import aiohttp
raise aiohttp.ClientError("connection refused")
status = self._statuses.pop(0) if self._statuses else 200
body = self._bodies.pop(0) if self._bodies else b""
ctype = self._content_types.pop(0) if self._content_types else "text/plain"
return _FakeGETResp(status, body, ctype)
def _mock_public_dns(monkeypatch, ip="93.184.216.34"):
"""R18b round 4 #B: check_port80 now refuses to probe domains
whose A records point at private/loopback/metadata IP space
(SSRF guard). Tests that exercise the success path must
monkeypatch DNS to a public-looking IP so the guard allows the
probe through."""
def fake_gethostbyname_ex(domain):
return (domain, [], [ip])
monkeypatch.setattr(socket, "gethostbyname_ex", fake_gethostbyname_ex)
@pytest.mark.asyncio
async def test_check_port80_ok_on_200(monkeypatch):
_mock_public_dns(monkeypatch)
def _ctor(*args, **kwargs):
return _FakeSession(statuses=[200, 200])
monkeypatch.setattr("aiohttp.ClientSession", _ctor)
out = await check_port80(["a.example.com", "b.example.com"])
assert out["status"] == "ok"
assert all(t["ok"] for t in out["details"]["targets"])
@pytest.mark.asyncio
async def test_check_port80_ok_on_404(monkeypatch):
"""404 on /.well-known/acme-challenge/* is a valid 'served' signal."""
_mock_public_dns(monkeypatch)
def _ctor(*args, **kwargs):
return _FakeSession(statuses=[404])
monkeypatch.setattr("aiohttp.ClientSession", _ctor)
out = await check_port80(["a.example.com"])
assert out["status"] == "ok"
@pytest.mark.asyncio
async def test_check_port80_warns_when_200_carries_a_web_page(monkeypatch):
"""The failure that motivated the GET probe.
A reverse proxy whose /.well-known/acme-challenge/ location has drifted away
falls through to its catch-all and serves the SPA. The status is 200, so the old
`status in (200, 404)` rule called the install healthy while every validation
failed. Only the body distinguishes them.
"""
_mock_public_dns(monkeypatch)
spa = b'<!doctype html><html><head><title>HAProxy OpenManager</title></head>'
def _ctor(*args, **kwargs):
return _FakeSession(statuses=[200], bodies=[spa], content_types=["text/html"])
monkeypatch.setattr("aiohttp.ClientSession", _ctor)
out = await check_port80(["a.example.com"])
assert out["status"] == "warn"
target = out["details"]["targets"][0]
assert target["body_class"] == "html"
assert not target.get("ok")
assert "HTML" in out["message"]
@pytest.mark.asyncio
async def test_check_port80_falls_back_to_status_when_the_body_cannot_be_read(monkeypatch):
"""A body that cannot be read is missing evidence, not a verdict.
Turning a connection reset mid-response into a hard failure would make a healthy
404 fail intermittently, so the check keeps its original status-only semantics
whenever there is nothing to judge.
"""
_mock_public_dns(monkeypatch)
def _ctor(*args, **kwargs):
return _FakeSession(statuses=[404], bodies=[None])
monkeypatch.setattr("aiohttp.ClientSession", _ctor)
out = await check_port80(["a.example.com"])
assert out["status"] == "ok"
assert out["details"]["targets"][0]["body_class"] == "unread"
@pytest.mark.asyncio
async def test_check_port80_warns_on_a_redirect(monkeypatch):
_mock_public_dns(monkeypatch)
def _ctor(*args, **kwargs):
return _FakeSession(statuses=[301])
monkeypatch.setattr("aiohttp.ClientSession", _ctor)
out = await check_port80(["a.example.com"])
assert out["status"] == "warn"
assert out["details"]["targets"][0]["diagnosis"]
@pytest.mark.asyncio
async def test_check_port80_warn_on_egress_timeout(monkeypatch):
"""Corporate egress blocks port 80 outbound — warn, don't fail."""
_mock_public_dns(monkeypatch)
def _ctor(*args, **kwargs):
return _FakeSession(raise_timeout=True)
monkeypatch.setattr("aiohttp.ClientSession", _ctor)
out = await check_port80(["a.example.com"])
assert out["status"] == "warn"
assert out["severity"] == "warn"
@pytest.mark.asyncio
async def test_check_port80_skips_private_ip_for_ssrf_guard(monkeypatch):
"""R18b round 4 #B: SSRF guard. A domain that resolves to a
private/loopback/metadata IP must NOT trigger an outbound HTTP
request — the diagnostic must skip it with a warn-level row.
Pre-fix this was a usable SSRF primitive for any authenticated
operator."""
def fake_gethostbyname_ex(domain):
# AWS / GCP metadata IP — most dangerous SSRF target
return (domain, [], ["169.254.169.254"])
monkeypatch.setattr(socket, "gethostbyname_ex", fake_gethostbyname_ex)
# Track whether ClientSession.head was called — it must not be.
head_called = []
class _SpyClientSession:
def __init__(self, *args, **kwargs):
pass
async def __aenter__(self):
return self
async def __aexit__(self, *a, **k):
return None
def head(self, url, **kwargs):
head_called.append(url)
class _Resp:
async def __aenter__(self_inner):
self_inner.status = 200
return self_inner
async def __aexit__(self_inner, *a, **k):
return None
return _Resp()
monkeypatch.setattr("aiohttp.ClientSession", _SpyClientSession)
out = await check_port80(["evil.example.com"])
assert head_called == [], (
"SSRF guard regression: check_port80 issued an outbound HEAD "
"to a private-IP domain"
)
targets = out["details"]["targets"]
assert any("non-public" in (t.get("skip") or "") for t in targets), (
"SSRF guard regression: skip row missing for non-public IP"
)
@pytest.mark.asyncio
async def test_check_port80_skips_loopback_ip_for_ssrf_guard(monkeypatch):
"""SSRF guard must also block loopback (127.0.0.1)."""
def fake_gethostbyname_ex(domain):
return (domain, [], ["127.0.0.1"])
monkeypatch.setattr(socket, "gethostbyname_ex", fake_gethostbyname_ex)
head_called = []
class _SpyClientSession:
def __init__(self, *args, **kwargs):
pass
async def __aenter__(self):
return self
async def __aexit__(self, *a, **k):
return None
def head(self, url, **kwargs):
head_called.append(url)
raise RuntimeError("should never be called")
monkeypatch.setattr("aiohttp.ClientSession", _SpyClientSession)
out = await check_port80(["loopback.example.com"])
assert head_called == []
assert any("non-public" in (t.get("skip") or "") for t in out["details"]["targets"])
@pytest.mark.asyncio
async def test_check_port80_skips_rfc1918_for_ssrf_guard(monkeypatch):
"""SSRF guard must also block RFC1918 (10.0.0.0/8)."""
def fake_gethostbyname_ex(domain):
return (domain, [], ["10.0.0.42"])
monkeypatch.setattr(socket, "gethostbyname_ex", fake_gethostbyname_ex)
head_called = []
class _SpyClientSession:
def __init__(self, *args, **kwargs):
pass
async def __aenter__(self):
return self
async def __aexit__(self, *a, **k):
return None
def head(self, url, **kwargs):
head_called.append(url)
raise RuntimeError("should never be called")
monkeypatch.setattr("aiohttp.ClientSession", _SpyClientSession)
out = await check_port80(["internal.example.com"])
assert head_called == []
@pytest.mark.asyncio
async def test_check_port80_fail_on_client_error(monkeypatch):
_mock_public_dns(monkeypatch)
def _ctor(*args, **kwargs):
return _FakeSession(raise_client_error=True)
monkeypatch.setattr("aiohttp.ClientSession", _ctor)
out = await check_port80(["a.example.com"])
assert out["status"] == "fail"
assert out["severity"] == "error"
@pytest.mark.asyncio
async def test_check_port80_skipped_when_only_wildcards(monkeypatch):
"""We never probe wildcards (HTTP-01 is not applicable)."""
out = await check_port80(["*.example.com"])
assert out["status"] == "skipped"
@pytest.mark.asyncio
async def test_check_port80_fail_on_500(monkeypatch):
_mock_public_dns(monkeypatch)
def _ctor(*args, **kwargs):
return _FakeSession(statuses=[500])
monkeypatch.setattr("aiohttp.ClientSession", _ctor)
out = await check_port80(["a.example.com"])
assert out["status"] == "fail"
# ----------------------------------------------------------------------------
# Routing check
# ----------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_check_routing_warn_when_no_clusters():
conn = AsyncMock()
out = await check_routing(conn, ["a.example.com"], [])
assert out["status"] == "warn"
conn.fetch.assert_not_awaited()
@pytest.mark.asyncio
async def test_check_routing_fail_when_no_port80_frontend():
conn = AsyncMock()
conn.fetch.return_value = []
out = await check_routing(conn, ["a.example.com"], [1])
assert out["status"] == "fail"
assert "No HTTP frontend" in out["message"]
def _routing_row(**over):
row = {"id": 1, "name": "fe-http", "bind_address": "0.0.0.0", "bind_port": 80,
"mode": "http", "default_backend": "be", "cluster_id": 1, "acme_enabled": True}
row.update(over)
return row
def _applied(has_route=True, server_line="server _acme_mgmt 10.90.1.4:80"):
return {"has_route": has_route, "server_line": server_line}
@pytest.mark.asyncio
async def test_check_routing_ok_when_challenge_route_is_in_the_applied_config():
# A port-80 frontend row alone is NOT enough. It describes what the database
# wants; the nodes run whatever was last applied. During the incident this
# function reported "ok" from the row count while the live config had no usable
# challenge route at all.
conn = AsyncMock()
conn.fetch.return_value = [_routing_row()]
conn.fetchrow.return_value = _applied()
out = await check_routing(conn, ["a.example.com"], [1])
assert out["status"] == "ok"
assert len(out["details"]["frontends"]) == 1
assert out["details"]["challenge_backends"] == {1: "10.90.1.4:80"}
@pytest.mark.asyncio
async def test_check_routing_warns_when_acme_is_disabled_on_the_cluster():
conn = AsyncMock()
conn.fetch.return_value = [_routing_row(acme_enabled=False)]
out = await check_routing(conn, ["a.example.com"], [1])
assert out["status"] == "warn"
assert "disabled" in out["message"]
@pytest.mark.asyncio
async def test_check_routing_warns_when_the_route_is_not_in_the_applied_config():
conn = AsyncMock()
conn.fetch.return_value = [_routing_row()]
conn.fetchrow.return_value = _applied(has_route=False)
out = await check_routing(conn, ["a.example.com"], [1])
assert out["status"] == "warn"
assert "apply the cluster" in out["message"].lower()
@pytest.mark.asyncio
async def test_check_routing_warns_when_the_challenge_backend_has_no_server_line():
# `haproxy -c` passes because the section exists, so nothing else catches this;
# every challenge request 503s from an empty backend.
conn = AsyncMock()
conn.fetch.return_value = [_routing_row()]
conn.fetchrow.return_value = _applied(server_line=None)
out = await check_routing(conn, ["a.example.com"], [1])
assert out["status"] == "warn"
assert "503" in out["message"]
@pytest.mark.asyncio
async def test_check_routing_warns_when_the_challenge_backend_is_loopback():
conn = AsyncMock()
conn.fetch.return_value = [_routing_row()]
conn.fetchrow.return_value = _applied(server_line="server _acme_mgmt 127.0.0.1:8080")
out = await check_routing(conn, ["a.example.com"], [1])
assert out["status"] == "warn"
assert "HAProxy node" in out["message"]
@pytest.mark.asyncio
async def test_check_routing_warns_rather_than_fails_on_a_tcp_only_port80_cluster():
# The `fail` branch must stay reachable only when NO port-80 frontend exists at
# all: SiteWizard blocks submit on any failing check, so turning this into a
# failure would lock tcp-only installs the day it ships.
conn = AsyncMock()
conn.fetch.return_value = [_routing_row(mode="tcp")]
out = await check_routing(conn, ["a.example.com"], [1])
assert out["status"] == "warn"
assert "tcp mode" in out["message"]
@pytest.mark.asyncio
async def test_check_routing_treats_null_mode_as_http_like_the_renderer():
conn = AsyncMock()
conn.fetch.return_value = [_routing_row(mode=None)]
conn.fetchrow.return_value = _applied()
out = await check_routing(conn, ["a.example.com"], [1])
assert out["status"] == "ok"
# ----------------------------------------------------------------------------
# Account check
# ----------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_check_account_fail_when_no_account_id():
conn = AsyncMock()
out = await check_account(conn, None)
assert out["status"] == "fail"
assert "no ACME account id" in out["message"]
@pytest.mark.asyncio
async def test_check_account_fail_when_not_found():
conn = AsyncMock()
conn.fetchrow.return_value = None
out = await check_account(conn, 99)
assert out["status"] == "fail"
assert "Account 99 not found" in out["message"]
@pytest.mark.asyncio
async def test_check_account_fail_when_status_invalid():
conn = AsyncMock()
conn.fetchrow.return_value = {
"id": 1,
"email": "ops@example.com",
"status": "deactivated",
"account_url": "https://acme/acct/1",
}
out = await check_account(conn, 1)
assert out["status"] == "fail"
assert "deactivated" in out["message"]
@pytest.mark.asyncio
async def test_check_account_warn_when_url_missing():
conn = AsyncMock()
conn.fetchrow.return_value = {
"id": 1, "email": "ops@example.com",
"status": "valid", "account_url": None,
}
out = await check_account(conn, 1)
assert out["status"] == "warn"
@pytest.mark.asyncio
async def test_check_account_ok():
conn = AsyncMock()
conn.fetchrow.return_value = {
"id": 1, "email": "ops@example.com",
"status": "valid", "account_url": "https://acme/acct/1",
}
out = await check_account(conn, 1)
assert out["status"] == "ok"
assert out["severity"] == "info"
# ----------------------------------------------------------------------------
# Agents check
# ----------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_check_agents_warn_when_no_clusters():
conn = AsyncMock()
out = await check_agents(conn, [])
assert out["status"] == "warn"
conn.fetch.assert_not_awaited()
@pytest.mark.asyncio
async def test_check_agents_fail_when_none_registered():
conn = AsyncMock()
conn.fetch.return_value = []
out = await check_agents(conn, [1])
assert out["status"] == "fail"
@pytest.mark.asyncio
async def test_check_agents_warn_when_none_active():
conn = AsyncMock()
conn.fetch.return_value = [
{"id": 1, "hostname": "h1", "status": "offline", "last_seen": None,
"cluster_id": 1, "cluster_name": "c1"},
]
out = await check_agents(conn, [1])
assert out["status"] == "warn"
@pytest.mark.asyncio
async def test_check_agents_ok_with_active():
conn = AsyncMock()
conn.fetch.return_value = [
{"id": 1, "hostname": "h1", "status": "active", "last_seen": None,
"cluster_id": 1, "cluster_name": "c1"},
{"id": 2, "hostname": "h2", "status": "offline", "last_seen": None,
"cluster_id": 1, "cluster_name": "c1"},
]
out = await check_agents(conn, [1])
assert out["status"] == "ok"
assert "1 of 2" in out["message"]
@pytest.mark.asyncio
async def test_bulgu84_check_agents_uses_last_seen_column():
"""Bulgu #84 (round-23 audit) — pin the SQL column name.
Pre-fix the query referenced a non-existent ``a.last_heartbeat``
column. AsyncMock returns whatever dict the test sets without
re-validating the SQL string, so the pre-fix test suite was
GREEN while every live ACME preflight call (the
``POST /api/sites/preflight-acme`` endpoint that the wizard
runs before showing the ACME step) crashed with
``UndefinedColumnError: column a.last_heartbeat does not
exist``. The crash blocked the entire wizard ACME preview
page on a production cluster, but unit tests never noticed
because they only assert on the helper's return shape, not
on the SQL string the helper sends to Postgres.
This pin inspects ``conn.fetch.call_args`` to assert the SQL
body actually queries ``a.last_seen`` (the canonical column
name used everywhere else in the codebase — see
routers/cluster.py:695-702 and routers/agent.py:281+ for
sibling readers). A future ``last_heartbeat`` typo would
re-fail this test without anyone noticing the live impact.
"""
conn = AsyncMock()
conn.fetch.return_value = [
{"id": 1, "hostname": "h1", "status": "active", "last_seen": None,
"cluster_id": 1, "cluster_name": "c1"},
]
await check_agents(conn, [1])
conn.fetch.assert_awaited_once()
sql_query = conn.fetch.call_args[0][0]
assert "a.last_seen" in sql_query, (
f"check_agents SQL must select a.last_seen (the canonical "
f"agents-table timestamp column); got: {sql_query!r}"
)
assert "a.last_heartbeat" not in sql_query, (
f"check_agents SQL still references a.last_heartbeat — this "
f"column does NOT exist on the agents table and the query "
f"will 500 with UndefinedColumnError at runtime. SQL: "
f"{sql_query!r}"
)
# ----------------------------------------------------------------------------
# run_checks orchestration
# ----------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_run_checks_full_suite_returns_all_five(monkeypatch):
monkeypatch.setattr(socket, "gethostbyname_ex",
lambda d: (d, [], ["10.0.0.1"]))
def _ctor(*args, **kwargs):
return _FakeSession(statuses=[200])
monkeypatch.setattr("aiohttp.ClientSession", _ctor)
conn = AsyncMock()
conn.fetch.return_value = []
conn.fetchrow.return_value = None
out = await run_checks(
conn,
domains=["a.example.com"],
cluster_ids=[1],
account_id=None,
)
ids = [c["id"] for c in out]
assert ids == ["dns", "port80", "routing", "account", "agents"]
@pytest.mark.asyncio
async def test_run_checks_only_filter(monkeypatch):
"""`only` lets the UI re-run a single check."""
conn = AsyncMock()
conn.fetchrow.return_value = {
"id": 1, "email": "x@y", "status": "valid", "account_url": "https://acme/1",
}
out = await run_checks(
conn, domains=["a.example.com"], cluster_ids=[1],
account_id=1, only=["account"],
)
assert len(out) == 1
assert out[0]["id"] == "account"
@pytest.mark.asyncio
async def test_run_checks_unknown_only_returns_empty():
conn = AsyncMock()
out = await run_checks(
conn, domains=["a.example.com"], cluster_ids=[1],
account_id=None, only=["bogus"],
)
assert out == []
# ----------------------------------------------------------------------------
# Bulgu #94 / #95 (Round-25 audit) — diagnostic-runner robustness
# ----------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_bulgu94_run_checks_swallows_single_check_crash(monkeypatch):
"""Bulgu #94 — a single check exception must NOT collapse the suite.
Pre-fix, an asyncpg UndefinedColumnError from check_agents (e.g. the
Bulgu #84 ``a.last_heartbeat`` typo on an older deploy) propagated
up to the FastAPI router which had no `except`, so the operator saw
HTTP 500 with no body. The diagnostic panel is precisely the place
that should SURFACE this — never opaque-500 it. We now wrap each
check; the failing one becomes a structured `fail` row and the
other four still render.
"""
monkeypatch.setattr(socket, "gethostbyname_ex",
lambda d: (d, [], ["10.0.0.1"]))
def _ctor(*args, **kwargs):
return _FakeSession(statuses=[200])
monkeypatch.setattr("aiohttp.ClientSession", _ctor)
conn = AsyncMock()
# check_routing + check_agents both call conn.fetch; explode on the
# FIRST call (which is check_routing) and return rows on the second.
call_count = {"n": 0}
async def _fetch(*args, **kwargs):
call_count["n"] += 1
if call_count["n"] == 1:
raise RuntimeError("simulated: column a.last_heartbeat does not exist")
return []
conn.fetch = _fetch
conn.fetchrow.return_value = None
out = await run_checks(
conn,
domains=["a.example.com"],
cluster_ids=[1],
account_id=None,
)
# All five checks must still appear in the response.
ids = [c["id"] for c in out]
assert ids == ["dns", "port80", "routing", "account", "agents"]
routing = next(c for c in out if c["id"] == "routing")
assert routing["status"] == "fail"
assert "Diagnostic check crashed" in routing["message"]
assert routing["details"]["exception_type"] == "RuntimeError"
assert "last_heartbeat" in routing["details"]["exception_message"]
@pytest.mark.asyncio
async def test_bulgu94_run_checks_coerces_string_cluster_ids(monkeypatch):
"""Bulgu #94 — cluster_ids stored as JSONB strings (legacy paths)
must not crash check_routing / check_agents with
``invalid input syntax for type integer: "1"``."""
monkeypatch.setattr(socket, "gethostbyname_ex",
lambda d: (d, [], ["10.0.0.1"]))
def _ctor(*args, **kwargs):
return _FakeSession(statuses=[200])
monkeypatch.setattr("aiohttp.ClientSession", _ctor)
captured_args = []
async def _fetch(*args, **kwargs):
captured_args.append(args)
return []
conn = AsyncMock()
conn.fetch = _fetch
conn.fetchrow.return_value = None
out = await run_checks(
conn,
domains=["a.example.com"],
cluster_ids=["1", "2", "garbage", None, 3],
account_id=None,
)
# The list passed to asyncpg should already be a pure-int list.
# check_routing is the first call that uses cluster_ids.
routing_call_args = [a for a in captured_args if "frontends" in a[0]]
assert routing_call_args, "check_routing should have queried frontends"
cluster_ids_arg = routing_call_args[0][1]
assert cluster_ids_arg == [1, 2, 3], (
f"cluster_ids must be coerced to ints before being passed to "
f"asyncpg's ::int[] cast; got: {cluster_ids_arg!r}"
)
assert all(c["status"] != "fail" or c["id"] != "routing"
for c in out
if c["id"] == "routing" and "Diagnostic check crashed" in (c.get("message") or "")
), "routing should not have crashed on coerced cluster_ids"
@pytest.mark.asyncio
async def test_bulgu94_safe_check_does_not_swallow_cancellation(monkeypatch):
"""`_safe_check` must catch `Exception` but NOT `BaseException`.
asyncio.CancelledError is a BaseException (Python 3.8+) so it must
propagate out of `_safe_check` — otherwise a request that the
client cancelled mid-flight would silently keep running diagnostic
checks instead of unwinding cleanly. We hand `_safe_check` a coro
that raises CancelledError and assert it bubbles up.
"""
from services.acme_diagnostics import _safe_check
async def _cancelled_coro():
raise asyncio.CancelledError()
with pytest.raises(asyncio.CancelledError):
await _safe_check("dns", "DNS resolution", _cancelled_coro())
@pytest.mark.asyncio
async def test_bulgu94_safe_check_handles_keyboardinterrupt(monkeypatch):
"""`_safe_check` must also not swallow `KeyboardInterrupt`."""
from services.acme_diagnostics import _safe_check
async def _interrupt_coro():
raise KeyboardInterrupt()
with pytest.raises(KeyboardInterrupt):
await _safe_check("dns", "DNS resolution", _interrupt_coro())
@pytest.mark.asyncio
async def test_bulgu94_coerce_cluster_ids_handles_none():
"""Coercion must handle None input without raising."""
from services.acme_diagnostics import _coerce_cluster_ids
assert _coerce_cluster_ids(None) == []
assert _coerce_cluster_ids([]) == []
assert _coerce_cluster_ids([1, 2, 3]) == [1, 2, 3]
assert _coerce_cluster_ids(["1", "2"]) == [1, 2]
assert _coerce_cluster_ids([1.5]) == [1] # int() truncates floats
assert _coerce_cluster_ids(["abc", None, "5"]) == [5]
@pytest.mark.asyncio
async def test_bulgu94_run_checks_filters_invalid_domains(monkeypatch):
"""Non-string entries in `domains` must not reach the DNS resolver."""
seen_domains = []
def fake_gethostbyname_ex(domain):
seen_domains.append(domain)
return (domain, [], ["10.0.0.1"])
monkeypatch.setattr(socket, "gethostbyname_ex", fake_gethostbyname_ex)
def _ctor(*args, **kwargs):
return _FakeSession(statuses=[200])
monkeypatch.setattr("aiohttp.ClientSession", _ctor)
conn = AsyncMock()
conn.fetch.return_value = []
conn.fetchrow.return_value = None
out = await run_checks(
conn,
domains=["a.example.com", None, "", 42, "b.example.com"],
cluster_ids=[1],
account_id=None,
)
# Both check_dns and check_port80 resolve DNS, so each valid domain
# may appear multiple times — but invalid entries (None, "", 42)
# must never reach the resolver.
assert set(seen_domains) == {"a.example.com", "b.example.com"}
assert None not in seen_domains
assert "" not in seen_domains
assert 42 not in seen_domains
dns_check = next(c for c in out if c["id"] == "dns")
assert dns_check["status"] == "ok"
Reference in New Issue View Git Blame Copy Permalink