patrickandClaude Sonnet 5.5
347f55cf7c
Includes atomic migrations, per-rule reminder semantics, first-run helper script and the note on the undecided UI chat idea. Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
111 lines
4.8 KiB
Python
111 lines
4.8 KiB
Python
import json
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from app.checkmk import (COLUMNS, FAILURE_UNEXPECTED, FAILURE_UNREACHABLE, HOST_COLUMNS, CheckmkClient,
|
|
CheckmkError, CheckmkUnreachable)
|
|
from app.models import HOST_SERVICE_NAME
|
|
|
|
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 host_item(name="srv1", state=1, **extra):
|
|
data = {"name": name, "state": state, "plugin_output": "DOWN - no route", "last_state_change": 42,
|
|
"scheduled_downtime_depth": 0, "acknowledged": 0}
|
|
data.update(extra)
|
|
return {"extensions": data}
|
|
|
|
|
|
async def test_problem_hosts_posts_query_and_columns():
|
|
seen = {}
|
|
|
|
def handler(request):
|
|
seen["url"] = str(request.url)
|
|
seen["body"] = json.loads(request.content)
|
|
return httpx.Response(200, json={"value": [host_item()]})
|
|
|
|
hosts = await make_client(handler).problem_hosts()
|
|
assert seen["url"] == f"{BASE}/domain-types/host/collections/all"
|
|
assert seen["body"]["columns"] == HOST_COLUMNS
|
|
assert seen["body"]["query"] == {"op": "!=", "left": "state", "right": "0"}
|
|
assert [(h.host, h.state, h.output, h.last_state_change) for h in hosts] == [("srv1", 2, "DOWN - no route", 42)]
|
|
|
|
|
|
async def test_host_states_map_to_internal_numbers():
|
|
body = {"value": [host_item("a", 1), host_item("b", 2)]}
|
|
hosts = await make_client(lambda r: httpx.Response(200, json=body)).problem_hosts()
|
|
assert [h.state for h in hosts] == [2, 3]
|
|
|
|
|
|
async def test_host_downtime_and_ack_flags():
|
|
body = {"value": [host_item(scheduled_downtime_depth=1, acknowledged=1)]}
|
|
(host,) = await make_client(lambda r: httpx.Response(200, json=body)).problem_hosts()
|
|
assert host.in_downtime and host.acknowledged
|
|
|
|
|
|
async def test_host_as_service_uses_host_service_name_and_is_hard():
|
|
(host,) = await make_client(lambda r: httpx.Response(200, json={"value": [host_item(acknowledged=1)]})
|
|
).problem_hosts()
|
|
service = host.as_service()
|
|
assert service.service == HOST_SERVICE_NAME and service.hard and service.acknowledged and not service.flapping
|
|
|
|
|
|
async def test_service_columns_include_acknowledged_and_flag_is_parsed():
|
|
assert "acknowledged" in COLUMNS
|
|
item = {"extensions": {"host_name": "h", "description": "s", "state": 2, "state_type": 1, "plugin_output": "x",
|
|
"last_state_change": 1, "scheduled_downtime_depth": 0, "is_flapping": 0,
|
|
"acknowledged": 1}}
|
|
(service,) = await make_client(lambda r: httpx.Response(200, json={"value": [item]})).problem_services()
|
|
assert service.acknowledged is True
|
|
|
|
|
|
async def test_missing_acknowledged_column_means_not_acknowledged():
|
|
item = {"extensions": {"host_name": "h", "description": "s", "state": 2}}
|
|
(service,) = await make_client(lambda r: httpx.Response(200, json={"value": [item]})).problem_services()
|
|
assert service.acknowledged is False
|
|
|
|
|
|
def failing(error):
|
|
def handler(request):
|
|
raise error
|
|
return handler
|
|
|
|
|
|
async def test_transport_error_is_unreachable():
|
|
with pytest.raises(CheckmkUnreachable) as caught:
|
|
await make_client(failing(httpx.ConnectError("boom"))).problem_services()
|
|
assert caught.value.kind == FAILURE_UNREACHABLE and caught.value.brief == "ConnectError"
|
|
|
|
|
|
async def test_timeout_is_unreachable():
|
|
with pytest.raises(CheckmkError) as caught:
|
|
await make_client(failing(httpx.ReadTimeout("slow"))).problem_hosts()
|
|
assert caught.value.kind == FAILURE_UNREACHABLE
|
|
|
|
|
|
async def test_http_5xx_is_unreachable_and_4xx_is_unexpected():
|
|
with pytest.raises(CheckmkError) as caught:
|
|
await make_client(lambda r: httpx.Response(503, text="busy")).problem_services()
|
|
assert caught.value.kind == FAILURE_UNREACHABLE and caught.value.brief == "HTTP 503"
|
|
with pytest.raises(CheckmkError) as caught:
|
|
await make_client(lambda r: httpx.Response(401, text="geheim: s3cret")).problem_services()
|
|
assert caught.value.kind == FAILURE_UNEXPECTED and caught.value.brief == "HTTP 401"
|
|
assert "s3cret" not in caught.value.brief
|
|
|
|
|
|
async def test_non_json_and_wrong_schema_are_unexpected():
|
|
with pytest.raises(CheckmkError) as caught:
|
|
await make_client(lambda r: httpx.Response(200, text="<html>")).problem_services()
|
|
assert caught.value.kind == FAILURE_UNEXPECTED
|
|
with pytest.raises(CheckmkError) as caught:
|
|
await make_client(lambda r: httpx.Response(200, json={"value": [{"extensions": {"foo": 1}}]})
|
|
).problem_services()
|
|
assert caught.value.kind == FAILURE_UNEXPECTED and caught.value.brief == "Antwortformat unerwartet"
|
|
with pytest.raises(CheckmkError):
|
|
await make_client(lambda r: httpx.Response(200, json=[1, 2])).problem_hosts()
|