checkbot/tests/test_selftest.py

import json
import os

import httpx

from app.checkmk import CheckmkClient
from app.crypto import SecretBox
from app.db import Database
from app.selftest import (CHECK_DOWNTIMES, CHECK_HOSTS, CHECK_SERVICES, CHECK_VERSION, STATUS_DEVIATED,
                          STATUS_FAILED, STATUS_OK, SelftestNotConfigured, SelftestRepository, SelftestRun,
                          SelftestService, message_for)
from app.selftest.checks import check_downtimes, check_hosts, check_services, check_version
from app.selftest.model import CheckResult
from app.settings import SettingsStore


def make_client(handler):
    return CheckmkClient("https://cmk.local", "prod", "bot", "s3cret", transport=httpx.MockTransport(handler))


def service_row(host="h1", service="s1", state=2, **extra):
    data = {"host_name": host, "description": service, "state": state, "state_type": 1, "plugin_output": "x",
            "last_state_change": 1, "scheduled_downtime_depth": 0, "is_flapping": 0, "acknowledged": 0}
    data.update(extra)
    return {"extensions": data}


def host_row(name="h2", state=1, **extra):
    data = {"name": name, "state": state, "plugin_output": "x", "last_state_change": 1,
            "scheduled_downtime_depth": 0, "acknowledged": 0}
    data.update(extra)
    return {"extensions": data}


def good_handler(request):
    if request.url.path.endswith("/version"):
        return httpx.Response(200, json={"versions": {"checkmk": "2.5.0p1"}})
    if request.url.path.endswith("/service/collections/all"):
        return httpx.Response(200, json={"value": [service_row()]})
    if request.url.path.endswith("/host/collections/all"):
        return httpx.Response(200, json={"value": [host_row()]})
    if request.url.path.endswith("/downtime/collections/all"):
        return httpx.Response(200, json={"value": []})
    raise AssertionError(request.url.path)


async def test_check_version_ok_deviated_and_failed():
    ok, version = await check_version(make_client(good_handler))
    assert ok.status == STATUS_OK and ok.name == CHECK_VERSION and version == "2.5.0p1"

    deviated, version = await check_version(make_client(lambda r: httpx.Response(200, json={"versions": {}})))
    assert deviated.status == STATUS_DEVIATED and version == ""

    failed, version = await check_version(make_client(lambda r: httpx.Response(401, text="x")))
    assert failed.status == STATUS_FAILED and version == ""


async def test_check_services_reports_missing_columns():
    client = make_client(lambda r: httpx.Response(200, json={"value": [service_row()]}))
    result, services = await check_services(client)
    assert result.status == STATUS_OK and len(services) == 1

    missing_optional = make_client(lambda r: httpx.Response(
        200, json={"value": [{"extensions": {"host_name": "h", "description": "s", "state": 2}}]}))
    result, services = await check_services(missing_optional)
    assert result.status == STATUS_DEVIATED and "last_state_change" in result.text

    missing_required = make_client(lambda r: httpx.Response(
        200, json={"value": [{"extensions": {"description": "s", "state": 2}}]}))
    result, services = await check_services(missing_required)
    assert result.status == STATUS_FAILED and "host_name" in result.text


async def test_check_hosts_reports_get_fallback_as_deviated():
    def handler(request):
        if request.method == "POST":
            return httpx.Response(422, text="x")
        return httpx.Response(200, json={"value": [host_row(state=2)]})

    result, hosts = await check_hosts(make_client(handler))
    assert result.status == STATUS_DEVIATED and "GET-Fallback" in result.text and len(hosts) == 1


async def test_check_downtimes_falls_back_to_query_then_to_derived():
    def query_fallback(request):
        if "downtime_type" in request.url.params:
            return httpx.Response(422, text="x")
        return httpx.Response(200, json={"value": []})

    result = await check_downtimes(make_client(query_fallback), [], [])
    assert result.status == STATUS_DEVIATED and "query" in result.text

    def all_fail(request):
        return httpx.Response(404, text="x")

    from app.checkmk import ServiceState
    services = [ServiceState("h1", "s1", 2, True, "", 1, True, False)]
    result = await check_downtimes(make_client(all_fail), services, [])
    assert result.status == STATUS_DEVIATED and "abgeleitet" in result.text and "1 Downtime" in result.text


async def test_check_services_probes_sample_when_no_problems():
    def handler(request):
        body = json.loads(request.content)
        if "query" in body:
            return httpx.Response(200, json={"value": []})
        return httpx.Response(200, json={"value": [service_row()]})

    result, services = await check_services(make_client(handler))
    assert result.status == STATUS_OK and "Stichprobe" in result.text and services == []


