Michael Sauer
b8c7febb8f
Der Reset von partials_since_full beim Einschlafen liess den faelligen Ausgleichs-Vollaufbau dauerhaft ausfallen: nach einem Buendel Teilaufbauten schlief das Panel mit dem Geisterbild ein und zeigte es stundenlang weiter, weil stale nie mehr feuern konnte. Reset ersatzlos gestrichen - der Ausgleich bleibt ausstehend und wird zum 30-Minuten-Termin nachgeholt; der in Fixrunde 3 beseitigte Bias-Zyklus kommt dadurch nicht zurueck. Die gesamte Zeitsteuerung steckt jetzt in RefreshPlanner statt in run_daemon()s Endlosschleife und ist ohne Schleife, ohne time.sleep() und ohne Hardware pruefbar. Neue Tests spielen sechs Stunden Betrieb durch (statischer Inhalt, Aenderung alle 3/9/11 Minuten, Buendel-dann-Ruhe) und schliessen die drei vom Reviewer per Mutation belegten Luecken aus Fixrunde 3.
661 lines
29 KiB
Python
661 lines
29 KiB
Python
import json
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "display"))
|
|
|
|
import display # noqa: E402
|
|
import epdconfig # noqa: E402
|
|
import render # noqa: E402
|
|
|
|
|
|
def test_off_mode_says_so_and_shows_nothing_else():
|
|
lines = render.lines({"mode": "off"})
|
|
assert any("Station aus" in l for l in lines)
|
|
assert not any("SSID" in l for l in lines)
|
|
|
|
|
|
def test_migration_shows_detected_lamp():
|
|
lines = render.lines({
|
|
"mode": "migration", "phase": "auto", "served": 3,
|
|
"ssid": "qconnex-firmware",
|
|
"last_seen": {"product_name": "Klyqa GU10 RGB CW/WW", "target": "gu10_rgb",
|
|
"variant": "prod", "lamp_version": "V1.1.0.1138"},
|
|
"recent": [{"target": "gu10_rgb", "status": "ok", "ts": "2026-09-13T13:04:02Z"}]})
|
|
text = "\n".join(lines)
|
|
assert "MIGRATION" in text
|
|
assert "Klyqa GU10 RGB CW/WW" in text
|
|
assert "V1.1.0.1138" in text
|
|
assert "prod" in text
|
|
assert "13:04" in text
|
|
|
|
|
|
def test_recovery_mode_without_traffic():
|
|
lines = render.lines({"mode": "recovery", "ssid": "ninlight-recovery",
|
|
"served": 0, "recent": []})
|
|
text = "\n".join(lines)
|
|
assert "RETTUNG" in text
|
|
assert "ninlight-recovery" in text
|
|
|
|
|
|
def test_long_product_name_is_not_truncated_by_lines():
|
|
"""lines() kuerzt bewusst NICHT (mehr) - Breitenkuerzung braucht eine echte
|
|
Textmessung (schriftabhaengig, siehe render.py-Modul-Docstring/Fixrunde 2) und
|
|
passiert deshalb erst in compose(). lines() bleibt so ohne Pillow testbar und liefert
|
|
hier den vollen, ungekuerzten Produktnamen zurueck."""
|
|
lines = render.lines({"mode": "migration", "phase": "listen", "served": 0, "recent": [],
|
|
"ssid": "qconnex-firmware",
|
|
"last_seen": {"product_name": "X" * 90, "target": "e27_rgb",
|
|
"variant": "dev", "lamp_version": "V1"}})
|
|
assert "X" * 90 in lines
|
|
|
|
|
|
def test_truncate_to_width_shortens_and_appends_ellipsis():
|
|
"""_truncate_to_width() selbst importiert kein Pillow - eine einfache Zeichenzahl-
|
|
Messfunktion reicht, um die Kuerzungslogik ohne Hardware/Pillow zu pruefen."""
|
|
result = render._truncate_to_width("X" * 50, measure=len, max_width_px=10)
|
|
assert len(result) <= 10
|
|
assert result.endswith("…")
|
|
|
|
|
|
def test_truncate_to_width_leaves_short_text_untouched():
|
|
assert render._truncate_to_width("kurz", measure=len, max_width_px=10) == "kurz"
|
|
|
|
|
|
def test_compose_does_not_draw_into_the_last_column_for_long_lines():
|
|
"""Regressionstest zu Fixrunde 3: der Reviewer hat per Mutation belegt, dass eine
|
|
deaktivierte Kuerzung in compose() (fitted = line) die vorherige Testsuite NICHT
|
|
aufgehalten haette, weil der alte Test die Kuerzung selbst nachbaute (mit derselben
|
|
_truncate_to_width()-Funktion, die auch geprueft werden sollte) statt das von
|
|
compose() tatsaechlich GEZEICHNETE BILD zu inspizieren.
|
|
|
|
Dieser Test prueft stattdessen die Pixel direkt: bei korrekt wirkender Kuerzung
|
|
bleibt die letzte Panel-Spalte (x = PANEL_WIDTH_PX - 1) fuer eine absichtlich sehr
|
|
lange Zeile weiss, weil die Kuerzung auf PANEL_WIDTH_PX - 2*TEXT_MARGIN_PX zielt
|
|
(empirisch mit echtem Pillow bestaetigt: die rechte Kante liegt bei ca. Spalte 237,
|
|
deutlich vor Spalte 249). Mit deaktivierter Kuerzung fuellt der zeichenweise laengere
|
|
Text die Zeile bis zum Rand - ebenfalls empirisch bestaetigt (die letzte Spalte
|
|
enthaelt dann einen schwarzen Pixel)."""
|
|
state = {"mode": "migration", "phase": "listen", "served": 0, "recent": [],
|
|
"ssid": "qconnex-firmware",
|
|
"last_seen": {"product_name": "W" * 200, "target": "e27_rgb",
|
|
"variant": "dev", "lamp_version": "V1"}}
|
|
|
|
image = render.compose(state)
|
|
pixels = image.load()
|
|
last_col = render.PANEL_WIDTH_PX - 1
|
|
|
|
black_in_last_column = any(pixels[last_col, y] == 0 for y in range(render.PANEL_HEIGHT_PX))
|
|
assert not black_in_last_column, "Text erreicht die letzte Spalte - Kuerzung wirkt nicht"
|
|
|
|
|
|
def test_missing_fields_do_not_raise():
|
|
"""status.json kann jederzeit halb geschrieben oder aus einer aelteren Version sein."""
|
|
for state in ({}, {"mode": "migration"}, {"mode": "migration", "last_seen": {}}):
|
|
assert render.lines(state)
|
|
|
|
|
|
def test_armed_phase_is_visible():
|
|
lines = render.lines({"mode": "migration", "phase": "armed", "armed_target": "g95_cw",
|
|
"served": 0, "recent": [], "ssid": "qconnex-firmware"})
|
|
assert "FREIGEGEBEN" in "\n".join(lines)
|
|
|
|
|
|
# --- display.build_state() (Fixrunde 1, I3) ---------------------------------------
|
|
#
|
|
# Dieselbe Unterscheidung, die im CLI (tools/station/ninlight-station, status_cmd) zweimal
|
|
# nachgebessert werden musste: Phase/armed_target kommen aus control.json und sind nur im
|
|
# Migrationsmodus gueltig; ein control.json-Rest aus einer frueheren Migration darf im
|
|
# Rettungsmodus nicht als "FREIGEGEBEN" durchschlagen.
|
|
|
|
def test_control_json_is_ignored_outside_migration_mode(tmp_path, monkeypatch):
|
|
monkeypatch.setattr(display, "_service_active", lambda service_name: True)
|
|
(tmp_path / "mode").write_text("recovery")
|
|
(tmp_path / "control.json").write_text(json.dumps({"phase": "armed", "armed_target": "g95_cw"}))
|
|
conf_path = tmp_path / "station.conf"
|
|
conf_path.write_text("SSID='ninlight-recovery'\n")
|
|
|
|
state = display.build_state(run_dir=tmp_path, conf_path=conf_path, service_name="x")
|
|
|
|
assert state["mode"] == "recovery"
|
|
assert "phase" not in state
|
|
assert "armed_target" not in state
|
|
assert "FREIGEGEBEN" not in "\n".join(render.lines(state))
|
|
|
|
|
|
def test_build_state_off_when_service_not_active(tmp_path, monkeypatch):
|
|
monkeypatch.setattr(display, "_service_active", lambda service_name: False)
|
|
# Selbst ein noch so frisches/gueltiges status.json darf "off" nicht verhindern -
|
|
# "ninlight-station off" laesst status.json bewusst unangetastet liegen.
|
|
(tmp_path / "mode").write_text("migration")
|
|
(tmp_path / "status.json").write_text(json.dumps({"mode": "migration", "served": 5}))
|
|
|
|
state = display.build_state(run_dir=tmp_path, conf_path=tmp_path / "station.conf",
|
|
service_name="x")
|
|
|
|
assert state == {"mode": "off"}
|
|
|
|
|
|
def test_build_state_mode_comes_from_mode_file_not_status_json(tmp_path, monkeypatch):
|
|
"""Der Modus-Vorrang liegt bei RUN_DIR/mode, nicht bei status.json's "mode"-Feld -
|
|
status.json kann veraltet sein (Fixrunde 1, Kleinigkeit: der Vorrang war zuerst
|
|
umgekehrt)."""
|
|
monkeypatch.setattr(display, "_service_active", lambda service_name: True)
|
|
(tmp_path / "mode").write_text("recovery")
|
|
(tmp_path / "status.json").write_text(json.dumps({"mode": "migration", "served": 1}))
|
|
conf_path = tmp_path / "station.conf"
|
|
conf_path.write_text("SSID='ninlight-recovery'\n")
|
|
|
|
state = display.build_state(run_dir=tmp_path, conf_path=conf_path, service_name="x")
|
|
|
|
assert state["mode"] == "recovery"
|
|
assert state["served"] == 1
|
|
|
|
|
|
def test_build_state_falls_back_to_recovery_without_mode_file(tmp_path, monkeypatch):
|
|
monkeypatch.setattr(display, "_service_active", lambda service_name: True)
|
|
state = display.build_state(run_dir=tmp_path, conf_path=tmp_path / "station.conf",
|
|
service_name="x")
|
|
assert state["mode"] == "recovery"
|
|
|
|
|
|
def test_read_conf_accepts_export_prefix(tmp_path):
|
|
"""setup_pi.sh prueft beim Nachtragen fehlender Schluessel ausdruecklich auf ein
|
|
optionales "export "-Praefix (^(export[[:space:]]+)?KEY=) - eine bereits vorhandene
|
|
Konfigurationsdatei mit "export SSID=..." muss also genauso gelesen werden koennen."""
|
|
conf_path = tmp_path / "station.conf"
|
|
conf_path.write_text("export SSID='ninlight-recovery'\nexport MIGRATE_SSID='qconnex-firmware'\n")
|
|
|
|
values = display._read_conf(conf_path)
|
|
|
|
assert values["SSID"] == "ninlight-recovery"
|
|
assert values["MIGRATE_SSID"] == "qconnex-firmware"
|
|
|
|
|
|
# --- Panel-Stubs fuer die V2/V3-Aufwach-/Schlafsequenz (Fixrunde 2, N1) --------------
|
|
#
|
|
# gpiozero/spidev sind auf der Entwicklungsmaschine nicht installiert (und selbst wenn
|
|
# sie es waeren, haengt hier kein echtes Panel) - diese Fakes implementieren nur genau
|
|
# die Teilmenge der Schnittstelle, die epdconfig.py und die vendored Panel-Treiber
|
|
# tatsaechlich aufrufen, damit Panel() end-to-end (ohne echte Hardware) durchlaufen kann.
|
|
|
|
class _FakeDigitalOutputDevice:
|
|
def __init__(self, pin):
|
|
self.pin = pin
|
|
self.value = False
|
|
|
|
def close(self):
|
|
pass
|
|
|
|
|
|
class _FakeDigitalInputDevice:
|
|
def __init__(self, pin, pull_up=False):
|
|
self.pin = pin
|
|
self.value = 0 # "idle", nie "busy" - sonst haengt ReadBusy() in einer Schleife
|
|
|
|
def close(self):
|
|
pass
|
|
|
|
|
|
class _FakeGpiozeroModule:
|
|
DigitalOutputDevice = _FakeDigitalOutputDevice
|
|
DigitalInputDevice = _FakeDigitalInputDevice
|
|
|
|
class GPIOZeroError(Exception):
|
|
pass
|
|
|
|
|
|
class _FakeSpiDev:
|
|
def __init__(self):
|
|
self.max_speed_hz = None
|
|
self.mode = None
|
|
|
|
def open(self, bus, device):
|
|
pass
|
|
|
|
def writebytes(self, data):
|
|
pass
|
|
|
|
def writebytes2(self, data):
|
|
pass
|
|
|
|
def close(self):
|
|
pass
|
|
|
|
|
|
class _FakeSpidevModule:
|
|
SpiDev = _FakeSpiDev
|
|
|
|
|
|
def _stub_hardware(monkeypatch):
|
|
"""Ersetzt epdconfig's Sicht auf gpiozero/spidev durch die Fakes oben, taeuscht ein
|
|
vorhandenes /dev/spidev0.0 vor, macht delay_ms() ein No-op (epd.sleep() wartet sonst
|
|
echte 2 Sekunden) und stellt sicher, dass epdconfig mit "nichts offen" in den Test
|
|
geht - epdconfig haelt seinen Zustand in Modulglobalen, die sonst zwischen Tests
|
|
durchsickern koennten."""
|
|
monkeypatch.setattr(epdconfig, "gpiozero", _FakeGpiozeroModule)
|
|
monkeypatch.setattr(epdconfig, "spidev", _FakeSpidevModule)
|
|
monkeypatch.setattr(epdconfig.os.path, "exists", lambda path: True)
|
|
monkeypatch.setattr(epdconfig, "delay_ms", lambda ms: None)
|
|
monkeypatch.setattr(epdconfig, "_spi", None)
|
|
monkeypatch.setattr(epdconfig, "_rst_pin", None)
|
|
monkeypatch.setattr(epdconfig, "_dc_pin", None)
|
|
monkeypatch.setattr(epdconfig, "_busy_pin", None)
|
|
|
|
|
|
def test_v3_partial_draw_after_sleep_does_not_crash(monkeypatch):
|
|
"""Regressionstest zu Fixrunde 2, N1 - genau der vom Reviewer nachgestellte Ablauf:
|
|
Panel("V3") -> clear -> draw_full -> sleep -> draw_partial. Vorher stuerzte das mit
|
|
AttributeError ab: epd.sleep() schliesst ueber epdconfig.module_exit() die
|
|
GPIO-Pins, und epd2in13_V3.displayPartial() fasst als Erstes die Reset-Leitung an -
|
|
von main()s except (DisplayUnavailable, ImportError) NICHT gefangen, also ein
|
|
echter Absturz (Exit 1, Neustartschleife). V3 ist die Vorgaberevision, und der
|
|
Vollaufbau-nach-jedem-Bild-Ansatz aus Fixrunde 1 erreichte diesen Pfad bei der
|
|
ersten Inhaltsaenderung ohne Moduswechsel - binnen Sekunden nach dem Start.
|
|
|
|
Mit dem Panel._awake-Tracking wertet draw_partial() sich nach sleep() selbst zu
|
|
einem Vollaufbau auf, statt die geschlossenen Pins anzufassen - dieser Test waere
|
|
mit der Fixrunde-1-Fassung von Panel fehlgeschlagen und besteht jetzt."""
|
|
_stub_hardware(monkeypatch)
|
|
|
|
panel = display.Panel("V3")
|
|
image = render.compose({"mode": "off"})
|
|
|
|
panel.clear()
|
|
panel.draw_full(image)
|
|
panel.sleep()
|
|
panel.draw_partial(image) # das ist der Schritt, der vorher abgestuerzt waere
|
|
|
|
|
|
def test_v2_partial_draw_after_sleep_does_not_crash(monkeypatch):
|
|
"""Dieselbe Rauchprobe fuer V2 wie oben fuer V3 (Fixrunde 2, N3) - dass hier NICHTS
|
|
crasht, ist aber nur die halbe Absicherung: ob dabei auch die RICHTIGEN Register
|
|
gesetzt werden, prueft erst der staerkere Test unten
|
|
(test_v2_wake_after_sleep_sets_full_update_registers) mit einem Kommando-Rekorder.
|
|
Der Reviewer hat zu Recht angemerkt, dass eine Fassung dieses Tests ohne jede
|
|
Zusicherung nichts ueber die Registerlage aussagt - deshalb hier wenigstens die
|
|
beobachtbaren Panel-Zustandsuebergaenge pruefen."""
|
|
_stub_hardware(monkeypatch)
|
|
|
|
panel = display.Panel("V2")
|
|
image = render.compose({"mode": "off"})
|
|
|
|
panel.clear()
|
|
panel.draw_full(image)
|
|
panel.sleep()
|
|
assert panel._awake is False # Vorbedingung fuer den eigentlich interessanten Schritt
|
|
|
|
panel.draw_partial(image) # das ist der Schritt, der bei falscher Sequenz froeren wuerde
|
|
|
|
assert panel._awake is True
|
|
assert epdconfig._spi is not None # Panel hat sich nach dem Schlafen neu initialisiert
|
|
|
|
|
|
def test_panel_sleep_is_idempotent(monkeypatch):
|
|
"""Ein zweiter sleep()-Aufruf, waehrend das Panel schon schlaeft, darf nicht
|
|
versuchen, ueber die von module_exit() bereits geschlossenen Pins SPI-Befehle zu
|
|
schicken (derselbe AttributeError-Mechanismus wie N1, nur selbst herbeigefuehrt).
|
|
Die staerkere Fassung unten (test_panel_sleep_sends_no_commands_when_already_asleep)
|
|
prueft das zusaetzlich direkt an der gesendeten Kommandofolge."""
|
|
_stub_hardware(monkeypatch)
|
|
|
|
panel = display.Panel("V3")
|
|
panel.clear()
|
|
panel.sleep()
|
|
assert panel._awake is False
|
|
|
|
panel.sleep() # darf nicht erneut versuchen, epd.sleep() aufzurufen
|
|
|
|
assert panel._awake is False
|
|
|
|
|
|
# --- Kommando-Rekorder: Registerzustand nach echten Bytes pruefen, nicht nur "kein Absturz"
|
|
#
|
|
# Der Reviewer hat per Mutation gezeigt, dass die Stubs oben (_stub_hardware) N2/N3 nicht
|
|
# wirklich absichern: displayPartBaseImage(buf) durch display(buf) ersetzen, oder die
|
|
# V2-Registerfolge nach sleep() kaputt machen, liess alle bisherigen Tests gruen. Ein
|
|
# stiller Registerzustandsfehler wirft keine Ausnahme - die einzige Art, ihn in Software
|
|
# statt erst am Panel selbst zu fassen, ist die tatsaechlich gesendete Byte-Folge zu
|
|
# protokollieren. Ersetzt dafuer epdconfig's Funktionen direkt (nicht gpiozero/spidev
|
|
# darunter) und wertet aus, welche Bytes waehrend DC=0 (Kommando) vs. DC=1 (Daten)
|
|
# gesendet wurden - siehe send_command()/send_data() in den vendored Treibern.
|
|
|
|
class _RecordingHardware:
|
|
def __init__(self):
|
|
self.commands: list[int] = []
|
|
self._dc = 0
|
|
self._busy = 0
|
|
|
|
def digital_write(self, pin, value):
|
|
if pin == epdconfig.DC_PIN:
|
|
self._dc = value
|
|
# RST_PIN/CS_PIN sind fuer die Kommandofolge nicht relevant.
|
|
|
|
def digital_read(self, pin):
|
|
return self._busy
|
|
|
|
def delay_ms(self, ms):
|
|
pass
|
|
|
|
def spi_writebyte(self, data):
|
|
if self._dc == 0:
|
|
self.commands.extend(data)
|
|
|
|
def spi_writebyte2(self, data):
|
|
pass # Nutzdaten (Bildpuffer/LUT) - fuer die Kommandofolge nicht relevant.
|
|
|
|
def module_init(self):
|
|
return 0
|
|
|
|
def module_exit(self):
|
|
pass
|
|
|
|
|
|
def _install_recorder(monkeypatch) -> _RecordingHardware:
|
|
recorder = _RecordingHardware()
|
|
monkeypatch.setattr(epdconfig, "digital_write", recorder.digital_write)
|
|
monkeypatch.setattr(epdconfig, "digital_read", recorder.digital_read)
|
|
monkeypatch.setattr(epdconfig, "delay_ms", recorder.delay_ms)
|
|
monkeypatch.setattr(epdconfig, "spi_writebyte", recorder.spi_writebyte)
|
|
monkeypatch.setattr(epdconfig, "spi_writebyte2", recorder.spi_writebyte2)
|
|
monkeypatch.setattr(epdconfig, "module_init", recorder.module_init)
|
|
monkeypatch.setattr(epdconfig, "module_exit", recorder.module_exit)
|
|
return recorder
|
|
|
|
|
|
def test_v3_full_refresh_writes_reference_buffer(monkeypatch):
|
|
"""Regressionstest zu Fixrunde 2, N2 - mit einer echten Zusicherung (Fixrunde 3): ein
|
|
Vollaufbau muss bei V3 sowohl RAM 0x24 als auch 0x26 beschreiben
|
|
(displayPartBaseImage() statt display()), sonst arbeitet der naechste Teilaufbau
|
|
gegen einen undefinierten Referenzpuffer. Der Reviewer hat per Mutation
|
|
(displayPartBaseImage(buf) -> display(buf)) gezeigt, dass die vorherige Testsuite das
|
|
nicht bemerkt haette - dieser Test waere mit dieser Mutation rot."""
|
|
recorder = _install_recorder(monkeypatch)
|
|
panel = display.Panel("V3")
|
|
image = render.compose({"mode": "off"})
|
|
|
|
panel.clear()
|
|
recorder.commands.clear() # nur der eigentliche Vollaufbau interessiert hier
|
|
|
|
panel.draw_full(image)
|
|
|
|
assert 0x24 in recorder.commands, f"0x24 (aktuelles RAM) fehlt: {recorder.commands}"
|
|
assert 0x26 in recorder.commands, f"0x26 (Referenzpuffer) fehlt: {recorder.commands}"
|
|
assert recorder.commands.index(0x26) > recorder.commands.index(0x24)
|
|
|
|
|
|
def test_v2_wake_after_sleep_sets_full_update_registers(monkeypatch):
|
|
"""Regressionstest zu Fixrunde 2, N3 - mit einer echten Zusicherung (Fixrunde 3,
|
|
statt der vorherigen Fassung ganz ohne): nach sleep() muss der naechste Aufbau
|
|
(draw_partial(), das sich wegen Panel._awake selbst zu einem Vollaufbau aufwertet)
|
|
die Register senden, die NUR der FULL_UPDATE-Zweig von epd2in13_V2.init() setzt
|
|
(0x01 Driver output control, 0x11 Data entry mode, 0x44/0x45 RAM-Fenster, 0x4E/0x4F
|
|
RAM-Zaehler), und danach sowohl 0x24 als auch 0x26 beschreiben. Dieser Fehler wirft
|
|
keine Ausnahme - er ist nur an der gesendeten Byte-Folge erkennbar, sonst erst am
|
|
Panel selbst, wo er von einem Verdrahtungsfehler nicht zu unterscheiden waere."""
|
|
recorder = _install_recorder(monkeypatch)
|
|
panel = display.Panel("V2")
|
|
image = render.compose({"mode": "off"})
|
|
|
|
panel.clear()
|
|
panel.draw_full(image)
|
|
panel.sleep()
|
|
recorder.commands.clear() # nur die Kommandofolge NACH dem Schlafen interessiert hier
|
|
|
|
panel.draw_partial(image)
|
|
|
|
required = [0x01, 0x11, 0x44, 0x45, 0x4E, 0x4F]
|
|
positions = [recorder.commands.index(cmd) for cmd in required
|
|
if cmd in recorder.commands]
|
|
assert len(positions) == len(required), (
|
|
f"FULL_UPDATE-Register fehlen nach sleep(): {recorder.commands}")
|
|
assert positions == sorted(positions), (
|
|
f"FULL_UPDATE-Register nicht in der erwarteten Reihenfolge: {recorder.commands}")
|
|
|
|
assert 0x24 in recorder.commands and 0x26 in recorder.commands, (
|
|
f"0x24/0x26 (Referenzpuffer) fehlen: {recorder.commands}")
|
|
idx_24 = recorder.commands.index(0x24)
|
|
idx_26 = recorder.commands.index(0x26)
|
|
assert idx_24 > positions[-1], "0x24 vor den FULL_UPDATE-Registern geschrieben"
|
|
assert idx_26 > idx_24, "0x26 (Referenzpuffer) fehlt oder kommt vor 0x24"
|
|
|
|
|
|
def test_panel_clear_establishes_reference_buffer(monkeypatch):
|
|
"""Regressionstest zu Fixrunde 3, F2: clear() setzte vorher self._awake=True, ohne
|
|
einen 0x26-Referenzpuffer zu schreiben (epd.Clear() schreibt nur 0x24, bei V2 ist das
|
|
0x26-Schreiben in Clear() sogar auskommentiert) - ein draw_partial() direkt nach
|
|
clear() liefe gegen denselben undefinierten Puffer, den N2 eigentlich beheben
|
|
sollte. clear() verwendet jetzt wie draw_full() displayPartBaseImage()."""
|
|
recorder = _install_recorder(monkeypatch)
|
|
panel = display.Panel("V3")
|
|
|
|
panel.clear()
|
|
|
|
assert 0x24 in recorder.commands
|
|
assert 0x26 in recorder.commands
|
|
assert panel._awake is True
|
|
|
|
|
|
def test_panel_sleep_sends_no_commands_when_already_asleep(monkeypatch):
|
|
"""Staerkere Fassung von test_panel_sleep_is_idempotent (Fixrunde 3): ein zweiter
|
|
sleep()-Aufruf darf NICHT nochmal versuchen, ueber die schon geschlossenen Pins
|
|
SPI-Befehle zu schicken - hier direkt an der aufgezeichneten Kommandofolge geprueft
|
|
statt nur daran, dass keine Ausnahme fliegt."""
|
|
recorder = _install_recorder(monkeypatch)
|
|
panel = display.Panel("V3")
|
|
panel.clear()
|
|
panel.sleep()
|
|
recorder.commands.clear()
|
|
|
|
panel.sleep() # bereits asleep - darf keine weiteren Kommandos senden
|
|
|
|
assert recorder.commands == []
|
|
|
|
|
|
def test_draw_partial_reports_whether_it_upgraded_itself(monkeypatch):
|
|
"""Fixrunde 3, F4 - vom Reviewer per Mutation als ungetestet belegt: gibt
|
|
draw_partial() nach dem Aufwerten `False` statt `True` zurueck, blieben alle 21 Tests
|
|
gruen. Der Rueckgabewert ist aber die einzige Information, an der RefreshPlanner
|
|
erkennt, dass die 30-Minuten-Uhr neu beginnt - ohne ihn kaeme der naechste
|
|
Ghosting-Ausgleich zu frueh und alterte das Panel ohne Not."""
|
|
_install_recorder(monkeypatch)
|
|
panel = display.Panel("V3")
|
|
image = render.compose({"mode": "off"})
|
|
|
|
panel.clear()
|
|
assert panel.draw_partial(image) is False, "waches Panel: kein Vollaufbau noetig"
|
|
|
|
panel.sleep()
|
|
assert panel.draw_partial(image) is True, (
|
|
"nach sleep() wertet draw_partial() intern zu einem Vollaufbau auf - das muss es "
|
|
"auch zurueckmelden")
|
|
assert panel.draw_partial(image) is False, "danach ist das Panel wieder wach"
|
|
|
|
|
|
def test_compose_survives_a_newline_inside_a_value():
|
|
"""Fixrunde 3, F5 - ebenfalls per Mutation als ungetestet belegt (`line = raw_line`
|
|
liess alle Tests gruen): draw.textlength() wirft ValueError("can't measure length of
|
|
multiline text"), sobald eine Zeile ein "\\n" enthaelt. product_name kommt aus der
|
|
Lampenerkennung, ist also letztlich Netzwerkdatum - und ein ValueError hier faengt
|
|
main()s `except (DisplayUnavailable, ImportError)` NICHT ab, das waere ein echter
|
|
Absturz des Dienstes."""
|
|
state = {"mode": "migration", "phase": "auto", "served": 0, "recent": [],
|
|
"last_seen": {"product_name": "Klyqa\nGU10", "target": "gu10_rgb",
|
|
"variant": "prod", "lamp_version": "V1"}}
|
|
|
|
image = render.compose(state) # vor F5: ValueError
|
|
|
|
assert image.size == (render.PANEL_WIDTH_PX, render.PANEL_HEIGHT_PX)
|
|
|
|
|
|
# --- RefreshPlanner: die Zeitsteuerung ohne Schleife durchspielen (Fixrunde 4) ---------
|
|
#
|
|
# 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.
|
|
|
|
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
|
|
|
|
|
|
def _times(log, action):
|
|
return [t for t, a in log if a == action]
|
|
|
|
|
|
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)}"]
|
|
|
|
|
|
def test_static_content_never_triggers_another_full_refresh(monkeypatch):
|
|
"""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.
|
|
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)
|
|
|
|
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"
|
|
|
|
|
|
def test_change_every_3_minutes_stays_awake_and_compensates_every_30_minutes():
|
|
log, _ = _run_schedule(_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"
|
|
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), (
|
|
f"Vollaufbauten zu dicht: {fulls}")
|
|
|
|
|
|
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)
|
|
|
|
fulls = _times(log, "full")
|
|
assert _times(log, "sleep") == []
|
|
assert _times(log, "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)
|
|
|
|
|
|
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)
|
|
|
|
assert _times(log, "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"
|
|
# 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')}"
|
|
|
|
|
|
def test_burst_then_quiet_still_gets_its_ghosting_compensation():
|
|
"""Fixrunde 4, N1 - der Dauerzustand, den jemand am laengsten ansieht: ein Buendel von
|
|
30 Teilaufbauten in 2,5 Minuten (Migration mehrerer Lampen), danach stundenlang Ruhe.
|
|
|
|
Die Fassung aus Fixrunde 3 setzte partials_since_full beim Einschlafen auf 0; das
|
|
Panel schlief 10 Minuten nach dem Buendel mit dem entstandenen Geisterbild ein, stale
|
|
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):
|
|
# 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}"]
|
|
|
|
log, awake_s = _run_schedule(content_at, SIX_HOURS_S, mode="migration")
|
|
|
|
assert len(_times(log, "partial")) == 30, (
|
|
f"das Buendel muss aus Teilaufbauten bestehen: {_times(log, 'partial')}")
|
|
|
|
fulls = _times(log, "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")
|
|
|
|
|
|
def test_planner_counts_an_upgraded_partial_as_a_full_refresh():
|
|
"""Direkt an der Zustandsmaschine (F4): meldet der Aufrufer eine von
|
|
Panel.draw_partial() selbst aufgewertete Teilaktualisierung als Vollaufbau, beginnt die
|
|
30-Minuten-Uhr neu und es gibt nichts mehr auszugleichen."""
|
|
planner = display.RefreshPlanner(0.0)
|
|
planner.record_draw(0.0, ["a"], "recovery", was_full=True)
|
|
planner.record_draw(10.0, ["b"], "recovery", was_full=False)
|
|
assert planner.partials_since_full == 1
|
|
|
|
planner.record_draw(20.0, ["c"], "recovery", was_full=True) # aufgewerteter Teilaufbau
|
|
|
|
assert planner.partials_since_full == 0
|
|
assert planner.decide(20.0 + display.FULL_REFRESH_INTERVAL_S, ["c"], "recovery") != "full"
|
|
|
|
|
|
def test_planner_draws_fully_on_mode_change():
|
|
planner = display.RefreshPlanner(0.0)
|
|
planner.record_draw(0.0, ["RETTUNG"], "recovery", was_full=True)
|
|
|
|
assert planner.decide(5.0, ["MIGRATION"], "migration") == "full"
|