checkbot/app/selftest/checks.py

from typing import Sequence

from ..checkmk import (COLUMNS, DOWNTIME_MODE_QUERY, HOST_COLUMNS, HOST_QUERY_GET, CheckmkClient, CheckmkError,
                       CheckmkUnreachable, HostState, ServiceState, downtimes_from_states)
from .constants import (CHECK_DOWNTIMES, CHECK_HOSTS, CHECK_SERVICES, CHECK_VERSION, STATUS_DEVIATED,
                        STATUS_FAILED, STATUS_OK)
from .model import CheckResult

SERVICE_REQUIRED = ("host_name", "description", "state")
SERVICE_OPTIONAL = tuple(c for c in COLUMNS if c not in SERVICE_REQUIRED)
HOST_REQUIRED = ("name", "state")
HOST_OPTIONAL = tuple(c for c in HOST_COLUMNS if c not in HOST_REQUIRED)


def _missing_columns(rows: Sequence[dict], required: Sequence[str],
                     optional: Sequence[str]) -> tuple[list[str], list[str]]:
    missing_required: set[str] = set()
    missing_optional: set[str] = set()
    for row in rows:
        missing_required |= {c for c in required if c not in row}
        missing_optional |= {c for c in optional if c not in row}
    return sorted(missing_required), sorted(missing_optional)


def _unreachable_or_unexpected(name: str, exc: CheckmkError) -> CheckResult:
    if isinstance(exc, CheckmkUnreachable):
        return CheckResult(name, STATUS_FAILED, f"nicht erreichbar ({exc.brief})")
    return CheckResult(name, STATUS_FAILED, f"Antwort unerwartet ({exc.brief})")


NOT_CHECKABLE_TEXT = "nicht prüfbar (keine Objekte)"


def _columns_result(name: str, rows: Sequence[dict], required: Sequence[str], optional: Sequence[str],
                    ok_text: str, sample: bool = False) -> CheckResult:
    missing_required, missing_optional = _missing_columns(rows, required, optional)
    if missing_required:
        return CheckResult(name, STATUS_FAILED, f"Pflichtspalte(n) fehlen: {', '.join(missing_required)}")
    if missing_optional:
        prefix = "Spalten anhand Stichprobe geprüft; " if sample else ""
        return CheckResult(name, STATUS_DEVIATED,
                           f"{prefix}optionale Spalte(n) fehlen, Fallback genutzt: {', '.join(missing_optional)}")
    return CheckResult(name, STATUS_OK, ok_text)


async def check_version(client: CheckmkClient) -> tuple[CheckResult, str]:
    try:
        info = await client.version()
    except CheckmkError as exc:
        return _unreachable_or_unexpected(CHECK_VERSION, exc), ""
    versions = info.get("versions")
    version = versions.get("checkmk") if isinstance(versions, dict) else None
    if not version:
        return CheckResult(CHECK_VERSION, STATUS_DEVIATED, "Feld versions.checkmk fehlt in der Antwort"), ""
    return CheckResult(CHECK_VERSION, STATUS_OK, f"Checkmk {version}"), str(version)


async def check_services(client: CheckmkClient) -> tuple[CheckResult, list[ServiceState]]:
    try:
        rows, services = await client.selftest_services()
    except CheckmkError as exc:
        return _unreachable_or_unexpected(CHECK_SERVICES, exc), []
    if rows:
        return _columns_result(CHECK_SERVICES, rows, SERVICE_REQUIRED, SERVICE_OPTIONAL,
                               f"{len(services)} Probleme gelesen, alle Spalten vorhanden"), services
    try:
        sample = await client.sample_service()
    except CheckmkError as exc:
        return _unreachable_or_unexpected(CHECK_SERVICES, exc), services
    if sample is None:
        return CheckResult(CHECK_SERVICES, STATUS_OK, NOT_CHECKABLE_TEXT), services
    return _columns_result(CHECK_SERVICES, [sample], SERVICE_REQUIRED, SERVICE_OPTIONAL,
                           "0 Probleme gelesen, Spalten anhand Stichprobe geprüft", sample=True), services


async def check_hosts(client: CheckmkClient) -> tuple[CheckResult, list[HostState]]:
    try:
        rows, hosts, mode = await client.selftest_hosts()
    except CheckmkError as exc:
        return _unreachable_or_unexpected(CHECK_HOSTS, exc), []
    notes = []
    if mode == HOST_QUERY_GET:
        notes.append("GET-Fallback verwendet, POST wurde abgelehnt")
    sample_rows, sample_used = rows, False
    if not sample_rows:
        try:
            sample = await client.sample_host()
        except CheckmkError as exc:
            return _unreachable_or_unexpected(CHECK_HOSTS, exc), hosts
        if sample is None:
            notes.append(NOT_CHECKABLE_TEXT)
            status = STATUS_DEVIATED if mode == HOST_QUERY_GET else STATUS_OK
            return CheckResult(CHECK_HOSTS, status, "; ".join(notes)), hosts
        sample_rows, sample_used = [sample], True
    missing_required, missing_optional = _missing_columns(sample_rows, HOST_REQUIRED, HOST_OPTIONAL)
    if missing_required:
        return CheckResult(CHECK_HOSTS, STATUS_FAILED,
                           f"Pflichtspalte(n) fehlen: {', '.join(missing_required)}"), hosts
    if missing_optional:
        prefix = "Spalten anhand Stichprobe geprüft; " if sample_used else ""
        notes.append(f"{prefix}optionale Spalte(n) fehlen, Fallback genutzt: {', '.join(missing_optional)}")
    if notes:
        return CheckResult(CHECK_HOSTS, STATUS_DEVIATED, "; ".join(notes)), hosts
    text = ("0 Probleme gelesen, Spalten anhand Stichprobe geprüft" if sample_used else
           f"{len(hosts)} Probleme gelesen, alle Spalten vorhanden")
    return CheckResult(CHECK_HOSTS, STATUS_OK, text), hosts


async def check_downtimes(client: CheckmkClient, services: Sequence[ServiceState],
                          hosts: Sequence[HostState]) -> CheckResult:
    try:
        downtimes, mode = await client.downtimes_with_fallback()
    except CheckmkUnreachable as exc:
        return CheckResult(CHECK_DOWNTIMES, STATUS_FAILED, f"nicht erreichbar ({exc.brief})")
    except CheckmkError as exc:
        derived = downtimes_from_states(services, hosts)
        return CheckResult(CHECK_DOWNTIMES, STATUS_DEVIATED,
                           f"Downtime-Endpunkt abgelehnt ({exc.brief}), Fallback über "
                           f"scheduled_downtime_depth: {len(derived)} Downtime(s) abgeleitet")
    if mode == DOWNTIME_MODE_QUERY:
        return CheckResult(CHECK_DOWNTIMES, STATUS_DEVIATED,
                           f"downtime_type abgelehnt, Fallback über query verwendet, {len(downtimes)} gelesen")
    return CheckResult(CHECK_DOWNTIMES, STATUS_OK, f"{len(downtimes)} Downtime(s) gelesen")