import httpx from app.checkmk import (FAILURE_UNEXPECTED, FAILURE_UNREACHABLE, CheckmkError, CheckmkUnreachable, HostState, ServiceState) from app.db import Database from app.models import HOST_SERVICE_NAME, KIND_PROBLEM, KIND_RECOVERY from app.poller import (FAILURE_THRESHOLD, BaselineGuard, OutageTracker, Poller, PollerStatus, StateStore, classify) NOW = 100_000 SECRET = "geheimes-token-123" def svc(state=2, host="h1", service="cpu", changed=NOW - 3600, output="boom"): return ServiceState(host, service, state, True, output, changed, False, False) def down_host(name="h1", state=1): return HostState(name, state, "no route", NOW - 600, False, False) class FakeClient: def __init__(self, services=(), hosts=(), error=None): self.services, self.hosts, self.error = list(services), list(hosts), error async def problem_services(self): if self.error: raise self.error return self.services async def problem_hosts(self): return self.hosts async def aclose(self): pass class Notifier: def __init__(self): self.outages, self.restored_notices, self.baselines = [], [], [] async def outage(self, notice): self.outages.append(notice) async def restored(self, notice): self.restored_notices.append(notice) async def baseline(self, rows): self.baselines.append(rows) class MemoryStore: def get_int(self, key, default, minimum=0): return max(default, minimum) class Env: def __init__(self, client=None, baseline=True): self.db = Database(":memory:") self.state = StateStore(self.db) self.guard = BaselineGuard(self.db) self.notifier = Notifier() self.events = [] self.client = client or FakeClient() self.now = NOW self.poller = Poller(MemoryStore(), self.state, self.sink, PollerStatus(), client_factory=lambda: self.client, clock=lambda: self.now, notifier=self.notifier, baseline=self.guard if baseline else None) async def sink(self, events): self.events.extend(events) async def cycle(self, advance=30): self.now += advance await self.poller.run_cycle() def marker(self, key): return bool(self.db.query("SELECT 1 FROM app_state WHERE key=?", (key,))) async def test_first_cycle_is_recorded_silently_with_one_summary(): env = Env(FakeClient([svc(2), svc(1, service="disk")], [down_host()])) await env.cycle() assert env.events == [] assert len(env.notifier.baselines) == 1 assert sorted((r.service, r.state) for r in env.notifier.baselines[0]) == [ (HOST_SERVICE_NAME, 2), ("cpu", 2), ("disk", 1)] stored = env.state.load() assert len(stored) == 3 and all(r.reported_state == r.state for r in stored.values()) assert env.marker("baseline.done") async def test_following_cycles_are_normal_and_do_not_repeat_the_summary(): env = Env(FakeClient([svc(2)])) await env.cycle() env.client.services = [svc(2), svc(2, service="new")] await env.cycle() assert [e.service for e in env.events] == ["new"] and len(env.notifier.baselines) == 1 async def test_clean_first_cycle_still_sets_marker_and_sends_empty_summary(): env = Env(FakeClient([])) await env.cycle() assert env.notifier.baselines == [[]] and env.marker("baseline.done") env.client.services = [svc(2)] await env.cycle() assert [e.kind for e in env.events] == [KIND_PROBLEM] async def test_baseline_waits_for_first_successful_cycle(): env = Env(FakeClient(error=CheckmkUnreachable("x", brief="ConnectError"))) await env.cycle() assert env.notifier.baselines == [] and not env.marker("baseline.done") env.client = FakeClient([svc(2)]) await env.cycle() assert len(env.notifier.baselines) == 1 and env.events == [] async def test_new_request_records_again_silently_and_without_recoveries(): env = Env(FakeClient([svc(2, service="old")])) await env.cycle() env.guard.request() assert env.marker("baseline.requested") env.client.services = [svc(1, service="fresh")] await env.cycle() assert env.events == [] and len(env.notifier.baselines) == 2 assert [r.service for r in env.notifier.baselines[1]] == ["fresh"] assert set(env.state.load()) == {("h1", "fresh")} assert not env.marker("baseline.requested") await env.cycle() assert env.events == [] and len(env.notifier.baselines) == 2 async def test_recovery_of_baseline_problem_is_emitted(): env = Env(FakeClient([svc(2)])) await env.cycle() env.client.services = [] await env.cycle() assert [(e.kind, e.service) for e in env.events] == [(KIND_RECOVERY, "cpu")] async def test_no_baseline_without_guard(): env = Env(FakeClient([svc(2)]), baseline=False) await env.cycle() assert [e.kind for e in env.events] == [KIND_PROBLEM] and env.notifier.baselines == [] def test_guard_needed_logic(): db = Database(":memory:") guard = BaselineGuard(db) assert guard.needed(state_empty=True) and not guard.needed(state_empty=False) db.execute_many(guard.completion()) assert not guard.needed(state_empty=True) guard.request() assert guard.needed(state_empty=False) async def test_failures_below_threshold_stay_silent(): env = Env(FakeClient(error=CheckmkUnreachable("x", brief="ConnectError")), baseline=False) for _ in range(FAILURE_THRESHOLD - 1): await env.cycle() assert env.notifier.outages == [] async def test_unreachable_outage_is_announced_once_with_first_failure_time(): env = Env(FakeClient(error=CheckmkUnreachable("x", brief="ConnectError")), baseline=False) for _ in range(FAILURE_THRESHOLD + 2): await env.cycle(advance=30) assert len(env.notifier.outages) == 1 notice = env.notifier.outages[0] assert notice.kind == FAILURE_UNREACHABLE and notice.since == NOW + 30 and notice.brief == "ConnectError" async def test_unexpected_answer_is_a_different_notice_kind(): env = Env(FakeClient(error=CheckmkError(f"HTTP 403: {SECRET}", brief="HTTP 403")), baseline=False) for _ in range(FAILURE_THRESHOLD): await env.cycle() (notice,) = env.notifier.outages assert notice.kind == FAILURE_UNEXPECTED and notice.brief == "HTTP 403" and SECRET not in repr(notice) async def test_restoration_is_announced_once_after_announced_outage(): env = Env(FakeClient(error=CheckmkUnreachable("x", brief="ConnectError")), baseline=False) for _ in range(FAILURE_THRESHOLD): await env.cycle() env.client = FakeClient() await env.cycle(advance=300) await env.cycle() (notice,) = env.notifier.restored_notices assert notice.since == NOW + 30 and notice.now == env.now - 30 async def test_short_outage_without_announcement_has_no_restoration_notice(): env = Env(FakeClient(error=CheckmkUnreachable("x", brief="ConnectError")), baseline=False) await env.cycle() env.client = FakeClient() await env.cycle() assert env.notifier.restored_notices == [] async def test_no_recoveries_during_outage_and_state_is_kept(): env = Env(FakeClient([svc(2)])) await env.cycle() stored = env.state.load() env.client = FakeClient(error=CheckmkUnreachable("x", brief="ConnectError")) for _ in range(FAILURE_THRESHOLD + 1): await env.cycle() assert env.events == [] and env.state.load() == stored async def test_notifier_failure_does_not_break_cycle(): env = Env(FakeClient(error=CheckmkUnreachable("x", brief="ConnectError")), baseline=False) async def broken(notice): raise RuntimeError("chat kaputt") env.notifier.outage = broken for _ in range(FAILURE_THRESHOLD): await env.cycle() assert env.poller.status.last_error def test_tracker_counts_consecutive_failures_only(): tracker = OutageTracker(threshold=3) assert tracker.failed(FAILURE_UNREACHABLE, "a", 10) is None assert tracker.succeeded(20) is None assert tracker.failed(FAILURE_UNREACHABLE, "a", 30) is None assert tracker.failed(FAILURE_UNREACHABLE, "a", 40) is None notice = tracker.failed(FAILURE_UNEXPECTED, "b", 50) assert notice.since == 30 and notice.kind == FAILURE_UNEXPECTED assert tracker.failed(FAILURE_UNEXPECTED, "b", 60) is None restored = tracker.succeeded(70) assert (restored.since, restored.now) == (30, 70) and tracker.succeeded(80) is None def test_classify_maps_exceptions_to_kind_and_safe_brief(): assert classify(CheckmkUnreachable("x", brief="HTTP 502")) == (FAILURE_UNREACHABLE, "HTTP 502") assert classify(CheckmkError("x", brief="HTTP 401")) == (FAILURE_UNEXPECTED, "HTTP 401") assert classify(RuntimeError(SECRET)) == (FAILURE_UNEXPECTED, "RuntimeError") assert classify(httpx.ConnectError(SECRET)) == (FAILURE_UNEXPECTED, "ConnectError")