ninharp/ninlight-w600
1
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

feat(station): Rettung folgt dem Bildsatz-Umschalter, Server meldet Lebenszeichen

Browse Source 
server.py: image_path() liefert jetzt aus rescue_dir() (nach dem Umschalter
ohne-wlan/mit-wlan), genau wie migrate_dir() es fuer die Migration schon tut.
Bewusst kein Fallback auf den alten flachen Bestand in images/ - ein Pi mit
altem Bestand muss einmal per deploy.sh neu bestueckt werden, sonst wuerde
"ohne-wlan" den alten Zustand (immer mit Zugangsdaten) stillschweigend
fortsetzen.

Station.last_activity/"last_request_ts" in status.json: Zeitpunkt jeder
Anfrage (auch erfolgloser), damit die Anzeige ein Lebenszeichen zeigen kann,
statt nur am Servierzaehler abzulesen, ob ueberhaupt etwas ankommt.
This commit is contained in:
Michael Sauer
2026-09-23 21:22:22 +02:00
parent eec9fa5a21
commit ac8dfb1077
2 changed files with 166 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tools/station/tests/test_server.py
+102 -10
View File
@@ -43,6 +43,14 @@ class Station:
def __init__(self, tmp_path, *extra):
self.image_dir = tmp_path / "images"
self.image_dir.mkdir()
# image_path() folgt jetzt dem Bildsatz-Umschalter, genauso wie migrate_dir() es
# tut - ohne --imageset-file gilt die Vorgabe "ohne-wlan" (server.py,
# DEFAULT_IMAGE_SET), also landen Rettungsabbilder in image_dir/rescue-nocreds.
# Die meisten Tests hier pruefen nur die Ausliefermechanik, nicht den
# Bildsatz-Umschalter selbst - die Bildsatz-spezifischen Tests weiter unten legen
# ihre Abbilder gezielt in andere Verzeichnisse.
self.rescue_dir = self.image_dir / "rescue-nocreds"
self.rescue_dir.mkdir()
self.state = tmp_path / "status.json"
self.port = free_port()
self.proc = subprocess.Popen(
@@ -140,7 +148,7 @@ def station(tmp_path):
def test_serves_a_known_image(station):
data = make_image(station.image_dir / "ninlight_evk_ota.img", "evk", "1.2.3", b"payload" * 500)
data = make_image(station.rescue_dir / "ninlight_evk_ota.img", "evk", "1.2.3", b"payload" * 500)
c = station.conn()
c.request("GET", "/ninlight_evk_ota.img")
r = c.getresponse()
@@ -157,7 +165,7 @@ def test_missing_image_is_404(station):
def test_unknown_target_is_404(station):
make_image(station.image_dir / "ninlight_evk_ota.img", "evk", "1.0.0", b"x" * 100)
make_image(station.rescue_dir / "ninlight_evk_ota.img", "evk", "1.0.0", b"x" * 100)
for path in ("/ninlight_nosuch_ota.img", "/ninlight_evk_ota.img.bak", "/", "/images/"):
c = station.conn()
c.request("GET", path)
@@ -216,7 +224,7 @@ def test_log_error_is_not_silenced_by_log_message(capsys):
def test_range_request_returns_206_and_the_right_slice(station):
data = make_image(station.image_dir / "ninlight_e27_rgb_ota.img", "e27_rgb", "2.0.0", b"abc" * 2000)
data = make_image(station.rescue_dir / "ninlight_e27_rgb_ota.img", "e27_rgb", "2.0.0", b"abc" * 2000)
c = station.conn()
c.request("GET", "/ninlight_e27_rgb_ota.img", headers={"Range": "bytes=1000-"})
r = c.getresponse()
@@ -227,7 +235,7 @@ def test_range_request_returns_206_and_the_right_slice(station):
def test_state_file_records_a_delivery(station):
make_image(station.image_dir / "ninlight_evk_ota.img", "evk", "0.3.4", b"z" * 1000)
make_image(station.rescue_dir / "ninlight_evk_ota.img", "evk", "0.3.4", b"z" * 1000)
c = station.conn()
c.request("GET", "/ninlight_evk_ota.img")
c.getresponse().read()
@@ -241,6 +249,21 @@ def test_state_file_records_a_delivery(station):
assert entry["bytes"] == 1000 + TRAILER_SIZE
def test_state_file_carries_a_last_request_timestamp_for_the_display(station):
"""display.py zeigt einen kleinen Lebenszeichen-Punkt, sobald eine Anfrage jeder Art
hereinkommt (auch ein 404) - dafuer muss status.json den Zeitpunkt tragen, nicht nur
den Erfolgszaehler. Vor der ersten Anfrage ist das Feld da, aber None."""
initial = station.read_state()
assert "last_request_ts" in initial
assert initial["last_request_ts"] is None
c = station.conn()
c.request("GET", "/ninlight_g95_cw_ota.img") # existiert nicht - zaehlt trotzdem
c.getresponse().read()
state = station.read_state(lambda s: s["last_request_ts"] is not None)
assert isinstance(state["last_request_ts"], str) and state["last_request_ts"]
def test_state_file_records_a_missing_image(station):
c = station.conn()
c.request("GET", "/ninlight_g95_cw_ota.img")
@@ -266,7 +289,7 @@ def test_state_file_is_world_readable_even_under_a_restrictive_umask(tmp_path):
finally:
os.umask(old_umask)
try:
make_image(s.image_dir / "ninlight_evk_ota.img", "evk", "1.0.0", b"w" * 100)
make_image(s.rescue_dir / "ninlight_evk_ota.img", "evk", "1.0.0", b"w" * 100)
def world_readable() -> bool:
mode = stat.S_IMODE(s.state.stat().st_mode)
@@ -289,7 +312,7 @@ def test_state_file_is_world_readable_even_under_a_restrictive_umask(tmp_path):
def test_recent_is_capped_at_eight(station):
make_image(station.image_dir / "ninlight_evk_ota.img", "evk", "1.0.0", b"q" * 10)
make_image(station.rescue_dir / "ninlight_evk_ota.img", "evk", "1.0.0", b"q" * 10)
for _ in range(12):
c = station.conn()
c.request("GET", "/ninlight_evk_ota.img")
@@ -302,7 +325,7 @@ def test_recent_is_capped_at_eight(station):
def test_aborted_transfer_is_recorded(tmp_path):
s = Station(tmp_path, "--stall-after", "256")
try:
make_image(s.image_dir / "ninlight_evk_ota.img", "evk", "1.0.0", b"y" * 100000)
make_image(s.rescue_dir / "ninlight_evk_ota.img", "evk", "1.0.0", b"y" * 100000)
sock = socket.create_connection(("127.0.0.1", s.port), timeout=5)
sock.sendall(b"GET /ninlight_evk_ota.img HTTP/1.1\r\nHost: x\r\n\r\n")
sock.recv(4096)
@@ -326,7 +349,7 @@ def test_aborted_transfer_is_recorded(tmp_path):
def test_stall_after_stops_sending_but_keeps_the_connection(tmp_path):
s = Station(tmp_path, "--stall-after", "512")
try:
make_image(s.image_dir / "ninlight_evk_ota.img", "evk", "1.0.0", b"y" * 200000)
make_image(s.rescue_dir / "ninlight_evk_ota.img", "evk", "1.0.0", b"y" * 200000)
sock = socket.create_connection(("127.0.0.1", s.port), timeout=5)
sock.sendall(b"GET /ninlight_evk_ota.img HTTP/1.1\r\nHost: x\r\n\r\n")
got = b""
@@ -369,7 +392,7 @@ def test_stall_after_budget_survives_a_resumed_request(tmp_path):
"""
s = Station(tmp_path, "--stall-after", "1000")
try:
make_image(s.image_dir / "ninlight_evk_ota.img", "evk", "1.0.0", b"y" * 200000)
make_image(s.rescue_dir / "ninlight_evk_ota.img", "evk", "1.0.0", b"y" * 200000)
# First request: burns through the whole --stall-after budget, then the server
# holds the connection open and sends nothing further.
@@ -419,7 +442,7 @@ def test_concurrent_downloads_are_all_recorded(station):
concurrent requests can lose increments (two threads read the same old value) or have
one thread's status.json write clobber another's.
"""
make_image(station.image_dir / "ninlight_evk_ota.img", "evk", "1.0.0", b"c" * 5000)
make_image(station.rescue_dir / "ninlight_evk_ota.img", "evk", "1.0.0", b"c" * 5000)
expected_len = 5000 + TRAILER_SIZE
n_workers = 20
errors = []
@@ -1092,6 +1115,75 @@ def test_the_version_comes_from_the_manifest_of_the_selected_set(migration_serve
)["recent"][-1]["version"] == "0.4.0-wlan"
# --- Derselbe Umschalter gilt jetzt auch fuer die RETTUNG ---------------------------
#
# Der Eigentuemer hat das ausdruecklich verlangt: bisher bekam eine Rettung IMMER ein
# Abbild MIT Zugangsdaten (die flachen Abzuege in images/ kamen aus build/). image_path()
# folgt jetzt demselben Umschalter wie migrate_dir() - dieselben Tests wie oben fuer die
# Migration, nur gegen den Rettungspfad "/ninlight_<target>_ota.img".
def _both_rescue_sets(imgdir):
(imgdir / "rescue-nocreds").mkdir(exist_ok=True)
make_image(imgdir / "rescue-nocreds" / "ninlight_g95_cw_ota.img", "g95_cw", "1.0.0",
b"OHNE" * 100)
(imgdir / "rescue-wlan").mkdir(exist_ok=True)
make_image(imgdir / "rescue-wlan" / "ninlight_g95_cw_ota.img", "g95_cw", "1.0.0",
b"MITW" * 100)
def test_rescue_default_image_set_is_the_one_without_wifi_credentials(server):
"""Wie bei der Migration: ohne Umschalterdatei gilt die sichere Vorgabe - ein
Fehlgriff darf niemandem das eigene WLAN-Passwort in ein fremdes Geraet schreiben.
Vor diesem Umschalter bekam eine Rettung IMMER das Abbild MIT Zugangsdaten; das war
genau die Nebenwirkung, die der Eigentuemer sichtbar gemacht haben wollte."""
srv, imgdir = server
_both_rescue_sets(imgdir)
r = srv.get("/ninlight_g95_cw_ota.img")
assert r.status == 200
assert r.body.startswith(b"OHNE")
def test_rescue_switch_decides_which_file_is_actually_delivered(server):
srv, imgdir = server
_both_rescue_sets(imgdir)
srv.set_image_set("mit-wlan")
assert srv.get("/ninlight_g95_cw_ota.img").body.startswith(b"MITW")
# ... und wieder zurueck, ohne Dienstneustart.
srv.set_image_set("ohne-wlan")
assert srv.get("/ninlight_g95_cw_ota.img").body.startswith(b"OHNE")
def test_rescue_missing_selected_set_delivers_nothing(server):
"""Der gewaehlte Rettungs-Bildsatz liegt nicht auf dem Pi: NICHTS geht raus -
insbesondere nicht das gleichnamige Abbild des anderen, danebenliegenden Satzes.
Auf den anderen Satz auszuweichen waere genau der Fehlgriff, gegen den der
Umschalter da ist (siehe rescue_dir() in server.py)."""
srv, imgdir = server
(imgdir / "rescue-nocreds").mkdir(exist_ok=True)
make_image(imgdir / "rescue-nocreds" / "ninlight_g95_cw_ota.img", "g95_cw", "1.0.0",
b"OHNE" * 100)
srv.set_image_set("mit-wlan") # rescue-wlan/ existiert nicht
r = srv.get("/ninlight_g95_cw_ota.img")
assert r.status == 404
assert b"OHNE" not in r.body
st = srv.status(lambda s: len(s["recent"]) >= 1)
assert st["served"] == 0
assert st["recent"][-1]["status"] == "notfound"
def test_rescue_old_flat_layout_is_no_longer_served(server):
"""Rueckwaertskompatibilitaet bewusst NICHT gegeben (server.py, rescue_dir()): ein Pi
mit dem alten, flachen Bestand (vor diesem Umschalter, immer MIT Zugangsdaten aus
build/) muss einmal per deploy.sh neu bestueckt werden. Wuerde image_path() den
flachen Altbestand noch lesen, wuerde die Vorgabe "ohne-wlan" NICHTS daran aendern -
genau die alte, jetzt unerwuenschte Grundhaltung stillschweigend fortgesetzt."""
srv, imgdir = server
make_image(imgdir / "ninlight_g95_cw_ota.img", "g95_cw", "1.0.0", b"ALTBESTAND" * 50)
r = srv.get("/ninlight_g95_cw_ota.img")
assert r.status == 404
def test_read_image_set_defaults_without_a_file(tmp_path):
module = _load_server_module()
assert module.read_image_set(None) == "ohne-wlan"