checkbot/tests/test_poller.py

from app.checkmk import ServiceState
from app.db import Database
from app.models import (KIND_FLAPPING, KIND_PROBLEM, KIND_RECOVERY, SUPPRESS_DOWNTIME, STATE_OK)
from app.poller import Poller, PollerStatus, StateStore, compute_events
from app.poller.events import FLAP_QUIET_SECONDS

NOW = 100_000


def svc(state=2, host="h1", service="cpu", hard=True, output="boom", changed=NOW - 3600,
        downtime=False, flapping=False):
    return ServiceState(host, service, state, hard, output, changed, downtime, flapping)


def run(previous, current, now=NOW, min_duration=0):
    events, rows = compute_events(previous, current, now, lambda *_: min_duration)
    return events, {row.key: row for row in rows}


def test_new_problem_emits_event_and_state():
    events, rows = run({}, [svc()])
    assert [(e.kind, e.state, e.host, e.service) for e in events] == [(KIND_PROBLEM, 2, "h1", "cpu")]
    assert rows[("h1", "cpu")].reported_state == 2


def test_unchanged_problem_is_silent():
    _, rows = run({}, [svc()])
    events, _ = run(rows, [svc()], now=NOW + 30)
    assert events == []


def test_warn_to_crit_emits_event():
    _, rows = run({}, [svc(state=1)])
    events, _ = run(rows, [svc(state=2)], now=NOW + 30)
    assert [(e.kind, e.state) for e in events] == [(KIND_PROBLEM, 2)]


def test_recovery_when_service_disappears():
    _, rows = run({}, [svc()])
    events, new_rows = run(rows, [], now=NOW + 30)
    assert [(e.kind, e.state) for e in events] == [(KIND_RECOVERY, STATE_OK)]
    assert new_rows == {}


def test_soft_state_ignored():
    events, rows = run({}, [svc(hard=False)])
    assert events == [] and rows == {}


def test_soft_state_keeps_previous_row_and_gives_no_recovery():
    _, rows = run({}, [svc(state=2)])
    events, new_rows = run(rows, [svc(state=1, hard=False)], now=NOW + 30)
    assert events == []
    assert new_rows[("h1", "cpu")].reported_state == 2


def test_never_reported_service_has_no_recovery():
    _, rows = run({}, [svc(hard=False)])
    events, _ = run(rows, [], now=NOW + 30)
    assert events == []


def test_min_duration_defers_then_reports():
    current = [svc(changed=NOW - 10)]
    events, rows = run({}, current, min_duration=60)
    assert events == []
    events, _ = run(rows, current, now=NOW + 60, min_duration=60)
    assert [e.kind for e in events] == [KIND_PROBLEM]


def test_deferred_problem_vanishing_gives_no_recovery():
    _, rows = run({}, [svc(changed=NOW - 10)], min_duration=60)
    events, _ = run(rows, [], now=NOW + 30, min_duration=60)
    assert events == []


def test_downtime_event_is_suppressed():
    events, _ = run({}, [svc(downtime=True)])
    assert events[0].suppressed is True
    assert events[0].suppress_reason == SUPPRESS_DOWNTIME


def test_flapping_reported_once():
    first, rows = run({}, [svc(flapping=True)])
    assert [e.kind for e in first] == [KIND_FLAPPING]
    second, rows = run(rows, [svc(state=1, flapping=True)], now=NOW + 30)
    assert second == []
    third, _ = run(rows, [svc(state=2, flapping=True)], now=NOW + 60)
    assert third == []


def test_flapping_service_vanishing_gives_no_recovery_and_no_repeat():
    _, rows = run({}, [svc(flapping=True)])
    events, rows = run(rows, [], now=NOW + 30)
    assert events == []
    events, rows = run(rows, [svc(flapping=True)], now=NOW + 60)
    assert events == []


def test_flapping_state_dropped_after_quiet_period():
    _, rows = run({}, [svc(flapping=True)])
    _, rows = run(rows, [], now=NOW + 30)
    _, rows = run(rows, [], now=NOW + 30 + FLAP_QUIET_SECONDS)
    assert rows == {}


def test_flapping_end_with_same_reported_state_is_silent():
    _, rows = run({}, [svc(state=2)])
    _, rows = run(rows, [svc(state=2, flapping=True)], now=NOW + 30)
    events, _ = run(rows, [svc(state=2)], now=NOW + 60)
    assert events == []


def test_first_run_reports_existing_problems():
    events, rows = run({}, [svc(host="a"), svc(host="b", state=1)])
    assert len(events) == 2 and len(rows) == 2


class FakeClient:
    def __init__(self, problems=None, error=None):
        self._problems, self._error, self.closed = problems or [], error, False

    async def problem_services(self):
        if self._error:
            raise self._error
        return self._problems

    async def problem_hosts(self):
        return []

    async def aclose(self):
        self.closed = True


class MemoryStore:
    def get_int(self, key, default, minimum=0):
        return max(default, minimum)


def make_poller(db, client, sink):
    state = StateStore(db)
    poller = Poller(MemoryStore(), state, sink, PollerStatus(), client_factory=lambda: client,
                    clock=lambda: NOW)
    return poller, state


async def test_api_failure_keeps_state_and_emits_nothing():
    db = Database(":memory:")
    collected = []

    async def sink(events):
        collected.extend(events)

    poller, state = make_poller(db, FakeClient([svc()]), sink)
    await poller.run_cycle()
    assert len(collected) == 1
    stored = state.load()

    poller, _ = make_poller(db, FakeClient(error=RuntimeError("down")), sink)
    await poller.run_cycle()
    assert len(collected) == 1
    assert state.load() == stored
    assert poller.status.last_error and "down" in poller.status.last_error


async def test_cycle_records_success_and_state_log():
    db = Database(":memory:")

    async def sink(events):
        pass

    poller, state = make_poller(db, FakeClient([svc(state=1)]), sink)
    await poller.run_cycle()
    assert poller.status.last_success == NOW
    poller, _ = make_poller(db, FakeClient([svc(state=2)]), sink)
    await poller.run_cycle()
    changes = state.recent_changes("h1", "cpu")
    assert [c.state for c in changes] == [2, 1]


async def test_unconfigured_cycle_is_skipped():
    db = Database(":memory:")
    calls = []

    async def sink(events):
        calls.append(events)

    state = StateStore(db)
    poller = Poller(MemoryStore(), state, sink, PollerStatus(), client_factory=lambda: None)
    await poller.run_cycle()
    assert calls == [] and poller.status.last_success is None