"""Tests fuer den Tasterdienst (tools/station/button/button.py). Die Bedienlogik ist reine Zustandsfuehrung und wird hier OHNE Hardware geprueft - aber bewusst NICHT, indem der Ablauf im Test nachgebaut wird: getrieben wird die ECHTE Hauptschleife button.run_daemon() mit einer Fake-Uhr, einem nach Zeitplan gedrueckten Taster-Doppel, dem ECHTEN CliActions (nur dessen subprocess.run ist ersetzt) und einem protokollierenden Schreiber. Der Grund steht in display.py/test_render.py: bei der Zeitsteuerung der Anzeige trieben die Tests eine KOPIE der Verdrahtung zwischen Zustandsmaschine und Ausfuehrung, und man konnte die eigentliche Aktion durch eine falsche ersetzen, ohne dass ein Test rot wurde. Deshalb laeuft hier alles ueber dieselbe Schleife, die auch auf dem Pi laeuft.""" import json import os import subprocess import sys import types from pathlib import Path import pytest sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "button")) import button # noqa: E402 # --- Zustaende, von denen die Menueeintraege abhaengen ------------------------------- STATE_OFF = {"running": False, "mode": "recovery", "target": None, "imageset": "ohne-wlan"} STATE_RECOVERY = {"running": True, "mode": "recovery", "target": None, "imageset": "ohne-wlan"} STATE_MIGRATION = {"running": True, "mode": "migration", "target": "gu10_rgb", "imageset": "ohne-wlan"} STATE_MIGRATION_BLIND = {"running": True, "mode": "migration", "target": None, "imageset": "ohne-wlan"} # --- Fake-Uhr, Taster-Doppel, Protokoll --------------------------------------------- class _StopDaemon(Exception): """Bricht run_daemon()s Endlosschleife ab, sobald die simulierte Zeit abgelaufen ist.""" class _FakeClock: def __init__(self, until_s: float) -> None: self.t = 0.0 self.until_s = until_s def monotonic(self) -> float: return self.t def advance(self, seconds: float) -> None: self.t += seconds def sleep(self, seconds: float) -> None: self.t += seconds if self.t > self.until_s: raise _StopDaemon class _Recorder: """Gemeinsame Zeitleiste von Schreiber und Ausfuehrung - nur so faellt auf, wenn die Reihenfolge in run_daemon() kippt (z.B. das "Bitte warten" erst NACH dem sekundenlangen CLI-Aufruf geschrieben wuerde).""" def __init__(self, clock: _FakeClock) -> None: self.clock = clock self.events: list[tuple[float, str, object]] = [] def note(self, kind: str, payload: object) -> None: self.events.append((self.clock.monotonic(), kind, payload)) def views(self) -> list[tuple[float, dict]]: return [(t, p) for t, k, p in self.events if k == "view"] def runs(self) -> list[tuple[float, list]]: return [(t, p) for t, k, p in self.events if k == "run"] def argvs(self) -> list[list]: return [p for _t, p in self.runs()] def kinds(self) -> list[str]: return [v.get("view") for _t, v in self.views()] def first_view(self, name: str): for t, v in self.views(): if v.get("view") == name: return t, v return None class _ScriptedButton: """Taster-Doppel: gedrueckt genau waehrend der angegebenen Zeitfenster. `is_pressed` ist wie bei gpiozero ein PEGEL (die Entprellung macht dort bounce_time), keine Flanke - genau das, was run_daemon() abfragt.""" def __init__(self, clock: _FakeClock, presses) -> None: self.clock = clock self.presses = list(presses) @property def is_pressed(self) -> bool: t = self.clock.monotonic() return any(start <= t < end for start, end in self.presses) class _RecordingWriter: """Schreiber-Doppel: protokolliert jeden tatsaechlich geschriebenen Bedienzustand. Dedupliziert wie der echte UiWriter, damit die Zeitleiste die ANZEIGEWECHSEL zeigt und nicht 50 identische Zeilen je Sekunde.""" def __init__(self, rec: _Recorder) -> None: self.rec = rec self.path = Path("/dev/null") self._last = None def write(self, view: dict) -> bool: if view == self._last: return False self._last = dict(view) self.rec.note("view", dict(view)) return True def _fake_runner(clock: _FakeClock, rec: _Recorder, duration_s: float = 0.0, rc: int = 0): """Ersetzt subprocess.run in CliActions - der Rest von CliActions (und damit die ZUSAMMENSETZUNG der Kommandozeile, auf die es sicherheitsrelevant ankommt) ist echt.""" def run(argv, check=False, timeout=None): rec.note("run", list(argv)) # Ein CLI-Aufruf dauert Sekunden (Profilumschaltung) - die simulierte Uhr muss # das mitmachen, sonst pruefen die Tests eine Schleife, die es so nicht gibt. clock.advance(duration_s) return types.SimpleNamespace(returncode=rc) return run def _drive(presses, duration_s: float, state=STATE_OFF, action_duration_s: float = 0.0, rc: int = 0) -> _Recorder: """Laesst das echte button.run_daemon() `duration_s` simulierte Sekunden laufen.""" clock = _FakeClock(duration_s) rec = _Recorder(clock) actions = button.CliActions(cli="/usr/local/bin/ninlight-station", runner=_fake_runner(clock, rec, action_duration_s, rc)) try: button.run_daemon(button=_ScriptedButton(clock, presses), writer=_RecordingWriter(rec), actions=actions, clock=clock, read_state=lambda: dict(state)) except _StopDaemon: pass return rec def _short(at: float): """Ein kurzer Druck (deutlich unter SHORT_PRESS_MAX_S, deutlich ueber DEBOUNCE_S).""" return (at, at + 0.2) # --- Menueaufbau -------------------------------------------------------------------- def _labels(state): return [a.label for a in button.menu_actions(state)] def test_menu_off_offers_starting_and_shutdown(): assert _labels(STATE_OFF) == ["Rettung starten", "Migration: horchen", "Migration: AUTOMATIK", "Umschalten: mit WLAN", "Herunterfahren"] def test_the_listening_migration_comes_before_the_automatic_one(): """Der horchende Modus ist der erste und naheliegende Eintrag - wer nur blaettert und irgendwo loslaesst, soll nicht im Automatikmodus landen.""" labels = _labels(STATE_OFF) assert labels.index("Migration: horchen") < labels.index("Migration: AUTOMATIK") def test_menu_in_recovery_offers_stopping_refreshing_and_shutdown(): assert _labels(STATE_RECOVERY) == ["Station aus", "Umschalten: mit WLAN", "Anzeige auffrischen", "Herunterfahren"] def test_menu_in_migration_offers_the_detected_target(): assert _labels(STATE_MIGRATION) == ["Freigabe: gu10_rgb", "Station aus", "Umschalten: mit WLAN", "Anzeige auffrischen", "Herunterfahren"] def test_menu_in_migration_without_a_detected_target_offers_no_release(): """Ohne last_seen gibt es nichts freizugeben - der Eintrag faellt ganz weg, statt eine Freigabe ins Leere anzubieten (die das CLI ohnehin abweisen wuerde).""" assert "Freigabe: " not in " ".join(_labels(STATE_MIGRATION_BLIND)) assert _labels(STATE_MIGRATION_BLIND) == ["Station aus", "Umschalten: mit WLAN", "Anzeige auffrischen", "Herunterfahren"] def test_only_the_refresh_entry_skips_the_confirmation(): """Zusage: JEDE veraendernde Auswahl bekommt eine Rueckfrage. "Anzeige auffrischen" veraendert nichts und braucht keine.""" for state in (STATE_OFF, STATE_RECOVERY, STATE_MIGRATION): for action in button.menu_actions(state): assert action.confirm is (action.label != "Anzeige auffrischen"), action.label def test_actions_run_through_the_station_cli(): """Die Aktionen gehen ueber das vorhandene CLI, nicht durch nachgebaute Logik - und das Herunterfahren als einzige an systemd selbst.""" actions = button.CliActions(cli="/usr/local/bin/ninlight-station") assert actions.argv_for(button.menu_actions(STATE_OFF)[0]) == [ "/usr/local/bin/ninlight-station", "recovery", "on"] assert actions.argv_for(button.action_shutdown()) == ["systemctl", "poweroff"] # --- Bedienung, getrieben durch die echte Hauptschleife ------------------------------ def test_a_short_press_opens_the_menu(): rec = _drive([_short(1.0)], 4.0) assert rec.argvs() == [] t, view = rec.first_view("menu") assert view["hero"] == "> Rettung starten" assert view["active"] is True # Spuerbar, aber bedienbar: das Menue steht binnen eines Schleifentakts nach dem # Loslassen (die Verzoegerung am Panel kommt aus dessen eigenem Takt, siehe # display.UI_POLL_INTERVAL_S). assert 1.2 <= t <= 1.3 def test_a_short_press_steps_to_the_next_entry(): rec = _drive([_short(1.0), _short(2.0)], 4.0) assert rec.argvs() == [] assert rec.views()[-1][1]["hero"] == "> Migration: horchen" def test_a_short_press_wraps_around_the_last_entry(): rec = _drive([_short(1.0), _short(2.0), _short(3.0), _short(4.0), _short(5.0), _short(6.0)], 8.0) assert rec.views()[-1][1]["hero"] == "> Rettung starten" def test_holding_one_and_a_half_seconds_selects_the_entry(): rec = _drive([_short(1.0), (2.0, 4.0)], 6.0) t, view = rec.first_view("confirm") assert view["hero"] == "Rettung starten" assert 3.5 <= t <= 3.6 # Die Auswahl allein fuehrt noch nichts aus - dafuer ist die Rueckfrage da. assert rec.argvs() == [] def test_holding_in_the_confirmation_executes_the_entry(): rec = _drive([_short(1.0), (2.0, 4.0), (5.0, 7.0)], 9.0) assert rec.argvs() == [["/usr/local/bin/ninlight-station", "recovery", "on"]] assert 6.5 <= rec.runs()[0][0] <= 6.6 def test_a_short_press_in_the_confirmation_cancels_and_returns_to_the_menu(): rec = _drive([_short(1.0), (2.0, 4.0), _short(5.0)], 8.0) assert rec.argvs() == [] assert rec.views()[-1][1]["view"] == "menu" def test_the_refresh_entry_runs_without_a_confirmation(): rec = _drive([_short(1.0), _short(2.0), _short(3.0), (4.0, 6.0)], 8.0, state=STATE_RECOVERY) assert rec.argvs() == [["/usr/local/bin/ninlight-station", "display", "test"]] assert "confirm" not in rec.kinds() def test_the_listening_entry_calls_the_cli_with_listen(): """Der zweite Eintrag ("Migration: horchen") ruft das CLI MIT --listen: der Zugangspunkt steht, es geht aber nichts raus, bis jemand eine erkannte Lampe freigibt.""" rec = _drive([_short(1.0), _short(2.0), (3.0, 5.0), (6.0, 8.0)], 10.0) assert rec.argvs() == [["/usr/local/bin/ninlight-station", "migrate", "on", "--listen"]] def test_the_automatic_entry_calls_the_cli_without_listen(): """Der dritte Eintrag ("Migration: AUTOMATIK") ruft dasselbe CLI OHNE --listen - das ist der ganze Unterschied zwischen den beiden Eintraegen, und er entscheidet, ob fremde Lampen ungefragt migriert werden.""" rec = _drive([_short(1.0), _short(2.0), _short(3.0), (4.0, 6.0), (7.0, 9.0)], 11.0) assert rec.argvs() == [["/usr/local/bin/ninlight-station", "migrate", "on"]] assert "--listen" not in rec.argvs()[0] def test_the_automatic_entry_asks_a_louder_question_than_the_others(): """Zusage: die Rueckfrage des Automatikmodus benennt die Folge, statt nur "Wirklich?" zu fragen - und sie unterscheidet sich sichtbar von der der uebrigen Eintraege.""" rec = _drive([_short(1.0), _short(2.0), _short(3.0), (4.0, 6.0)], 8.0) _t, view = rec.first_view("confirm") assert view["title"] == "ACHTUNG" assert view["hero"] == "Migration: AUTOMATIK" text = " ".join(view["details"]).lower() assert "jede" in text and "fremde" in text # Die Bedienhinweise bleiben trotzdem stehen, und zwar zuletzt. assert view["details"][-2:] == ["halten (1,5 s): ja", "kurz: nein"] # Gegenprobe am horchenden Eintrag: uebliche Rueckfrage, keine Warnung. rec = _drive([_short(1.0), _short(2.0), (3.0, 5.0)], 7.0) _t, plain = rec.first_view("confirm") assert plain["title"] == "RUECKFRAGE" assert plain["details"] == ["halten (1,5 s): ja", "kurz: nein"] def test_the_cancel_fires_at_three_seconds_while_the_button_is_still_held(): """Der Abbruch feuert BEI 3 s, nicht erst beim Loslassen - der Bediener soll die Wirkung am Panel sehen, waehrend er noch haelt.""" rec = _drive([(1.0, 8.0)], 10.0) assert rec.argvs() == [["/usr/local/bin/ninlight-station", "off"]] t = rec.runs()[0][0] assert 4.0 <= t <= 4.1 # ... und danach laeuft am Panel der Countdown der zweiten Stufe. later = [v for vt, v in rec.views() if vt > t] assert any("Herunterfahren in" in v.get("hero", "") for v in later) def test_releasing_between_three_and_ten_seconds_does_not_shut_down(): rec = _drive([(1.0, 7.0)], 12.0) assert rec.argvs() == [["/usr/local/bin/ninlight-station", "off"]] assert ["systemctl", "poweroff"] not in rec.argvs() def test_holding_ten_seconds_shuts_down(): rec = _drive([(1.0, 12.0)], 14.0) assert rec.argvs() == [["/usr/local/bin/ninlight-station", "off"], ["systemctl", "poweroff"]] assert 11.0 <= rec.runs()[1][0] <= 11.1 def test_the_shutdown_countdown_is_visible_before_it_happens(): """Das Herunterfahren ist nur ueber eine SICHTBARE Zustandsaenderung hinweg erreichbar - der Countdown muss also vorher auf dem Panel gestanden haben.""" rec = _drive([(1.0, 12.0)], 14.0) heroes = [v.get("hero", "") for t, v in rec.views() if t < 11.0] assert "Herunterfahren in 6 s" in heroes assert "Herunterfahren in 1 s" in heroes def test_a_blocking_action_does_not_break_the_hold_gesture(): """Der Abbruch-Aufruf blockiert die Schleife sekundenlang (Profilumschaltung). Danach wird derselbe PEGEL gegen dieselbe Uhr gemessen - die zweite Stufe muss trotzdem erreichbar bleiben.""" rec = _drive([(1.0, 12.0)], 14.0, action_duration_s=2.0) assert rec.argvs()[-1] == ["systemctl", "poweroff"] def test_the_waiting_screen_is_written_before_the_slow_call(): rec = _drive([_short(1.0), _short(2.0), _short(3.0), (4.0, 6.0)], 9.0, state=STATE_RECOVERY, action_duration_s=4.0) kinds = [k for _t, k, _p in rec.events] run_at = kinds.index("run") assert rec.events[run_at - 1][2].get("view") == "busy" def test_the_menu_closes_after_twenty_seconds_without_input(): rec = _drive([_short(1.0)], 26.0) assert rec.kinds()[-1] == "idle" closed_at = [t for t, v in rec.views() if v.get("view") == "idle" and t > 2.0][0] assert 21.2 <= closed_at <= 21.4 def test_the_confirmation_expires_after_ten_seconds_as_a_cancel(): rec = _drive([_short(1.0), (2.0, 4.0)], 18.0) assert rec.argvs() == [] assert rec.kinds()[-1] == "idle" expired_at = [t for t, v in rec.views() if v.get("view") == "idle" and t > 4.0][0] assert 13.5 <= expired_at <= 13.7 def test_an_expired_confirmation_leaves_nothing_pending(): """Der Abbruch muss die vorgemerkte Aktion auch wirklich vergessen - eine Aktion, die nach dem Verfall noch in der Maschine liegt, ist genau das, was eine Rueckfrage verhindern soll.""" ui = button.ButtonUI(0.0, read_state=lambda: dict(STATE_OFF)) ui.update(1.0, True) ui.update(1.2, False) # kurzer Druck: Menue auf ui.update(2.0, True) ui.update(3.6, True) # 1,6 s gehalten: Eintrag gewaehlt -> Rueckfrage ui.update(3.8, False) assert ui.screen == "confirm" and ui.pending is not None ui.update(3.6 + button.CONFIRM_TIMEOUT_S + 0.1, False) assert ui.screen == "idle" assert ui.pending is None def test_a_failed_action_says_so_on_the_panel(): rec = _drive([_short(1.0), (2.0, 4.0), (5.0, 7.0)], 9.0, rc=1) assert any(v.get("view") == "result" and v.get("title") == "FEHLER" for _t, v in rec.views()) def test_nothing_happens_without_any_press(): rec = _drive([], 30.0) assert rec.argvs() == [] assert rec.kinds() == ["idle"] # --- Bedienzustandsdatei ------------------------------------------------------------ def test_the_ui_file_is_written_atomically_with_0644(tmp_path): """Dasselbe Muster und dieselben Rechte wie status.json/control.json - ohne 0644 kann der Anzeigedienst die Datei nicht lesen, ohne das Umbenennen sieht er sie halb geschrieben.""" path = tmp_path / "run" / "ui.json" writer = button.UiWriter(path) view = {"view": "menu", "title": "MENUE", "hero": "> Station aus", "details": [], "active": True} # Unter einer strengen umask geschrieben: die Rechte muessen ausdruecklich gesetzt # werden, nicht zufaellig aus der umask des Dienstes fallen - sonst laege die Zusage # "der Anzeigedienst kann die Datei lesen" an einer Umgebungsvariablen des Systems. old_umask = os.umask(0o077) try: assert writer.write(view) is True finally: os.umask(old_umask) assert json.loads(path.read_text()) == view assert oct(os.stat(path).st_mode & 0o777) == "0o644" # Keine Temp-Datei zurueckgelassen (sie wuerde sonst mit jedem Neustart ein weiteres # Bruchstueck im RuntimeDirectory hinterlassen). assert [p.name for p in path.parent.iterdir()] == ["ui.json"] def test_the_ui_file_is_only_written_on_change(tmp_path): writer = button.UiWriter(tmp_path / "ui.json") view = {"view": "idle", "active": False} assert writer.write(view) is True assert writer.write(dict(view)) is False def test_an_unwritable_ui_file_is_survived_and_logged_once(tmp_path, caplog): """Ein Fehler am Taster darf niemals eine laufende Migration stoeren - und eine Zeile je Schleifentakt wuerde das Journal fluten.""" blocker = tmp_path / "blocker" blocker.write_text("kein Verzeichnis") writer = button.UiWriter(blocker / "ui.json") with caplog.at_level("WARNING", logger="ninlight-button"): assert writer.write({"view": "menu", "active": True}) is False assert writer.write({"view": "confirm", "active": True}) is False assert len(caplog.records) == 1 # --- Der Taster ist nirgends Voraussetzung ------------------------------------------ def test_missing_gpiozero_logs_exactly_one_line_and_exits_zero(monkeypatch, caplog): """Dieselbe Zusage wie bei der Anzeige (dort seit fuenf Fixrunden): kein Taster -> GENAU EINE Protokollzeile und sys.exit(0), kein Fehler - sonst startet systemd (Restart=on-failure) den Dienst endlos neu und flutet das Journal.""" monkeypatch.setitem(sys.modules, "gpiozero", None) with caplog.at_level("INFO", logger="ninlight-button"): with pytest.raises(SystemExit) as excinfo: button.main() assert excinfo.value.code == 0 assert len(caplog.records) == 1 assert "gpiozero" in caplog.records[0].getMessage() def test_a_busy_pin_is_treated_like_a_missing_button(monkeypatch, caplog): """"GPIO busy" kommt je nach Pin-Factory als gpiozero-Fehler, als roher lgpio.error oder als OSError - alles davon heisst "kein Taster", nicht "abstuerzen".""" class _Busy(Exception): pass fake = types.SimpleNamespace( Button=lambda *a, **k: (_ for _ in ()).throw(_Busy("GPIO busy"))) monkeypatch.setitem(sys.modules, "gpiozero", fake) with caplog.at_level("INFO", logger="ninlight-button"): with pytest.raises(SystemExit) as excinfo: button.main() assert excinfo.value.code == 0 assert len(caplog.records) == 1 def test_a_failing_cli_call_never_escapes(monkeypatch): """subprocess-Fehler (Binary fehlt, Timeout) duerfen den Dienst nicht beenden.""" def boom(argv, check=False, timeout=None): raise OSError("kein CLI da") actions = button.CliActions(runner=boom) assert actions.run(button.action_cancel()) is False def slow(argv, check=False, timeout=None): raise subprocess.TimeoutExpired(argv, timeout) assert button.CliActions(runner=slow).run(button.action_cancel()) is False def test_an_unreadable_station_state_keeps_the_menu_closed(caplog): """Menueeintraege haengen am Stationszustand. Ist der nicht zu ermitteln, bleibt das Menue zu, statt einen falschen Zustand zu behaupten - und der Dienst laeuft weiter.""" ui = button.ButtonUI(0.0, read_state=lambda: (_ for _ in ()).throw(RuntimeError("weg"))) with caplog.at_level("WARNING", logger="ninlight-button"): ui.update(0.0, True) assert ui.update(0.2, False) is None assert ui.view(0.3) == {"view": "idle", "active": False} # --- Bildsatz-Umschalter im Menue ---------------------------------------------------- def test_the_image_set_entry_names_where_it_switches_to(): """Waehrend das Menue offen ist, ersetzt es das Statusbild samt Fusszeile - der Eintrag muss deshalb das ZIEL nennen, nicht den jetzigen Zustand.""" assert "Umschalten: mit WLAN" in _labels(STATE_OFF) ohne = dict(STATE_OFF, imageset="mit-wlan") assert "Umschalten: ohne WLAN" in _labels(ohne) assert "Umschalten: mit WLAN" not in _labels(ohne) def test_the_image_set_entry_switches_through_the_cli(): actions = button.CliActions(cli="/usr/local/bin/ninlight-station") assert actions.argv_for(button.action_imageset("ohne-wlan")) == [ "/usr/local/bin/ninlight-station", "bildsatz", "mit-wlan"] assert actions.argv_for(button.action_imageset("mit-wlan")) == [ "/usr/local/bin/ninlight-station", "bildsatz", "ohne-wlan"] def test_switching_to_wifi_credentials_asks_louder_than_switching_back(): loud = button.action_imageset("ohne-wlan") back = button.action_imageset("mit-wlan") assert loud.confirm and back.confirm assert loud.confirm_title == "ACHTUNG" assert back.confirm_title == "RUECKFRAGE" assert "WLAN-Passwort" in " ".join(loud.confirm_details) assert back.confirm_details is None def test_the_image_set_entry_really_runs_the_switch(tmp_path): """Durch die echte Hauptschleife, nicht nur ueber menu_actions(): vierter Eintrag im ausgeschalteten Zustand, Auswahl halten, Rueckfrage halten.""" rec = _drive([_short(1.0), _short(2.0), _short(3.0), _short(4.0), (5.0, 7.0), (8.0, 10.0)], 12.0) assert rec.argvs() == [["/usr/local/bin/ninlight-station", "bildsatz", "mit-wlan"]] def test_station_state_reads_the_same_switch_file_as_the_cli(tmp_path, monkeypatch): monkeypatch.setattr(button, "_service_active", lambda service_name: False) switch = tmp_path / "imageset" state = lambda: button.station_state(run_dir=tmp_path, service_name="x", imageset_path=switch) assert state()["imageset"] == "ohne-wlan" # Datei fehlt -> sichere Vorgabe switch.write_text("mit-wlan\n") assert state()["imageset"] == "mit-wlan" switch.write_text("quatsch") assert state()["imageset"] == "ohne-wlan"