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>
100 lines
3.8 KiB
Python
100 lines
3.8 KiB
Python
import json
|
|
|
|
import httpx
|
|
|
|
from app.checkmk import CheckmkClient
|
|
|
|
BASE = "https://cmk.local/prod/check_mk/api/1.0"
|
|
|
|
|
|
def make_client(handler):
|
|
return CheckmkClient("https://cmk.local", "prod", "bot", "s3cret", transport=httpx.MockTransport(handler))
|
|
|
|
|
|
def item(host="web01", service="CPU load", state=2):
|
|
return {"extensions": {"host_name": host, "description": service, "state": state, "state_type": 1,
|
|
"plugin_output": "x", "last_state_change": 1, "scheduled_downtime_depth": 0,
|
|
"is_flapping": 0}}
|
|
|
|
|
|
async def test_service_status_filters_and_returns():
|
|
seen = {}
|
|
|
|
def handler(request):
|
|
seen["url"] = str(request.url)
|
|
seen["body"] = json.loads(request.content)
|
|
return httpx.Response(200, json={"value": [item()]})
|
|
|
|
result = await make_client(handler).service_status("web01", "CPU load")
|
|
assert seen["url"] == f"{BASE}/domain-types/service/collections/all"
|
|
assert seen["body"]["query"] == {"op": "and", "expr": [
|
|
{"op": "=", "left": "host_name", "right": "web01"},
|
|
{"op": "=", "left": "description", "right": "CPU load"}]}
|
|
assert result.service == "CPU load" and result.state == 2
|
|
|
|
|
|
async def test_service_status_missing():
|
|
client = make_client(lambda r: httpx.Response(200, json={"value": []}))
|
|
assert await client.service_status("a", "b") is None
|
|
|
|
|
|
async def test_host_services():
|
|
seen = {}
|
|
|
|
def handler(request):
|
|
seen["body"] = json.loads(request.content)
|
|
return httpx.Response(200, json={"value": [item(state=0), item(service="Disk", state=1)]})
|
|
|
|
result = await make_client(handler).host_services("web01")
|
|
assert seen["body"]["query"] == {"op": "=", "left": "host_name", "right": "web01"}
|
|
assert [s.service for s in result] == ["CPU load", "Disk"]
|
|
|
|
|
|
async def test_problems_alias():
|
|
client = make_client(lambda r: httpx.Response(200, json={"value": [item()]}))
|
|
assert len(await client.problems()) == 1
|
|
|
|
|
|
async def test_service_status_accepts_members_instead_of_value():
|
|
body = {"members": [item()]}
|
|
result = await make_client(lambda r: httpx.Response(200, json=body)).service_status("web01", "CPU load")
|
|
assert result.service == "CPU load" and result.state == 2
|
|
|
|
|
|
async def test_state_type_as_string_hard_and_soft():
|
|
def hard_soft(state_type_value):
|
|
extensions = item()["extensions"]
|
|
extensions["state_type"] = state_type_value
|
|
return {"extensions": extensions}
|
|
|
|
client = make_client(lambda r: httpx.Response(200, json={"value": [hard_soft("hard")]}))
|
|
(service,) = await client.problem_services()
|
|
assert service.hard is True
|
|
|
|
client = make_client(lambda r: httpx.Response(200, json={"value": [hard_soft("SOFT")]}))
|
|
(service,) = await client.problem_services()
|
|
assert service.hard is False
|
|
|
|
client = make_client(lambda r: httpx.Response(200, json={"value": [hard_soft(None)]}))
|
|
(service,) = await client.problem_services()
|
|
assert service.hard is False
|
|
|
|
|
|
async def test_downtimes_all_and_by_host():
|
|
seen = []
|
|
|
|
def handler(request):
|
|
seen.append(request)
|
|
return httpx.Response(200, json={"value": [{"extensions": {
|
|
"host_name": "web01", "service_description": "", "start_time": "2026-01-01T00:00:00Z",
|
|
"end_time": "2026-01-02T00:00:00Z", "comment": "Wartung", "author": "admin"}}]})
|
|
|
|
client = make_client(handler)
|
|
result = await client.downtimes()
|
|
assert all(r.method == "GET" and r.url.path.endswith("/domain-types/downtime/collections/all") for r in seen)
|
|
assert [r.url.params["downtime_type"] for r in seen] == ["host", "service"]
|
|
assert all("host_name" not in r.url.params for r in seen)
|
|
assert len(result) == 2 and result[0].service is None and result[0].comment == "Wartung"
|
|
assert await client.downtimes("WEB01") == result
|
|
assert await client.downtimes("db01") == []
|