patrickandClaude Sonnet 5.5
b937a57602
Der Runner prüft die Checkmk-Version alle 5 Minuten und startet bei einem Wechsel sofort einen Selbsttest (Cooldown, Erstlauf ausgenommen). Die Spaltenprüfung fragt bei leerer Problemliste eine ungefilterte Stichprobe ab; leeres Checkmk gilt als nicht prüfbar, nicht als Fehler. Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
205 lines
7.0 KiB
Python
205 lines
7.0 KiB
Python
from datetime import datetime
|
|
|
|
import pytest
|
|
|
|
from app.db import Database
|
|
from app.selftest import CHECK_VERSION, STATUS_FAILED, STATUS_OK, CheckResult, SelftestRun, should_run
|
|
from app.selftest.runner import SelftestRunner
|
|
from app.selftest.service import SelftestNotConfigured
|
|
|
|
|
|
def at(day, hour, minute=0):
|
|
return datetime(2026, 11, day, hour, minute)
|
|
|
|
|
|
def stamp(moment):
|
|
return moment.timestamp()
|
|
|
|
|
|
@pytest.mark.parametrize("now, last, startup, expected", [
|
|
(at(6, 3, 59), None, False, False),
|
|
(at(6, 4, 0), None, False, True),
|
|
(at(6, 3, 59), None, True, True),
|
|
(at(6, 10), stamp(at(6, 4, 5)), True, False),
|
|
(at(6, 10), stamp(at(6, 4, 5)), False, False),
|
|
(at(6, 1), stamp(at(5, 4, 5)), True, True),
|
|
(at(6, 1), stamp(at(5, 4, 5)), False, False),
|
|
])
|
|
def test_should_run(now, last, startup, expected):
|
|
assert should_run(now, last, startup) is expected
|
|
|
|
|
|
class Moment:
|
|
def __init__(self, value):
|
|
self.value = value
|
|
|
|
def __call__(self):
|
|
return self.value
|
|
|
|
|
|
class FakeNotifier:
|
|
def __init__(self):
|
|
self.calls = []
|
|
|
|
async def selftest(self, ok, text):
|
|
self.calls.append((ok, text))
|
|
|
|
|
|
class FakeService:
|
|
def __init__(self, runs=None, error=None, versions=None):
|
|
self.runs = list(runs or [])
|
|
self.error = error
|
|
self.record = None
|
|
self.call_count = 0
|
|
self.versions = list(versions or [])
|
|
self.peek_calls = 0
|
|
|
|
def last_run(self):
|
|
return self.record
|
|
|
|
async def run(self):
|
|
self.call_count += 1
|
|
if self.error:
|
|
raise self.error
|
|
run = self.runs.pop(0)
|
|
self.record = run
|
|
return run
|
|
|
|
async def peek_version(self):
|
|
self.peek_calls += 1
|
|
return self.versions.pop(0) if self.versions else None
|
|
|
|
|
|
def run_with(ts, status, version_changed=False):
|
|
return SelftestRun(ts, "2.5.0", [CheckResult(CHECK_VERSION, status, "x")], version_changed)
|
|
|
|
|
|
def make_runner(service, notify_enabled=lambda: True, enabled=lambda: True, clock=None):
|
|
db = Database(":memory:")
|
|
return db, SelftestRunner(db, service, FakeNotifier(), enabled, notify_enabled,
|
|
clock=clock or Moment(at(6, 10)))
|
|
|
|
|
|
async def test_tick_runs_once_per_day_and_respects_enabled():
|
|
service = FakeService([run_with(int(at(6, 10).timestamp()), STATUS_OK)])
|
|
clock = Moment(at(6, 10))
|
|
db, runner = make_runner(service, clock=clock)
|
|
assert await runner.tick(startup=True) is True
|
|
clock.value = at(6, 11)
|
|
assert await runner.tick(startup=False) is False
|
|
assert service.call_count == 1
|
|
|
|
|
|
async def test_tick_disabled_never_runs():
|
|
service = FakeService([run_with(1, STATUS_OK)])
|
|
db, runner = make_runner(service, enabled=lambda: False)
|
|
assert await runner.tick(startup=True) is False
|
|
assert service.call_count == 0
|
|
|
|
|
|
async def test_tick_not_configured_is_tolerated():
|
|
service = FakeService(error=SelftestNotConfigured())
|
|
db, runner = make_runner(service)
|
|
assert await runner.tick(startup=True) is True
|
|
|
|
|
|
async def test_manual_rate_limit():
|
|
service = FakeService([run_with(1, STATUS_OK), run_with(2, STATUS_OK)])
|
|
db, runner = make_runner(service)
|
|
assert runner.try_manual() is True
|
|
assert runner.try_manual() is False
|
|
|
|
|
|
async def test_run_manual_returns_none_when_not_configured():
|
|
service = FakeService(error=SelftestNotConfigured())
|
|
db, runner = make_runner(service)
|
|
assert await runner.run_manual("admin1") is None
|
|
|
|
|
|
async def test_run_manual_writes_audit_entry():
|
|
service = FakeService([run_with(1, STATUS_OK)])
|
|
db, runner = make_runner(service)
|
|
await runner.run_manual("admin1")
|
|
rows = db.query("SELECT user, action, target FROM audit WHERE action='selftest.run'")
|
|
assert len(rows) == 1 and rows[0]["user"] == "admin1" and rows[0]["target"] == "selftest"
|
|
|
|
|
|
async def test_notify_fires_on_status_change_but_not_on_repeat():
|
|
service = FakeService([run_with(1, STATUS_OK), run_with(2, STATUS_FAILED), run_with(3, STATUS_FAILED)])
|
|
db = Database(":memory:")
|
|
notifier = FakeNotifier()
|
|
runner = SelftestRunner(db, service, notifier, lambda: True, lambda: True, clock=Moment(at(6, 10)))
|
|
|
|
await runner.run_manual("admin1")
|
|
assert notifier.calls == [] # erster Lauf, kein Vorlauf zum Vergleich
|
|
|
|
await runner.run_manual("admin1")
|
|
assert len(notifier.calls) == 1 and notifier.calls[0][0] is False
|
|
|
|
await runner.run_manual("admin1")
|
|
assert len(notifier.calls) == 1 # gleicher Status wie zuvor, keine Wiederholung
|
|
|
|
|
|
async def test_notify_disabled_sends_nothing():
|
|
service = FakeService([run_with(1, STATUS_OK), run_with(2, STATUS_FAILED)])
|
|
db, runner = make_runner(service, notify_enabled=lambda: False)
|
|
await runner.run_manual("admin1")
|
|
await runner.run_manual("admin1")
|
|
assert runner._notifier.calls == []
|
|
|
|
|
|
def make_version_runner(service, time_fn, enabled=lambda: True):
|
|
db = Database(":memory:")
|
|
return SelftestRunner(db, service, FakeNotifier(), enabled, lambda: True,
|
|
clock=Moment(at(6, 10)), time_fn=time_fn,
|
|
version_check_interval=300.0, version_change_cooldown=300.0)
|
|
|
|
|
|
async def test_version_change_triggers_immediate_run():
|
|
service = FakeService([run_with(1, STATUS_OK)], versions=["2.4.0", "2.5.0"])
|
|
runner = make_version_runner(service, Moment(1000.0))
|
|
# Erster Check lädt die bekannte Version (2.4.0), gilt nicht als Wechsel.
|
|
assert await runner.check_version_change() is False
|
|
assert service.call_count == 0
|
|
|
|
runner._time.value = 1301.0
|
|
assert await runner.check_version_change() is True
|
|
assert service.call_count == 1
|
|
|
|
|
|
async def test_version_change_first_run_ever_is_not_a_change():
|
|
service = FakeService([run_with(1, STATUS_OK)], versions=["2.5.0"])
|
|
runner = make_version_runner(service, Moment(1000.0))
|
|
assert await runner.check_version_change() is False
|
|
assert service.call_count == 0
|
|
|
|
|
|
async def test_version_change_is_debounced_between_checks():
|
|
service = FakeService([], versions=["2.4.0", "2.5.0"])
|
|
runner = make_version_runner(service, Moment(1000.0))
|
|
assert await runner.check_version_change() is False
|
|
runner._time.value = 1100.0
|
|
assert await runner.check_version_change() is False
|
|
assert service.peek_calls == 1
|
|
|
|
|
|
async def test_version_change_cooldown_prevents_second_run():
|
|
service = FakeService([run_with(1, STATUS_OK), run_with(2, STATUS_OK)],
|
|
versions=["2.4.0", "2.5.0", "2.6.0"])
|
|
runner = make_version_runner(service, Moment(1000.0))
|
|
await runner.check_version_change()
|
|
runner._time.value = 1301.0
|
|
assert await runner.check_version_change() is True
|
|
assert service.call_count == 1
|
|
|
|
runner._time.value = 1602.0
|
|
assert await runner.check_version_change() is True
|
|
assert service.call_count == 2
|
|
|
|
|
|
async def test_version_change_disabled_runner_never_checks():
|
|
service = FakeService([], versions=["2.4.0"])
|
|
runner = make_version_runner(service, Moment(1000.0), enabled=lambda: False)
|
|
assert await runner.check_version_change() is False
|
|
assert service.peek_calls == 0
|