From 22f92988dd72b55437f7f42f6c75054f405a63ca Mon Sep 17 00:00:00 2001 From: Michael Sauer Date: Sun, 13 Sep 2026 22:20:11 +0200 Subject: [PATCH] test(station): run_daemon selbst treiben statt eine Kopie der Schleife Die Verdrahtung zwischen RefreshPlanner und Panel - Aktion auf Panel-Methode, was_full=upgraded, record_sleep(), Reihenfolge decide/ausfuehren/zurueckmelden - ueberlebte jede Mutation, weil der Test eine nachgebaute Schleife trieb. Am schlimmsten: der "full"-Zweig auf draw_partial() umgebogen haette bedeutet, dass es auf einem wachen Panel nie wieder einen echten Vollaufbau gibt. run_daemon() nimmt jetzt Uhr, Panel und Zustandsquelle entgegen (Vorgaben unveraendert) und wird im Test mit Fake-Uhr und protokollierendem Panel-Doppel direkt getrieben; die Nachbildung _run_schedule() entfaellt ersatzlos. Dazu: RefreshPlanner liest die Intervallkonstanten wieder pro Entscheidung statt sie als Default-Argument zur Importzeit zu binden, record_sleep() verliert den toten now-Parameter. --- tools/station/display/display.py | 55 +++++-- tools/station/tests/test_render.py | 252 ++++++++++++++++++++--------- 2 files changed, 214 insertions(+), 93 deletions(-) diff --git a/tools/station/display/display.py b/tools/station/display/display.py index 83eb9e0..23ee305 100644 --- a/tools/station/display/display.py +++ b/tools/station/display/display.py @@ -328,10 +328,15 @@ class RefreshPlanner: _NO_MODE = object() # garantiert ungleich jedem echten Modus beim ersten Lauf def __init__(self, now: float, - full_refresh_interval_s: float = FULL_REFRESH_INTERVAL_S, - sleep_after_idle_s: float = SLEEP_AFTER_IDLE_S) -> None: - self.full_refresh_interval_s = full_refresh_interval_s - self.sleep_after_idle_s = sleep_after_idle_s + full_refresh_interval_s: float | None = None, + sleep_after_idle_s: float | None = None) -> None: + # Bewusst None statt der Modulkonstanten als Vorgabewert (Fixrunde 5): ein + # Default-Argument wird zur Importzeit ausgewertet, ein spaeteres + # monkeypatch.setattr(display, "FULL_REFRESH_INTERVAL_S", ...) haette dann keine + # Wirkung mehr auf neu gebaute Planer - die alte Schleife las die Konstanten pro + # Runde. Die Properties unten lesen sie weiterhin bei jeder Entscheidung. + self._full_refresh_interval_s = full_refresh_interval_s + self._sleep_after_idle_s = sleep_after_idle_s self.last_lines: list[str] | None = None self.last_mode: object = self._NO_MODE self.last_full = now @@ -363,6 +368,16 @@ class RefreshPlanner: self.partials_since_full = 0 self.asleep = False + @property + def full_refresh_interval_s(self) -> float: + return (FULL_REFRESH_INTERVAL_S if self._full_refresh_interval_s is None + else self._full_refresh_interval_s) + + @property + def sleep_after_idle_s(self) -> float: + return (SLEEP_AFTER_IDLE_S if self._sleep_after_idle_s is None + else self._sleep_after_idle_s) + def decide(self, now: float, lines: list[str], mode: object) -> str: """Was jetzt zu tun ist. Aendert nichts - das Ergebnis wird per record_*() zurueckgemeldet.""" @@ -395,7 +410,10 @@ class RefreshPlanner: else: self.partials_since_full += 1 - def record_sleep(self, now: float) -> None: + def record_sleep(self) -> None: + """Kein `now`-Parameter (Fixrunde 5): das Einschlafen verschiebt weder die + 30-Minuten-Uhr noch last_draw - es merkt sich nur, dass ein weiteres + panel.sleep() nichts mehr zu tun haette.""" self.asleep = True @@ -434,20 +452,31 @@ def run_clear(revision: str) -> None: log.info("Panel geleert (EPD_REVISION=%s)", revision) -def run_daemon(revision: str) -> None: +def run_daemon(revision: str, clock=time, panel: "Panel | None" = None, + read_state=build_state) -> None: """Endlosschleife: alle POLL_INTERVAL_S Sekunden den Zustand einlesen und die von RefreshPlanner entschiedene Aktion ausfuehren. Hier steht bewusst KEINE Zeitlogik mehr (Fixrunde 4) - die steckt vollstaendig in RefreshPlanner und ist dort ohne Schleife, - ohne time.sleep() und ohne Hardware pruefbar.""" - panel = Panel(revision) + ohne time.sleep() und ohne Hardware pruefbar. + + `clock` (braucht monotonic() und sleep()), `panel` und `read_state` sind einschleusbar, + damit die VERDRAHTUNG hier - welche Aktion welche Panel-Methode ruft, dass die + Selbst-Aufwertung von draw_partial() zurueckgemeldet wird, dass das Einschlafen + zurueckgemeldet wird, und die Reihenfolge decide -> ausfuehren -> zurueckmelden - + selbst pruefbar ist und nicht nur eine Kopie davon (Fixrunde 5, N2: alle drei + Mutationen an genau diesen Zeilen blieben vorher gruen, weil der Test eine nachgebaute + Schleife statt dieser hier trieb). Im Betrieb bleiben die Vorgaben: echte Uhr, echtes + Panel, echtes build_state().""" + if panel is None: + panel = Panel(revision) panel.clear() - planner = RefreshPlanner(time.monotonic()) + planner = RefreshPlanner(clock.monotonic()) while True: - state = build_state() + state = read_state() current_lines = render.lines(state) mode = state.get("mode") - now = time.monotonic() + now = clock.monotonic() action = planner.decide(now, current_lines, mode) if action == "full": @@ -460,9 +489,9 @@ def run_daemon(revision: str) -> None: planner.record_draw(now, current_lines, mode, was_full=upgraded) elif action == "sleep": panel.sleep() - planner.record_sleep(now) + planner.record_sleep() - time.sleep(POLL_INTERVAL_S) + clock.sleep(POLL_INTERVAL_S) def main() -> None: diff --git a/tools/station/tests/test_render.py b/tools/station/tests/test_render.py index e38eaf5..68db65d 100644 --- a/tools/station/tests/test_render.py +++ b/tools/station/tests/test_render.py @@ -488,87 +488,150 @@ def test_compose_survives_a_newline_inside_a_value(): assert image.size == (render.PANEL_WIDTH_PX, render.PANEL_HEIGHT_PX) -# --- RefreshPlanner: die Zeitsteuerung ohne Schleife durchspielen (Fixrunde 4) --------- +# --- run_daemon() gegen eine Fake-Uhr durchspielen (Fixrunde 4, verschaerft in 5) ----- # # run_daemon() hatte null Testabdeckung - die gesamte Zeitsteuerung steckte in einer -# Endlosschleife. Sie liegt jetzt in display.RefreshPlanner und laesst sich hier als -# Zeitreihe durchspielen: pro Poll-Runde decide() fragen, die Aktion "ausfuehren" (das -# Panel-Verhalten wird dabei nur so weit nachgebildet, wie es die Entscheidung beeinflusst: -# ein Teilaufbau auf einem schlafenden Panel wertet sich selbst zu einem Vollaufbau auf, -# siehe Panel.draw_partial()) und das Ergebnis zurueckmelden - genau die Reihenfolge, die -# run_daemon() ebenfalls verwendet. +# Endlosschleife. Sie liegt jetzt in display.RefreshPlanner, aber die ENTSCHEIDUNG allein +# zu pruefen reicht nicht: der Reviewer hat in Fixrunde 5 gezeigt, dass alle drei +# Mutationen an der VERDRAHTUNG zwischen Planer und Panel gruen blieben, solange der Test +# eine nachgebaute Schleife statt run_daemon() selbst trieb (was_full=upgraded verworfen, +# record_sleep() gestrichen, "full" auf draw_partial() umgebogen - letzteres haette +# bedeutet: nie wieder ein echter Vollaufbau auf wachem Panel). +# +# Deshalb wird hier das ECHTE run_daemon() getrieben - mit eingeschleuster Fake-Uhr und +# einem protokollierenden Panel-Doppel. Die frueher hier stehende Nachbildung +# (_run_schedule) ist ersatzlos entfallen, damit es nur eine Fassung der Schleife gibt. SIX_HOURS_S = 6 * 60 * 60 -def _run_schedule(content_at, duration_s, mode="recovery", - poll=display.POLL_INTERVAL_S, planner=None): - """Spielt `duration_s` Sekunden Anzeigebetrieb durch. `content_at(t)` liefert die - Zeilen, die build_state()/render.lines() zum Zeitpunkt t liefern wuerden. - - Gibt (Protokoll, Wachsekunden) zurueck; Protokoll ist eine Liste von (t, Aktion).""" - planner = planner if planner is not None else display.RefreshPlanner(0.0) - log: list[tuple[float, str]] = [] - awake = True # run_daemon() ruft direkt nach dem Start panel.clear() - awake_s = 0.0 - t = 0.0 - while t <= duration_s: - lines = content_at(t) - action = planner.decide(t, lines, mode) - if action == "full": - planner.record_draw(t, lines, mode, was_full=True) - awake = True - elif action == "partial": - # Panel.draw_partial() wertet auf einem schlafenden Panel selbst zu einem - # Vollaufbau auf und meldet das zurueck (F4, siehe Test oben) - im Protokoll - # als "wake-full" unterschieden, weil so ein Aufbau das Panel eben NICHT nur - # teilweise auffrischt. - upgraded = not awake - planner.record_draw(t, lines, mode, was_full=upgraded) - awake = True - if upgraded: - action = "wake-full" - elif action == "sleep": - planner.record_sleep(t) - awake = False - log.append((t, action)) - if awake: - awake_s += poll - t += poll - return log, awake_s +class _StopDaemon(Exception): + """Bricht run_daemon()s Endlosschleife ab, sobald die simulierte Zeit abgelaufen ist.""" -def _times(log, action): - return [t for t, a in log if a == action] +class _FakeClock: + """Ersetzt das time-Modul in run_daemon(): monotonic() liefert die simulierte Zeit, + sleep() laesst sie springen, statt tatsaechlich zu warten.""" + + def __init__(self, until_s: float) -> None: + self.t = 0.0 + self.until_s = until_s + + def monotonic(self) -> float: + return self.t + + def sleep(self, seconds: float) -> None: + self.t += seconds + if self.t > self.until_s: + raise _StopDaemon -def _changing_every(period_s): - """Inhalt, der sich alle `period_s` Sekunden aendert (und sonst gleich bleibt).""" - return lambda t: ["RETTUNG", f"Ausgeliefert: {int(t // period_s)}"] +class _RecordingPanel: + """Panel-Doppel: protokolliert jeden Aufruf mit Zeitstempel und bildet genau das + Verhalten des echten Panels nach, das die Zeitsteuerung beeinflusst - die + Selbst-Aufwertung von draw_partial() auf einem schlafenden Panel (dort als "wake-full" + protokolliert, weil ein solcher Aufbau eben NICHT nur teilweise auffrischt, siehe + test_draw_partial_reports_whether_it_upgraded_itself fuer den Nachweis am echten + Panel) und die Idempotenz von sleep() (ein Aufruf auf einem schon schlafenden Panel + wird als "sleep-noop" protokolliert: unschaedlich, aber ein Zeichen, dass run_daemon() + das Einschlafen nicht zurueckgemeldet hat).""" + + def __init__(self, clock: _FakeClock) -> None: + self.clock = clock + self.calls: list[tuple[float, str]] = [] + self.awake_s = 0.0 + self._awake = False + self._awake_since = 0.0 + + def _record(self, name: str) -> None: + self.calls.append((self.clock.monotonic(), name)) + + def _wake(self) -> None: + if not self._awake: + self._awake = True + self._awake_since = self.clock.monotonic() + + def clear(self) -> None: + self._record("clear") + self._wake() + + def draw_full(self, image) -> None: + assert image is not None + self._record("full") + self._wake() + + def draw_partial(self, image) -> bool: + assert image is not None + if not self._awake: + self._record("wake-full") + self._wake() + return True + self._record("partial") + return False + + def sleep(self) -> None: + if not self._awake: + self._record("sleep-noop") + return + self._record("sleep") + self.awake_s += self.clock.monotonic() - self._awake_since + self._awake = False + + def _finish(self, end_t: float) -> None: + if self._awake: + self.awake_s += end_t - self._awake_since -def test_static_content_never_triggers_another_full_refresh(monkeypatch): +def _drive_daemon(state_at, duration_s: float) -> _RecordingPanel: + """Laesst das echte display.run_daemon() `duration_s` simulierte Sekunden laufen. + `state_at(t)` liefert den Zustand, den build_state() zum Zeitpunkt t liefern wuerde.""" + clock = _FakeClock(duration_s) + panel = _RecordingPanel(clock) + try: + display.run_daemon("V3", clock=clock, panel=panel, + read_state=lambda: state_at(clock.monotonic())) + except _StopDaemon: + pass + panel._finish(clock.monotonic()) + return panel + + +def _times(panel: _RecordingPanel, call: str) -> list[float]: + return [t for t, name in panel.calls if name == call] + + +def _changing_every(period_s, mode="recovery"): + """Zustand, der sich alle `period_s` Sekunden aendert (und sonst gleich bleibt).""" + return lambda t: {"mode": mode, "served": int(t // period_s), "recent": []} + + +def test_static_content_never_triggers_another_full_refresh(): """Fixrunde 3, F3 - vom Reviewer per Mutation als ungetestet belegt: nimmt man die - Bedingung `partials_since_full > 0` aus `stale` heraus, bleiben alle 21 Tests gruen. + Bedingung `partials_since_full > 0` aus `stale` heraus, bleiben alle Tests gruen. Genau diese Bedingung verhindert aber den 33-%-Bias-Zyklus: bei UNVERAENDERTEM Inhalt gibt es keine Geisterbilder aus Teilaufbauten auszugleichen, ein Vollaufbau alle 30 Minuten waere reine Alterung - und weckt das Panel jedesmal fuer weitere 10 Minuten auf.""" - log, awake_s = _run_schedule(lambda t: ["RETTUNG", "Ausgeliefert: 0"], SIX_HOURS_S) + panel = _drive_daemon(lambda t: {"mode": "recovery", "served": 0, "recent": []}, + SIX_HOURS_S) - assert _times(log, "full") == [0.0], ( - f"nur der Erstaufbau darf voll sein, tatsaechlich: {_times(log, 'full')}") - assert _times(log, "partial") == [] - assert _times(log, "sleep") == [600.0], "genau einmal einschlafen, dann Ruhe" - assert awake_s <= 700, f"Panel war {awake_s} s wach - erwartet nur die ersten 10 min" + assert _times(panel, "full") == [0.0], ( + f"nur der Erstaufbau darf voll sein, tatsaechlich: {_times(panel, 'full')}") + assert _times(panel, "partial") == [] + assert _times(panel, "sleep") == [600.0], "genau einmal einschlafen, dann Ruhe" + assert _times(panel, "sleep-noop") == [], ( + "run_daemon() hat das Einschlafen nicht zurueckgemeldet - panel.sleep() wird " + "seitdem in jeder Runde erneut gerufen") + assert panel.awake_s <= 700, ( + f"Panel war {panel.awake_s} s wach - erwartet nur die ersten 10 min") def test_change_every_3_minutes_stays_awake_and_compensates_every_30_minutes(): - log, _ = _run_schedule(_changing_every(3 * 60), SIX_HOURS_S) + panel = _drive_daemon(_changing_every(3 * 60), SIX_HOURS_S) - fulls = _times(log, "full") - assert _times(log, "sleep") == [], "alle 3 min Verkehr - das Panel darf nicht schlafen" - assert len(_times(log, "partial")) > 100, "die Aenderungen muessen Teilaufbauten sein" + fulls = _times(panel, "full") + assert _times(panel, "sleep") == [], "alle 3 min Verkehr - das Panel darf nicht schlafen" + assert len(_times(panel, "partial")) > 100, "die Aenderungen muessen Teilaufbauten sein" assert len(fulls) >= 10, f"Ghosting-Ausgleich alle 30 min fehlt: {fulls}" gaps = [b - a for a, b in zip(fulls, fulls[1:])] assert all(gap >= display.FULL_REFRESH_INTERVAL_S for gap in gaps), ( @@ -578,11 +641,11 @@ def test_change_every_3_minutes_stays_awake_and_compensates_every_30_minutes(): def test_change_every_9_minutes_still_never_sleeps(): """9 Minuten liegen knapp unter SLEEP_AFTER_IDLE_S (10 min) - das Panel bleibt wach, die Teilaufbauten sammeln sich, und die 30-Minuten-Uhr gleicht sie aus.""" - log, _ = _run_schedule(_changing_every(9 * 60), SIX_HOURS_S) + panel = _drive_daemon(_changing_every(9 * 60), SIX_HOURS_S) - fulls = _times(log, "full") - assert _times(log, "sleep") == [] - assert _times(log, "partial"), "Aenderungen bei wachem Panel sind Teilaufbauten" + fulls = _times(panel, "full") + assert _times(panel, "sleep") == [] + assert _times(panel, "partial"), "Aenderungen bei wachem Panel sind Teilaufbauten" assert len(fulls) >= 5, f"Ghosting-Ausgleich fehlt: {fulls}" gaps = [b - a for a, b in zip(fulls, fulls[1:])] assert all(gap >= display.FULL_REFRESH_INTERVAL_S for gap in gaps) @@ -592,18 +655,24 @@ def test_change_every_11_minutes_sleeps_between_and_wakes_with_a_full_refresh(): """11 Minuten liegen knapp ueber SLEEP_AFTER_IDLE_S: das Panel schlaeft zwischen zwei Aenderungen ein, und der naechste Aufbau wird von Panel.draw_partial() selbst zu einem Vollaufbau aufgewertet - es darf also gar kein Teilaufbau mehr entstehen, und die - 30-Minuten-Uhr darf nie zusaetzlich feuern (jeder Aufbau IST schon ein Vollaufbau).""" - log, _ = _run_schedule(_changing_every(11 * 60), SIX_HOURS_S) + 30-Minuten-Uhr darf nie zusaetzlich feuern (jeder Aufbau IST schon ein Vollaufbau). - assert _times(log, "partial") == [], ( + Genau hier faellt auch auf, wenn run_daemon() die Rueckmeldung der Aufwertung + (was_full=upgraded) verwirft: dann zaehlen die Aufwertungen als Teilaufbauten und die + 30-Minuten-Uhr schiebt zusaetzliche Vollaufbauten ein (Fixrunde 5, N2).""" + panel = _drive_daemon(_changing_every(11 * 60), SIX_HOURS_S) + + assert _times(panel, "partial") == [], ( "nach dem Schlafen wertet das Panel jeden Aufbau selbst zu einem Vollaufbau auf") - assert len(_times(log, "wake-full")) == SIX_HOURS_S // (11 * 60), ( - f"pro Aenderung ein Aufbau erwartet: {_times(log, 'wake-full')}") - assert len(_times(log, "sleep")) >= 15, "zwischen den Aenderungen muss geschlafen werden" + assert len(_times(panel, "wake-full")) == SIX_HOURS_S // (11 * 60), ( + f"pro Aenderung ein Aufbau erwartet: {_times(panel, 'wake-full')}") + assert len(_times(panel, "sleep")) >= 15, "zwischen den Aenderungen muss geschlafen werden" + assert _times(panel, "sleep-noop") == [] # Nur der Erstaufbau ist ein regulaerer Vollaufbau: jeder weitere Aufbau IST wegen der # Aufwertung schon vollstaendig, ein zusaetzlicher "stale"-Vollaufbau waere ein Zeichen, # dass die Aufwertung nicht als Vollaufbau verbucht wurde (F4). - assert _times(log, "full") == [0.0], f"unerwartete Vollaufbauten: {_times(log, 'full')}" + assert _times(panel, "full") == [0.0], ( + f"unerwartete Vollaufbauten: {_times(panel, 'full')}") def test_burst_then_quiet_still_gets_its_ghosting_compensation(): @@ -615,27 +684,50 @@ def test_burst_then_quiet_still_gets_its_ghosting_compensation(): konnte danach nie mehr feuern, und genau dieses Bild stand die restlichen rund 5 h 50 min unveraendert da. Der faellige Ausgleich muss stattdessen nachgeholt werden - und das Panel danach wieder zur Ruhe kommen.""" - def content_at(t): + def state_at(t): # 30 Aenderungen im 5-Sekunden-Takt (t = 5 .. 150), danach unveraendert. step = min(int(t // display.POLL_INTERVAL_S), 30) - return ["MIGRATION", f"Ausgeliefert: {step}"] + return {"mode": "migration", "served": step, "recent": []} - log, awake_s = _run_schedule(content_at, SIX_HOURS_S, mode="migration") + panel = _drive_daemon(state_at, SIX_HOURS_S) - assert len(_times(log, "partial")) == 30, ( - f"das Buendel muss aus Teilaufbauten bestehen: {_times(log, 'partial')}") + assert len(_times(panel, "partial")) == 30, ( + f"das Buendel muss aus Teilaufbauten bestehen: {_times(panel, 'partial')}") - fulls = _times(log, "full") + fulls = _times(panel, "full") assert len(fulls) == 2, f"Erstaufbau + genau ein Ausgleich erwartet, war: {fulls}" ausgleich = fulls[1] assert display.FULL_REFRESH_INTERVAL_S <= ausgleich <= 35 * 60, ( f"Ghosting-Ausgleich faellt aus oder kommt zu spaet (t={ausgleich})") # ... und danach wirklich Ruhe: einmal nach dem Buendel, einmal nach dem Ausgleich. - assert len(_times(log, "sleep")) == 2, f"unruhiger Schlaf: {_times(log, 'sleep')}" - assert awake_s <= 30 * 60, ( - f"Panel war {awake_s} s von {SIX_HOURS_S} s wach - der Ausgleich darf nicht in " - "einen Wachzyklus ausarten") + assert len(_times(panel, "sleep")) == 2, f"unruhiger Schlaf: {_times(panel, 'sleep')}" + assert panel.awake_s <= 30 * 60, ( + f"Panel war {panel.awake_s} s von {SIX_HOURS_S} s wach - der Ausgleich darf nicht " + "in einen Wachzyklus ausarten") + + +def test_daemon_draws_fully_on_mode_change_and_partially_on_content_change(): + """Die Zuordnung Aktion -> Panel-Methode direkt an run_daemon() (Fixrunde 5, N2): + ruft der "full"-Zweig panel.draw_partial() statt panel.draw_full(), gaebe es auf einem + wachen Panel NIE wieder einen echten Vollaufbau - Moduswechsel wuerden nur teilweise + gezeichnet und der Ghosting-Ausgleich waere wirkungslos. Kurzer Verlauf, damit die + Zuordnung Zeitpunkt fuer Zeitpunkt nachvollziehbar ist.""" + def state_at(t): + if t < 60: + return {"mode": "recovery", "served": 0, "recent": []} + if t < 120: + return {"mode": "recovery", "served": 1, "recent": []} # nur Inhalt + return {"mode": "migration", "served": 1, "recent": []} # Moduswechsel + + panel = _drive_daemon(state_at, 180) + + assert panel.calls == [ + (0.0, "clear"), + (0.0, "full"), # Erstaufbau (Modus zum ersten Mal bekannt) + (60.0, "partial"), # nur der Inhalt hat sich geaendert + (120.0, "full"), # Moduswechsel - muss ein Vollaufbau sein + ], panel.calls def test_planner_counts_an_upgraded_partial_as_a_full_refresh():