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.
This commit is contained in:
Michael Sauer
@@ -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():
|
||||
|
||||
Reference in New Issue
Block a user