63 lines
2.3 KiB
Python
63 lines
2.3 KiB
Python
import json
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from app.checkmk import CheckmkClient, CheckmkError
|
|
|
|
|
|
def make_client(handler):
|
|
return CheckmkClient("https://cmk.local", "prod", "bot", "s3cret",
|
|
transport=httpx.MockTransport(handler))
|
|
|
|
|
|
async def test_problem_services_parsed_and_request_shape():
|
|
seen = {}
|
|
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
seen["url"] = str(request.url)
|
|
seen["auth"] = request.headers["authorization"]
|
|
seen["body"] = json.loads(request.content)
|
|
return httpx.Response(200, json={"value": [
|
|
{"extensions": {
|
|
"host_name": "web01", "description": "CPU load", "state": 2,
|
|
"state_type": 1, "plugin_output": "CRIT - load 9",
|
|
"last_state_change": 1730000000, "scheduled_downtime_depth": 1,
|
|
"is_flapping": 0,
|
|
}},
|
|
{"extensions": {
|
|
"host_name": "db01", "description": "Disk", "state": 1,
|
|
"state_type": 0, "plugin_output": "WARN", "last_state_change": 5,
|
|
"scheduled_downtime_depth": 0, "is_flapping": 1,
|
|
}},
|
|
]})
|
|
|
|
services = await make_client(handler).problem_services()
|
|
|
|
assert seen["url"] == "https://cmk.local/prod/check_mk/api/1.0/domain-types/service/collections/all"
|
|
assert seen["auth"] == "Bearer bot s3cret"
|
|
assert seen["body"]["query"] == {"op": "!=", "left": "state", "right": "0"}
|
|
web, db = services
|
|
assert (web.host, web.service, web.state, web.hard, web.in_downtime, web.flapping) == \
|
|
("web01", "CPU load", 2, True, True, False)
|
|
assert (db.hard, db.in_downtime, db.flapping) == (False, False, True)
|
|
|
|
|
|
async def test_version():
|
|
client = make_client(lambda r: httpx.Response(200, json={"versions": {"checkmk": "2.3.0p5"}}))
|
|
assert (await client.version())["versions"]["checkmk"] == "2.3.0p5"
|
|
|
|
|
|
async def test_http_error_raises():
|
|
client = make_client(lambda r: httpx.Response(401, text="unauthorized"))
|
|
with pytest.raises(CheckmkError, match="401"):
|
|
await client.version()
|
|
|
|
|
|
async def test_transport_error_raises():
|
|
def handler(request):
|
|
raise httpx.ConnectError("down")
|
|
|
|
with pytest.raises(CheckmkError, match="ConnectError"):
|
|
await make_client(handler).version()
|