async def test_check_services_sample_detects_missing_column():
    def handler(request):
        body = json.loads(request.content)
        if "query" in body:
            return httpx.Response(200, json={"value": []})
        return httpx.Response(200, json={"value": [
            {"extensions": {"host_name": "h", "description": "s", "state": 2}}]})

    result, services = await check_services(make_client(handler))
    assert result.status == STATUS_DEVIATED and "last_state_change" in result.text


async def test_check_services_empty_checkmk_is_ok_not_checkable():
    def handler(request):
        return httpx.Response(200, json={"value": []})

    result, services = await check_services(make_client(handler))
    assert result.status == STATUS_OK and "nicht prüfbar" in result.text and services == []


async def test_check_hosts_probes_sample_when_no_problems():
    def handler(request):
        body = json.loads(request.content)
        if "query" in body:
            return httpx.Response(200, json={"value": []})
        return httpx.Response(200, json={"value": [host_row()]})

    result, hosts = await check_hosts(make_client(handler))
    assert result.status == STATUS_OK and "Stichprobe" in result.text and hosts == []


async def test_check_hosts_empty_checkmk_is_ok_not_checkable():
    def handler(request):
        return httpx.Response(200, json={"value": []})

    result, hosts = await check_hosts(make_client(handler))
    assert result.status == STATUS_OK and "nicht prüfbar" in result.text and hosts == []


def make_env():
    db = Database(":memory:")
    store = SettingsStore(db, SecretBox(os.urandom(32)))
    return db, store


def configure(store):
    from app.constants import CHECKMK_SECRET, CHECKMK_SITE, CHECKMK_URL, CHECKMK_USER
    store.set(CHECKMK_URL, "https://cmk.local", "t")
    store.set(CHECKMK_SITE, "prod", "t")
    store.set(CHECKMK_USER, "bot", "t")
    store.set(CHECKMK_SECRET, "s3cret", "t", secret=True)


async def test_service_run_not_configured():
    db, store = make_env()
    service = SelftestService(store, SelftestRepository(db))
    try:
        await service.run()
    except SelftestNotConfigured:
        pass
    else:
        raise AssertionError("expected SelftestNotConfigured")


async def test_service_run_persists_and_detects_version_change(monkeypatch):
    db, store = make_env()
    configure(store)
    repo = SelftestRepository(db)
    service = SelftestService(store, repo,
                              client_factory=lambda s: make_client(good_handler), clock=lambda: 1_800_000_000.0)
    run = await service.run()
    assert run.version == "2.5.0p1" and run.version_changed is False
    assert repo.last_run().version == "2.5.0p1"

    def newer(request):
        if request.url.path.endswith("/version"):
            return httpx.Response(200, json={"versions": {"checkmk": "2.5.0p2"}})
        return good_handler(request)

    service2 = SelftestService(store, repo, client_factory=lambda s: make_client(newer),
                               clock=lambda: 1_800_000_100.0)
    run2 = await service2.run()
    assert run2.version == "2.5.0p2" and run2.version_changed is True


def test_overall_status_priority():
    ok = CheckResult("a", STATUS_OK, "")
    deviated = CheckResult("b", STATUS_DEVIATED, "")
    failed = CheckResult("c", STATUS_FAILED, "")
    assert SelftestRun(1, "v", [ok]).overall_status == STATUS_OK
    assert SelftestRun(1, "v", [ok, deviated]).overall_status == STATUS_DEVIATED
    assert SelftestRun(1, "v", [ok, deviated, failed]).overall_status == STATUS_FAILED


def run_with(status):
    return SelftestRun(1, "2.5.0", [CheckResult(CHECK_VERSION, status, "x")])


def test_message_for_no_previous_is_silent():
    assert message_for(None, run_with(STATUS_OK)) is None


def test_message_for_same_status_is_silent():
    assert message_for(run_with(STATUS_OK), run_with(STATUS_OK)) is None


def test_message_for_status_change_to_failed_and_back_to_ok():
    text = message_for(run_with(STATUS_OK), run_with(STATUS_FAILED))
    assert text is not None and "fehler" in text
    text = message_for(run_with(STATUS_FAILED), run_with(STATUS_OK))
    assert text == "Selbsttest wieder ok"


def test_message_for_version_change():
    previous = SelftestRun(1, "2.4.0", [CheckResult(CHECK_VERSION, STATUS_OK, "x")])
    current = SelftestRun(2, "2.5.0", [CheckResult(CHECK_VERSION, STATUS_OK, "x")], version_changed=True)
    text = message_for(previous, current)
    assert text == "Checkmk-Version von 2.4.0 auf 2.5.0, Selbsttest: ok"