checkbot/app/selftest/runner.py

import asyncio
import logging
import time
from datetime import datetime
from typing import Awaitable, Callable

from .. import audit
from ..alerts import SlidingWindowLimiter, SystemNotifier
from ..db import Database
from .constants import (AUDIT_RUN, AUDIT_TARGET, CHECK_INTERVAL, MANUAL_COOLDOWN_SECONDS, RUN_HOUR, START_DELAY,
                        SYSTEM_USER, VERSION_CHANGE_COOLDOWN, VERSION_CHECK_INTERVAL)
from .model import SelftestRun
from .notify import message_for
from .service import SelftestNotConfigured, SelftestService

log = logging.getLogger(__name__)


def should_run(now: datetime, last_run_ts: float | None, startup: bool) -> bool:
    if last_run_ts is not None and datetime.fromtimestamp(last_run_ts).date() == now.date():
        return False
    return startup or now.hour >= RUN_HOUR


class SelftestRunner:
    def __init__(self, db: Database, service: SelftestService, notifier: SystemNotifier,
                enabled: Callable[[], bool], notify_enabled: Callable[[], bool],
                clock: Callable[[], datetime] = datetime.now,
                sleep: Callable[[float], Awaitable[None]] = asyncio.sleep, start_delay: float = START_DELAY,
                interval: float = CHECK_INTERVAL, cooldown: float = MANUAL_COOLDOWN_SECONDS,
                time_fn: Callable[[], float] = time.time,
                version_check_interval: float = VERSION_CHECK_INTERVAL,
                version_change_cooldown: float = VERSION_CHANGE_COOLDOWN):
        self._db = db
        self._service = service
        self._notifier = notifier
        self._enabled = enabled
        self._notify_enabled = notify_enabled
        self._clock = clock
        self._sleep = sleep
        self._start_delay = start_delay
        self._interval = interval
        self._time = time_fn
        self._version_check_interval = version_check_interval
        self._version_change_cooldown = version_change_cooldown
        self._manual_limiter = SlidingWindowLimiter(time.time, cooldown)
        self._last_attempt: float | None = None
        # None = noch nicht bekannt (wird beim ersten Check aus selftest_state nachgeladen).
        self._known_version: str | None = None
        self._known_version_loaded = False
        self._last_version_check: float | None = None
        self._last_version_change_run: float | None = None

    def try_manual(self) -> bool:
        return self._manual_limiter.try_acquire(1)

    async def _execute(self, user: str) -> SelftestRun | None:
        previous = await asyncio.to_thread(self._service.last_run)
        run = await self._service.run()
        await asyncio.to_thread(audit.record, self._db, user, AUDIT_RUN, AUDIT_TARGET)
        if self._notify_enabled():
            text = message_for(previous, run)
            if text:
                await self._notifier.selftest(run.ok, text)
        return run

    async def run_manual(self, user: str) -> SelftestRun | None:
        try:
            return await self._execute(user)
        except SelftestNotConfigured:
            return None

    async def tick(self, startup: bool) -> bool:
        if not self._enabled():
            return False
        record = await asyncio.to_thread(self._service.last_run)
        known = [ts for ts in (record.ts if record else None, self._last_attempt) if ts is not None]
        now = self._clock()
        if not should_run(now, max(known) if known else None, startup):
            return False
        self._last_attempt = now.timestamp()
        try:
            await self._execute(SYSTEM_USER)
        except SelftestNotConfigured:
            pass
        except Exception:
            log.exception("Selbsttest fehlgeschlagen")
        return True

    async def check_version_change(self) -> bool:
        if not self._enabled():
            return False
        now = self._time()
        if (self._last_version_check is not None
                and now - self._last_version_check < self._version_check_interval):
            return False
        self._last_version_check = now
        if not self._known_version_loaded:
            record = await asyncio.to_thread(self._service.last_run)
            self._known_version = (record.version if record else "") or None
            self._known_version_loaded = True
        version = await self._service.peek_version()
        if not version or version == self._known_version:
            return False
        first_known = self._known_version is None
        # Neue Version sofort merken, damit ein hängender/fehlschlagender Lauf nicht erneut auslöst.
        self._known_version = version
        if first_known:
            return False
        if (self._last_version_change_run is not None
                and now - self._last_version_change_run < self._version_change_cooldown):
            return False
        self._last_version_change_run = now
        try:
            await self._execute(SYSTEM_USER)
        except SelftestNotConfigured:
            pass
        except Exception:
            log.exception("Selbsttest: Lauf nach Versionswechsel fehlgeschlagen")
        return True

    async def run(self) -> None:
        await self._sleep(self._start_delay)
        startup = True
        while True:
            try:
                await self.tick(startup)
            except Exception:
                log.exception("Selbsttest: Prüfung fehlgeschlagen")
            startup = False
            try:
                await self.check_version_change()
            except Exception:
                log.exception("Selbsttest: Versionsprüfung fehlgeschlagen")
            await self._sleep(min(self._interval, self._version_check_interval))