test(station): Freigabe bleibt bei fehlendem Image und Abbruch stehen

Von den drei restore_armed()-Pfaden in _do_get_migration() war nur der
juengste (fehlender Bildsatz) getestet. Das Abschlussreview hat die beiden
aelteren mit je einem "if False:" ausgehebelt - beide Male blieben alle 225
Tests gruen, obwohl eine Freigabe dann verfaellt, ohne dass ein Byte geflossen
ist.

Zwei Tests: Image fehlt im vorhandenen Bildsatz (Status notfound) und Transfer
abgebrochen (--stall-after, Verbindung geschlossen, Status aborted). Beide
pruefen phase UND armed_target. Beide Mutationen anschliessend gegengeprueft -
je genau der zugehoerige Test wird rot.

MigrationTestServer nimmt jetzt zusaetzliche Serverargumente entgegen, damit
--stall-after auch auf der Migrationsroute nutzbar ist.
This commit is contained in:
Michael Sauer
2026-09-15 03:48:25 +02:00
parent bbe2e842b6
commit fa93259426
+72 -2
View File
@@ -483,7 +483,7 @@ class MigrationTestServer:
kann. Jeder Aufrufer kennt sein eigenes Praedikat (das ohnehin schon als Assertion
dasteht), daher kein pauschales Warten hier."""
def __init__(self, tmp_path, mode):
def __init__(self, tmp_path, mode, *extra_args):
self.image_dir = tmp_path / "images"
self.image_dir.mkdir()
self.state_path = tmp_path / "status.json"
@@ -503,7 +503,7 @@ class MigrationTestServer:
"--bind", "127.0.0.1", "--port", str(self.port),
"--state", str(self.state_path), "--mode", mode,
"--control", str(self.control_path),
"--imageset-file", str(self.imageset_path)],
"--imageset-file", str(self.imageset_path), *extra_args],
stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True,
)
deadline = time.time() + 10
@@ -677,6 +677,76 @@ def test_armed_for_other_target_does_not_serve(migration_server):
assert srv.read_control()["phase"] == "armed" # Freigabe bleibt stehen
# --- Eine Freigabe wird NUR durch eine erfolgreiche Auslieferung verbraucht ----------
#
# Die Zusage (Spec 3.3): claim_delivery() nimmt die Freigabe schon unter der Sperre
# weg, BEVOR ueberhaupt gesendet wird (das ist der Punkt von C1 aus Fixrunde 1). Geht
# danach nichts raus, muss restore_armed() sie zurueckgeben - sonst steht der Bediener
# vor einer Lampe, die beim naechsten Boot nichts mehr bekommt, und muss raten, warum.
#
# Es gibt drei solche Stellen in _do_get_migration(): fehlender Bildsatz, fehlendes
# Einzelimage und abgebrochener Transfer. Nur die erste (die juengste) war getestet -
# das Abschlussreview hat die beiden anderen mit je einem "if False:" ausgehebelt, ohne
# dass ein einziger der 225 Tests rot wurde. Die beiden folgenden Tests schliessen das.
def test_a_missing_image_does_not_burn_the_release(migration_server):
"""Der Bildsatz ist da, nur DIESES Modell fehlt darin (Status "notfound"). Die
Freigabe darf daran nicht verfallen: sie war fuer genau dieses Ziel gedacht, und ein
fehlendes Image ist ein Deploy-Versaeumnis, kein Auslieferungsvorgang."""
srv, imgdir = migration_server
(imgdir / "migrate-nocreds").mkdir(exist_ok=True)
# Der Satz existiert (ein anderes Modell liegt darin), e14_cw fehlt darin.
(imgdir / "migrate-nocreds" / "g95_cw_dev.img").write_bytes(b"IMG")
srv.set_control(phase="armed", armed_target="e14_cw")
r = srv.get("/qcx_firmware_-klyqa.lighting.cw-ww.e14_V1_ota_light.img")
assert r.status == 404
st = srv.status(lambda s: len(s["recent"]) >= 1)
assert st["recent"][-1]["status"] == "notfound"
assert st["served"] == 0
control = srv.read_control()
assert control["phase"] == "armed", \
"die Freigabe wurde verbraucht, obwohl es gar kein Image auszuliefern gab"
assert control["armed_target"] == "e14_cw", "das freigegebene Ziel ging verloren"
def test_an_aborted_transfer_does_not_burn_the_release(tmp_path):
"""Bricht die Uebertragung ab (Lampe verliert Strom oder Funk mitten im Download),
muss die Freigabe stehen bleiben, damit die Lampe es beim naechsten Boot erneut
versuchen kann - sonst muss jemand vor Ort erneut freigeben, und genau diese
Nachsicht ist der Grund, warum restore_armed() ueberhaupt existiert.
Nachgestellt wie test_aborted_transfer_is_recorded oben: --stall-after laesst den
Server nach n Byte verstummen, der Test schliesst die Verbindung danach, und
_serve_file() meldet "aborted"."""
srv = MigrationTestServer(tmp_path, "migration", "--stall-after", "256")
try:
imgdir = srv.image_dir
(imgdir / "migrate-nocreds").mkdir(exist_ok=True)
(imgdir / "migrate-nocreds" / "g95_cw_dev.img").write_bytes(b"y" * 100000)
srv.set_control(phase="armed", armed_target="g95_cw")
sock = socket.create_connection(("127.0.0.1", srv.port), timeout=5)
try:
sock.sendall(("GET " + G95_DEV + " HTTP/1.1\r\nHost: x\r\n\r\n").encode())
sock.recv(4096)
time.sleep(0.3)
finally:
sock.close()
st = srv.status(lambda s: s["recent"] and s["recent"][-1]["status"] == "aborted")
assert st["recent"][-1]["bytes"] < 100000
assert st["served"] == 0
control = srv.read_control(lambda c: True)
assert control["phase"] == "armed", \
"die Freigabe wurde von einem ABGEBROCHENEN Transfer verbraucht"
assert control["armed_target"] == "g95_cw", "das freigegebene Ziel ging verloren"
finally:
srv.stop()
def test_unknown_product_id_is_404_and_logged(migration_server):
srv, _ = migration_server
r = srv.get("/qcx_firmware_-acme.bulb_V1_ota_light.img")