Michael Sauer
bb39105ac4
Zwei Befunde der Hardware-Abnahme. 1) Die Schrift war zu klein: compose() zeichnete alles in ImageFont.load_default() (rund 10 px) - auf 250x122 aus zwei Metern nicht lesbar. blocks() vergibt jetzt Rollen (header/hero/detail/big), compose() setzt jede Rolle eigen: Modus fett und weiss auf schwarzem Balken, die wichtigste Zeile gross (Migration: erkannte Lampe, Rettung: SSID), Beiwerk klein, "Station aus" gross und mittig. Die Schrift kommt per ImageFont.truetype() aus FONT_SEARCH_PATHS mit DejaVu (/usr/share/fonts/truetype/dejavu/) als Hauptweg - auf beiden Pis vorhanden. load_default(size=...) scheidet aus: das gibt es erst ab Pillow 10.1, Pi 2 hat 9.4.0. Letzter Rueckfall bleibt load_default() ohne Argument, damit eine fehlende Schrift die Anzeige (und damit nie eine Migration) aufhaelt. lines() bleibt ohne Pillow importierbar, die Breitenkuerzung misst weiter in Pixeln. 2) "display test" scheiterte bei laufendem Anzeigedienst mit einem rohen "lgpio.error: 'GPIO busy'": lgpio.error ist keine gpiozero.GPIOZeroError und fiel durch die Ausnahmeliste in epdconfig.module_init(). Sie ist jetzt aufgefuehrt - weiterhin gezielt, TypeError/AttributeError/NameError bleiben Programmierfehler - und die Meldung nennt bei belegten Leitungen den Anzeigedienst. "display test|clear" stoppt den Dienst fuer die Dauer des Handgriffs und startet ihn danach wieder, aber nur, wenn er vorher lief.
1031 lines
45 KiB
Python
1031 lines
45 KiB
Python
import json
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
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).
|
|
|
|
Geprueft wird erst UNTERHALB des Kopfbalkens: der ist seit der Schriftumstellung
|
|
absichtlich vollflaechig schwarz (weisse Schrift auf schwarzem Grund) und faerbt die
|
|
letzte Spalte in seinen eigenen Zeilen mit - das ist Layout, keine ueberlaufende
|
|
Zeile. Die absichtlich zu lange Zeile (der Produktname) steht ohnehin darunter."""
|
|
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.HEADER_BAR_HEIGHT_PX, render.PANEL_HEIGHT_PX))
|
|
assert not black_in_last_column, "Text erreicht die letzte Spalte - Kuerzung wirkt nicht"
|
|
|
|
|
|
# --- Schriftgroesse/Layout (Hardware-Abnahme: "die Schrift ist zu klein") -----------
|
|
#
|
|
# Alle Tests hier messen die TINTE im tatsaechlich gezeichneten Bild, nicht die
|
|
# Konstanten oben: eine Zusage ueber die Schriftgroesse, die nur die Konstante liest,
|
|
# haelt niemanden auf, der die Schriftsuche entfernt und wieder auf
|
|
# ImageFont.load_default() (rund 10 px) zurueckfaellt - genau die Mutation, gegen die
|
|
# diese Tests wachen sollen.
|
|
|
|
def _ink_box(image, y0=0, y1=None):
|
|
"""(links, oben, rechts, unten) der schwarzen Pixel im Band [y0, y1), oder None."""
|
|
y1 = image.height if y1 is None else y1
|
|
pixels = image.load()
|
|
xs, ys = [], []
|
|
for y in range(y0, y1):
|
|
for x in range(image.width):
|
|
if pixels[x, y] == 0:
|
|
xs.append(x)
|
|
ys.append(y)
|
|
if not xs:
|
|
return None
|
|
return (min(xs), min(ys), max(xs), max(ys))
|
|
|
|
|
|
def _migration_state():
|
|
return {"mode": "migration", "phase": "armed", "armed_target": "gu10_rgb",
|
|
"served": 3, "ssid": "qconnex-firmware",
|
|
"last_seen": {"product_name": "Klyqa GU10 RGB", "target": "gu10_rgb",
|
|
"variant": "prod", "lamp_version": "V1.1.0.1138"},
|
|
"recent": [{"target": "gu10_rgb", "status": "ok", "ts": "2026-09-13T13:04:02Z"}]}
|
|
|
|
|
|
def test_off_screen_is_drawn_big_and_centered():
|
|
""""Station aus" hat fast keinen Inhalt und muss aus zwei Metern lesbar sein.
|
|
|
|
Gemessen mit echtem Pillow: mit der Schriftsuche ist die Tinte 19 px hoch und links
|
|
wie rechts 56 px vom Rand entfernt; faellt compose() auf load_default() zurueck, sind
|
|
es 8 px - die Schwelle 14 trennt beide Faelle eindeutig."""
|
|
image = render.compose({"mode": "off"})
|
|
box = _ink_box(image)
|
|
assert box is not None, "gar nichts gezeichnet"
|
|
|
|
height = box[3] - box[1] + 1
|
|
assert height >= 14, f"Schrift nur {height} px hoch - Schriftsuche wirkt nicht"
|
|
|
|
left, right = box[0], render.PANEL_WIDTH_PX - 1 - box[2]
|
|
assert abs(left - right) <= 3, f"nicht waagerecht mittig (links {left}, rechts {right})"
|
|
top, bottom = box[1], render.PANEL_HEIGHT_PX - 1 - box[3]
|
|
assert abs(top - bottom) <= 4, f"nicht senkrecht mittig (oben {top}, unten {bottom})"
|
|
|
|
|
|
def test_header_is_a_black_bar_with_white_text():
|
|
"""Die Kopfzeile ist invertiert: schwarzer Balken ueber die volle Breite, Modus in
|
|
Weiss darin. Geprueft an einer Zeile mitten im Balken - die muss ueberwiegend schwarz
|
|
sein UND weisse Pixel enthalten (sonst stuende dort kein Text, sondern nur ein
|
|
Balken)."""
|
|
image = render.compose(_migration_state())
|
|
pixels = image.load()
|
|
row = render.HEADER_BAR_HEIGHT_PX // 2
|
|
|
|
black = sum(1 for x in range(render.PANEL_WIDTH_PX) if pixels[x, row] == 0)
|
|
white = sum(1 for x in range(render.PANEL_WIDTH_PX) if pixels[x, row] != 0)
|
|
assert black > 150, f"Kopfzeile ist kein schwarzer Balken (nur {black} schwarze Pixel)"
|
|
assert white >= 5, "keine weisse Schrift im Balken - der Balken uebermalt den Modus"
|
|
|
|
|
|
def test_hero_line_is_bigger_than_the_details():
|
|
"""Die wichtigste Zeile (hier: der erkannte Produktname) muss sich vom Beiwerk
|
|
abheben - sonst ist die Rollenvergabe in blocks() folgenlos. Gemessen: 15 px Tinte
|
|
fuer die grosse Zeile gegen 10 px fuer die erste Beiwerkzeile."""
|
|
image = render.compose(_migration_state())
|
|
hero_top = render.HEADER_BAR_HEIGHT_PX + render.TEXT_MARGIN_PX
|
|
hero_bottom = hero_top + render.HERO_LINE_HEIGHT_PX
|
|
|
|
hero = _ink_box(image, hero_top, hero_bottom)
|
|
detail = _ink_box(image, hero_bottom, hero_bottom + render.DETAIL_LINE_HEIGHT_PX)
|
|
assert hero is not None and detail is not None
|
|
|
|
hero_height = hero[3] - hero[1] + 1
|
|
detail_height = detail[3] - detail[1] + 1
|
|
assert hero_height >= detail_height + 3, (
|
|
f"grosse Zeile ({hero_height} px) hebt sich nicht vom Beiwerk "
|
|
f"({detail_height} px) ab")
|
|
|
|
|
|
def test_the_last_delivery_line_still_fits_on_the_panel():
|
|
"""Der vollste Zustand (Kopfbalken, grosse Zeile, sechs Beiwerkzeilen) muss
|
|
vollstaendig aufs Panel passen - mit groesserer Schrift ist das keine
|
|
Selbstverstaendlichkeit mehr, und ausgerechnet die letzte Zeile ("zuletzt
|
|
ausgeliefert") faellt sonst stumm hinten herunter."""
|
|
state = _migration_state()
|
|
expected = render.lines(state)
|
|
assert len(expected) == 8, expected # sonst misst dieser Test etwas anderes
|
|
|
|
image = render.compose(state)
|
|
last_line_top = (render.HEADER_BAR_HEIGHT_PX + render.TEXT_MARGIN_PX
|
|
+ render.HERO_LINE_HEIGHT_PX + 5 * render.DETAIL_LINE_HEIGHT_PX)
|
|
assert _ink_box(image, last_line_top, render.PANEL_HEIGHT_PX) is not None, (
|
|
"letzte Zeile fehlt - sie passt nicht mehr aufs Panel")
|
|
|
|
|
|
def test_compose_still_works_without_any_font_file(monkeypatch):
|
|
"""Die Zusage "eine fehlende Schrift haelt die Anzeige nicht auf" gilt unveraendert:
|
|
ohne einen einzigen Treffer in der Suchkette faellt compose() auf load_default()
|
|
zurueck und zeichnet - klein, aber vorhanden."""
|
|
monkeypatch.setattr(render, "FONT_SEARCH_PATHS", ())
|
|
image = render.compose({"mode": "off"})
|
|
assert _ink_box(image) is not None
|
|
|
|
|
|
# --- Die Schriftsuche selbst, ohne echte Schriftdateien und ohne altes Pillow -------
|
|
|
|
class _FakeImageFontModule:
|
|
"""Nachbau der zwei ImageFont-Funktionen, die _load_font() benutzt.
|
|
|
|
`available` sind die Pfade, die es "gibt"; jeder andere truetype()-Aufruf scheitert
|
|
mit OSError, genau wie Pillow es bei einer fehlenden Datei tut.
|
|
|
|
load_default() bildet Pillow 9.4.0 auf Pi 2 nach: KEIN size-Argument (das gibt es
|
|
erst ab 10.1). Ein `size=`-Aufruf schlaegt hier deshalb mit demselben TypeError fehl
|
|
wie am Geraet - so laesst sich die 9.4-Lage pruefen, ohne Pillow herunterzustufen."""
|
|
|
|
def __init__(self, available=()):
|
|
self.available = set(available)
|
|
self.truetype_calls = []
|
|
self.load_default_calls = 0
|
|
|
|
def truetype(self, path, size):
|
|
self.truetype_calls.append((path, size))
|
|
if path not in self.available:
|
|
raise OSError(f"cannot open resource: {path}")
|
|
return ("truetype", path, size)
|
|
|
|
def load_default(self, *args, **kwargs):
|
|
self.load_default_calls += 1
|
|
if args or kwargs:
|
|
raise TypeError(
|
|
"load_default() got an unexpected keyword argument 'size'")
|
|
return ("default",)
|
|
|
|
|
|
DEJAVU_REGULAR = "/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf"
|
|
DEJAVU_BOLD = "/usr/share/fonts/truetype/dejavu/DejaVuSans-Bold.ttf"
|
|
|
|
|
|
def test_font_search_takes_dejavu_first_on_the_pi():
|
|
"""Auf beiden Pis liegt DejaVu unter /usr/share/fonts/truetype/dejavu/ - das ist der
|
|
Hauptweg, und er muss VOR jedem Rueckfall kommen. Rot, sobald jemand die Suche
|
|
entfernt und wieder direkt auf load_default() geht."""
|
|
image_font = _FakeImageFontModule(available=(DEJAVU_REGULAR, DEJAVU_BOLD))
|
|
|
|
regular = render._load_font(image_font, 11)
|
|
bold = render._load_font(image_font, 20, bold=True)
|
|
|
|
assert regular == ("truetype", DEJAVU_REGULAR, 11)
|
|
assert bold == ("truetype", DEJAVU_BOLD, 20)
|
|
assert image_font.truetype_calls[0] == (DEJAVU_REGULAR, 11), \
|
|
"DejaVu wurde nicht zuerst probiert"
|
|
assert image_font.load_default_calls == 0, \
|
|
"load_default() benutzt, obwohl DejaVu da ist - die Schrift waere wieder winzig"
|
|
|
|
|
|
def test_font_search_falls_through_to_a_later_candidate():
|
|
"""Auf der Entwicklungsmaschine (macOS) gibt es den Debian-Pfad nicht - dann muss die
|
|
Kette weiterlaufen statt sofort auf load_default() zu fallen, sonst laeuft der Test
|
|
oben dort gegen eine 10-px-Schrift."""
|
|
arial = "/System/Library/Fonts/Supplemental/Arial.ttf"
|
|
image_font = _FakeImageFontModule(available=(arial,))
|
|
|
|
assert render._load_font(image_font, 16) == ("truetype", arial, 16)
|
|
assert image_font.load_default_calls == 0
|
|
|
|
|
|
def test_font_search_ends_at_load_default_without_a_size_argument():
|
|
"""Der letzte Rueckfall auf einem Rechner ohne jede Schriftdatei. Entscheidend ist,
|
|
dass load_default() OHNE size gerufen wird: auf Pi 2 (Pillow 9.4.0) waere
|
|
load_default(size=...) ein TypeError - der Fake wirft ihn hier genauso."""
|
|
image_font = _FakeImageFontModule()
|
|
|
|
assert render._load_font(image_font, 26, bold=True) == ("default",)
|
|
assert image_font.load_default_calls == 1
|
|
assert len(image_font.truetype_calls) == len(render.FONT_SEARCH_PATHS)
|
|
|
|
|
|
def test_font_search_survives_a_pillow_without_freetype():
|
|
"""Ein ohne FreeType gebautes Pillow laesst truetype() mit ImportError scheitern -
|
|
auch das darf die Anzeige nicht umbringen."""
|
|
|
|
class _NoFreetype(_FakeImageFontModule):
|
|
def truetype(self, path, size):
|
|
raise ImportError("The _imagingft C module is not installed")
|
|
|
|
image_font = _NoFreetype()
|
|
assert render._load_font(image_font, 20) == ("default",)
|
|
|
|
|
|
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)
|
|
|
|
|
|
# --- "GPIO busy" (Hardware-Abnahme, zweiter Befund) ---------------------------------
|
|
|
|
class _FakeLgpioModule:
|
|
"""Nur das, was epdconfig von lgpio kennt: die Ausnahmeklasse. Wie im echten Modul
|
|
erbt sie direkt von Exception - deshalb faengt eine Liste, die sie enthaelt, keinen
|
|
TypeError/AttributeError/NameError mit ein."""
|
|
|
|
class error(Exception):
|
|
pass
|
|
|
|
|
|
def _gpiozero_raising(exception):
|
|
"""Ein gpiozero-Ersatz, dessen erster Pin-Aufbau `exception` wirft."""
|
|
|
|
class _Raising:
|
|
GPIOZeroError = _FakeGpiozeroModule.GPIOZeroError
|
|
DigitalInputDevice = _FakeDigitalInputDevice
|
|
|
|
@staticmethod
|
|
def DigitalOutputDevice(pin):
|
|
raise exception
|
|
|
|
return _Raising
|
|
|
|
|
|
def test_lgpio_gpio_busy_becomes_display_unavailable(monkeypatch):
|
|
"""Der Befund vom Geraet: laeuft der Anzeigedienst, haelt er die GPIO-Leitungen, und
|
|
"ninlight-station display test" bekam ein rohes `lgpio.error: 'GPIO busy'` samt
|
|
Stapelabzug und rc=1 - lgpio.error ist KEINE gpiozero.GPIOZeroError und fiel durch
|
|
die Ausnahmeliste in module_init(). Erwartet ist stattdessen DisplayUnavailable,
|
|
das display.main() zu genau einer Protokollzeile und sys.exit(0) macht."""
|
|
_stub_hardware(monkeypatch)
|
|
monkeypatch.setattr(epdconfig, "lgpio", _FakeLgpioModule)
|
|
monkeypatch.setattr(epdconfig, "gpiozero",
|
|
_gpiozero_raising(_FakeLgpioModule.error("'GPIO busy'")))
|
|
|
|
with pytest.raises(epdconfig.DisplayUnavailable) as excinfo:
|
|
epdconfig.module_init()
|
|
|
|
message = str(excinfo.value)
|
|
assert "GPIO busy" in message
|
|
# Und die Meldung muss dem Bediener sagen, was hier praktisch immer los ist, statt
|
|
# ihn den lgpio-Fehler googeln zu lassen.
|
|
assert "ninlight-display" in message, f"kein Hinweis auf den Anzeigedienst: {message}"
|
|
|
|
|
|
def test_gpio_busy_hint_stays_away_from_unrelated_errors(monkeypatch):
|
|
"""Der Hinweis auf den Anzeigedienst gilt nur fuer belegte Leitungen - bei einem
|
|
anderen Fehler waere er eine Falschaussage, die den Bediener in die falsche Richtung
|
|
schickt."""
|
|
_stub_hardware(monkeypatch)
|
|
monkeypatch.setattr(epdconfig, "lgpio", _FakeLgpioModule)
|
|
monkeypatch.setattr(epdconfig, "gpiozero",
|
|
_gpiozero_raising(OSError("kein solches Geraet")))
|
|
|
|
with pytest.raises(epdconfig.DisplayUnavailable) as excinfo:
|
|
epdconfig.module_init()
|
|
|
|
assert "ninlight-display" not in str(excinfo.value)
|
|
|
|
|
|
def test_programming_errors_are_still_not_disguised_as_missing_hardware(monkeypatch):
|
|
"""Die Ausnahmeliste bleibt gezielt (Fixrunde 1, C2): ein TypeError ist ein
|
|
Programmierfehler und darf nicht als "Panel fehlt" durchgehen - sonst verschwindet er
|
|
hinter einer INFO-Zeile und sys.exit(0)."""
|
|
_stub_hardware(monkeypatch)
|
|
monkeypatch.setattr(epdconfig, "lgpio", _FakeLgpioModule)
|
|
monkeypatch.setattr(epdconfig, "gpiozero",
|
|
_gpiozero_raising(TypeError("falsches Argument")))
|
|
|
|
with pytest.raises(TypeError):
|
|
epdconfig.module_init()
|
|
|
|
|
|
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)
|
|
|
|
|
|
# --- 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, 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
|
|
|
|
|
|
class _StopDaemon(Exception):
|
|
"""Bricht run_daemon()s Endlosschleife ab, sobald die simulierte Zeit abgelaufen ist."""
|
|
|
|
|
|
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
|
|
|
|
|
|
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 _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 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."""
|
|
panel = _drive_daemon(lambda t: {"mode": "recovery", "served": 0, "recent": []},
|
|
SIX_HOURS_S)
|
|
|
|
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():
|
|
panel = _drive_daemon(_changing_every(3 * 60), SIX_HOURS_S)
|
|
|
|
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), (
|
|
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."""
|
|
panel = _drive_daemon(_changing_every(9 * 60), SIX_HOURS_S)
|
|
|
|
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)
|
|
|
|
|
|
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).
|
|
|
|
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(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(panel, "full") == [0.0], (
|
|
f"unerwartete Vollaufbauten: {_times(panel, '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 state_at(t):
|
|
# 30 Aenderungen im 5-Sekunden-Takt (t = 5 .. 150), danach unveraendert.
|
|
step = min(int(t // display.POLL_INTERVAL_S), 30)
|
|
return {"mode": "migration", "served": step, "recent": []}
|
|
|
|
panel = _drive_daemon(state_at, SIX_HOURS_S)
|
|
|
|
assert len(_times(panel, "partial")) == 30, (
|
|
f"das Buendel muss aus Teilaufbauten bestehen: {_times(panel, 'partial')}")
|
|
|
|
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(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():
|
|
"""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"
|