75 lines
2.6 KiB
Python
75 lines
2.6 KiB
Python
import json
|
|
from datetime import datetime, timezone
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from app.downtime import CheckmkDowntimeClient, DowntimeError
|
|
|
|
SECRET = "dt-secret-123"
|
|
START = datetime(2026, 11, 6, 8, 0, tzinfo=timezone.utc)
|
|
END = datetime(2026, 11, 6, 10, 0, tzinfo=timezone.utc)
|
|
|
|
|
|
def make(handler):
|
|
return CheckmkDowntimeClient("https://cmk.example.de/", "prod", "dtuser", SECRET,
|
|
transport=httpx.MockTransport(handler))
|
|
|
|
|
|
async def test_payload_and_auth():
|
|
seen = {}
|
|
|
|
def handler(request):
|
|
seen["method"], seen["url"] = request.method, str(request.url)
|
|
seen["auth"] = request.headers["authorization"]
|
|
seen["body"] = json.loads(request.content)
|
|
return httpx.Response(204)
|
|
|
|
await make(handler).create_service_downtime("web01", "CPU load", START, END, "Wartung")
|
|
assert seen["method"] == "POST"
|
|
assert seen["url"] == "https://cmk.example.de/prod/check_mk/api/1.0/domain-types/downtime/collections/service"
|
|
assert seen["auth"] == f"Bearer dtuser {SECRET}"
|
|
assert seen["body"] == {
|
|
"downtime_type": "service", "host_name": "web01", "service_descriptions": ["CPU load"],
|
|
"start_time": "2026-11-06T08:00:00Z", "end_time": "2026-11-06T10:00:00Z", "comment": "Wartung",
|
|
}
|
|
|
|
|
|
async def test_http_error_message_has_no_secret():
|
|
def handler(request):
|
|
return httpx.Response(403, json={"title": "Forbidden", "detail": f"denied for {SECRET}"})
|
|
|
|
with pytest.raises(DowntimeError) as info:
|
|
await make(handler).create_service_downtime("web01", "CPU load", START, END, "x")
|
|
assert "403" in str(info.value) and SECRET not in str(info.value)
|
|
|
|
|
|
async def test_network_error_has_no_secret():
|
|
def handler(request):
|
|
raise httpx.ConnectError(f"cannot connect with {SECRET}")
|
|
|
|
with pytest.raises(DowntimeError) as info:
|
|
await make(handler).version()
|
|
assert SECRET not in str(info.value) and info.value.__cause__ is None
|
|
|
|
|
|
async def test_non_json_error_body():
|
|
def handler(request):
|
|
return httpx.Response(500, text=f"<html>{SECRET}</html>")
|
|
|
|
with pytest.raises(DowntimeError) as info:
|
|
await make(handler).version()
|
|
assert str(info.value) == "HTTP 500:" or str(info.value).startswith("HTTP 500")
|
|
assert SECRET not in str(info.value)
|
|
|
|
|
|
async def test_version_is_read_only_get():
|
|
seen = {}
|
|
|
|
def handler(request):
|
|
seen["method"], seen["path"] = request.method, request.url.path
|
|
return httpx.Response(200, json={"versions": {"checkmk": "2.3.0"}})
|
|
|
|
info = await make(handler).version()
|
|
assert seen == {"method": "GET", "path": "/prod/check_mk/api/1.0/version"} and info["versions"]
|