patrickandClaude Sonnet 5.5
453c077386
Lesender Selbsttest (Version, Services, Hosts, Downtimes) beim Start und täglich 04:00, Seite /selftest, Teilcheck checkmk_api in /health, Chat-Meldung bei Wechsel. Toleranter Antwort-Parser (value/members, state_type int/String), Fallbacks für Host-Abfrage und Downtime-Liste, Fehlerhinweise je HTTP-Status. Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
121 lines
4.5 KiB
Python
121 lines
4.5 KiB
Python
import json
|
|
|
|
import httpx
|
|
|
|
from app.checkmk import (DOWNTIME_MODE_QUERY, HOST_QUERY_GET, HOST_QUERY_POST, CheckmkClient, CheckmkError,
|
|
Downtime, HostState, ServiceState, downtimes_from_states, status_hint)
|
|
|
|
def make_client(handler):
|
|
return CheckmkClient("https://cmk.local", "prod", "bot", "s3cret", transport=httpx.MockTransport(handler))
|
|
|
|
|
|
def host_row(name="srv1", state=1):
|
|
return {"extensions": {"name": name, "state": state, "plugin_output": "x", "last_state_change": 1,
|
|
"scheduled_downtime_depth": 0, "acknowledged": 0}}
|
|
|
|
|
|
async def test_problem_hosts_with_fallback_switches_to_get_and_remembers_mode():
|
|
calls = []
|
|
|
|
def handler(request):
|
|
calls.append(request.method)
|
|
if request.method == "POST":
|
|
return httpx.Response(422, text="Spalten abgelehnt")
|
|
return httpx.Response(200, json={"value": [host_row(state=0), host_row("srv2", state=2)]})
|
|
|
|
client = make_client(handler)
|
|
hosts, mode = await client.problem_hosts_with_fallback()
|
|
assert mode == HOST_QUERY_GET
|
|
assert [h.host for h in hosts] == ["srv2"]
|
|
assert calls == ["POST", "GET"]
|
|
|
|
calls.clear()
|
|
await client.problem_hosts()
|
|
assert calls == ["GET"]
|
|
|
|
|
|
async def test_problem_hosts_with_fallback_raises_when_both_fail():
|
|
def handler(request):
|
|
return httpx.Response(404, text="weg")
|
|
|
|
try:
|
|
await make_client(handler).problem_hosts_with_fallback()
|
|
except CheckmkError as exc:
|
|
assert exc.brief == "HTTP 404"
|
|
else:
|
|
raise AssertionError("expected CheckmkError")
|
|
|
|
|
|
async def test_selftest_hosts_reports_raw_rows_and_mode():
|
|
def handler(request):
|
|
return httpx.Response(200, json={"value": [host_row(state=2)]})
|
|
|
|
rows, hosts, mode = await make_client(handler).selftest_hosts()
|
|
assert mode == HOST_QUERY_POST
|
|
assert rows[0]["name"] == "srv1"
|
|
assert [h.state for h in hosts] == [3]
|
|
|
|
|
|
async def test_selftest_services_keeps_raw_rows_when_parse_fails():
|
|
body = {"value": [{"extensions": {"description": "s", "state": 2}}]} # host_name fehlt (Pflichtspalte)
|
|
|
|
rows, services = await make_client(lambda r: httpx.Response(200, json=body)).selftest_services()
|
|
assert rows == [{"description": "s", "state": 2}]
|
|
assert services == []
|
|
|
|
|
|
async def test_downtimes_with_fallback_uses_query_when_type_rejected():
|
|
seen = []
|
|
|
|
def handler(request):
|
|
seen.append(request)
|
|
if "downtime_type" in request.url.params:
|
|
return httpx.Response(422, text="abgelehnt")
|
|
return httpx.Response(200, json={"value": [{"extensions": {
|
|
"host_name": "web01", "service_description": "", "start_time": "", "end_time": "",
|
|
"comment": "", "author": ""}}]})
|
|
|
|
downtimes, mode = await make_client(handler).downtimes_with_fallback()
|
|
assert mode == DOWNTIME_MODE_QUERY
|
|
assert len(downtimes) == 2
|
|
queries = [json.loads(r.url.params["query"]) for r in seen if "query" in r.url.params]
|
|
assert {q["right"] for q in queries} == {"0", "1"}
|
|
|
|
|
|
async def test_downtimes_with_fallback_propagates_when_both_fail():
|
|
def handler(request):
|
|
return httpx.Response(404, text="weg")
|
|
|
|
try:
|
|
await make_client(handler).downtimes_with_fallback()
|
|
except CheckmkError as exc:
|
|
assert exc.brief == "HTTP 404"
|
|
else:
|
|
raise AssertionError("expected CheckmkError")
|
|
|
|
|
|
def test_downtimes_from_states_derives_from_scheduled_downtime_depth():
|
|
services = [ServiceState("h1", "s1", 2, True, "", 1, True, False), ServiceState("h1", "s2", 0, True, "", 1, False, False)]
|
|
hosts = [HostState("h2", 2, "", 1, True, False)]
|
|
derived = downtimes_from_states(services, hosts)
|
|
assert derived == [Downtime("h2", None, "", "", "", ""), Downtime("h1", "s1", "", "", "", "")]
|
|
|
|
|
|
def test_status_hint_mapping_never_touches_brief():
|
|
assert status_hint(401) == "Zugangsdaten prüfen"
|
|
assert status_hint(403) == "Rechte des Automation-Users prüfen"
|
|
assert status_hint(404) == "Pfad oder Site prüfen"
|
|
assert status_hint(422) == "Anfrage von Checkmk abgelehnt (Format/Spalten prüfen)"
|
|
assert status_hint(503) == "Checkmk-Serverfehler"
|
|
assert status_hint(418) == ""
|
|
|
|
|
|
async def test_request_error_carries_hint_but_brief_is_unchanged():
|
|
client = make_client(lambda r: httpx.Response(401, text="geheim"))
|
|
try:
|
|
await client.problem_services()
|
|
except CheckmkError as exc:
|
|
assert exc.brief == "HTTP 401" and exc.hint == "Zugangsdaten prüfen"
|
|
else:
|
|
raise AssertionError("expected CheckmkError")
|