fix(station): Signiervariante aus Produkt-ID und Version bestimmen

Eine Entwicklungs-G95 meldete sich am 2026-09-15 mit einer Produkt-ID OHNE
"-dev", aber einer Version MIT "-dev". products.py leitete die Variante
ausschliesslich aus der Produkt-ID ab, die Lampe galt als Produktionsgeraet,
bekam das produktionssigniertes Image und lehnte es ab - ihr Bootloader prueft
gegen den Entwicklungsschluessel.

Die Variante ist damit keine Eigenschaft der Produkt-ID, sondern der Anfrage.
PRODUCTS bildet nur noch auf Geraetetyp und Anzeigename ab; beide Schreibweisen
der Produkt-ID fuehren auf denselben Eintrag. classify_variant() in
parse_request() wertet Produkt-ID UND Version aus: "-dev" an einer der beiden
Stellen bedeutet Entwicklungsgeraet, in der Version nur als ganzes Wort
("-devel"/"-dev2" zaehlen nicht).

Damit eine Fehlklassifikation nicht wieder erst an der Lampe auffaellt, fuehren
status.json (last_seen und jeder recent-Eintrag) und das Panel die Quelle der
Einstufung mit: "g95_cw (dev/Ver)", "dev/ID", "dev/ID+Ver" oder "prod".

Tests: die drei am Geraet mitgeschnittenen Anfragen plus die vierte denkbare
Kombination, die Wortgrenze in der Version, zwei Ende-zu-Ende-Tests im Server
(dev-Image statt prod-Image) und die Panel-Beschriftung.
This commit is contained in:
Michael Sauer
2026-09-15 00:24:29 +02:00
parent 6f0137908c
commit 6325ce073f
6 changed files with 216 additions and 33 deletions
+21 -1
View File
@@ -167,6 +167,25 @@ PHASE_LABELS = {"auto": "AUTO", "listen": "LISTEN"}
ARMED_LABEL = "FREIGEGEBEN" ARMED_LABEL = "FREIGEGEBEN"
ARMED_DETAIL_PREFIX = "Freigabe: " ARMED_DETAIL_PREFIX = "Freigabe: "
# Woraus die Station die Signiervariante abgeleitet hat (products.classify_variant):
# "-dev" kann in der Produkt-ID stehen, in der Version, oder in beiden. Die Quelle steht
# mit auf dem Panel, weil genau hier am 2026-09-15 ein falsch signiertes Image
# ausgeliefert wurde: wer vor der Freigabe sieht, dass "dev" nur aus der Version kommt
# (oder dass "prod" bedeutet: nirgends ein "-dev"), kann eine Fehlklassifikation abfangen,
# bevor die Lampe das Image ablehnt. Kurzformen, weil die Beiwerkzeile schmal ist.
VARIANT_SOURCE_LABELS = {"both": "ID+Ver", "product_id": "ID", "version": "Ver"}
def _variant_label(last_seen: dict) -> str:
""""dev/Ver", "dev/ID", "dev/ID+Ver", "prod" - die Quelle nur, wo sie etwas aussagt.
Bei "prod" gibt es keine ("none"), und eine aeltere Serverfassung ohne das Feld
liefert wie bisher nur die Variante."""
variant = str(last_seen.get("variant") or "?")
label = VARIANT_SOURCE_LABELS.get(last_seen.get("variant_source"))
if not label:
return variant
return variant + "/" + label
def _hhmm(ts: object) -> str: def _hhmm(ts: object) -> str:
""""2026-09-13T13:04:02Z" -> "13:04". Alles andere (fehlt, falsch geformt, aus einer """"2026-09-13T13:04:02Z" -> "13:04". Alles andere (fehlt, falsch geformt, aus einer
@@ -275,7 +294,8 @@ def blocks(state: dict) -> list[tuple[str, str]]:
target = last_seen.get("target") target = last_seen.get("target")
variant = last_seen.get("variant") variant = last_seen.get("variant")
if target or variant: if target or variant:
out.append((f"{target or '?'} ({variant or '?'})", ROLE_DETAIL)) out.append(("{} ({})".format(target or "?", _variant_label(last_seen)),
ROLE_DETAIL))
lamp_version = last_seen.get("lamp_version") lamp_version = last_seen.get("lamp_version")
if lamp_version: if lamp_version:
out.append((f"FW: {lamp_version}", ROLE_DETAIL)) out.append((f"FW: {lamp_version}", ROLE_DETAIL))
+59 -12
View File
@@ -1,17 +1,37 @@
"""Produkt-IDs der alten Klyqa-Firmware und ihre Zuordnung zu ninlight-Zielen. """Produkt-IDs der alten Klyqa-Firmware und ihre Zuordnung zu ninlight-Zielen.
Vollstaendig uebernommen aus qcl_product_info.c der alten Firmware Vollstaendig uebernommen aus qcl_product_info.c der alten Firmware
(~/workspace/qcx-w600-firmware). Der "-dev"-Anhang entscheidet zugleich ueber den (~/workspace/qcx-w600-firmware).
Signierschluessel: Lampen mit Dev-Produkt-ID akzeptieren nur den Entwicklungs-,
Produktionslampen nur den Produktionsschluessel. Die Variante (dev/prod) entscheidet ueber den Signierschluessel: Lampen mit
Entwicklungs-Bootloader akzeptieren nur den Entwicklungs-, Produktionslampen nur den
Produktionsschluessel. Sie ist aber KEINE Eigenschaft der Produkt-ID, sondern eine
Eigenschaft der ANFRAGE - das ist der Befund vom 2026-09-15 an echter Hardware:
/qcx_firmware_-klyqa.lighting.cw-ww.g95_V1.1.1.2357-dev_ota_light.img
Dieselbe Entwicklungs-G95, aber das "-dev" steht nur in der VERSION, nicht in der
Produkt-ID. Die Station hielt sie damit fuer eine Produktionslampe, lieferte das
produktionssignierte Image aus, und die Lampe lehnte es ab. Die Machbarkeitsprobe im
September hatte eine andere Entwicklungs-G95 mit "-dev" an BEIDEN Stellen gesehen, eine
Produktionslampe an keiner - alle drei Formen kommen also vor.
Regel seither: "-dev" an einer der beiden Stellen (Produkt-ID oder Version) bedeutet
Entwicklungsgeraet. PRODUCTS bildet nur noch auf Geraetetyp und Anzeigename ab, beide
Schreibweisen der Produkt-ID fuehren auf denselben Eintrag; die Variante ermittelt
parse_request() aus beiden Teilen und meldet in .variant_source mit, WORAUS sie stammt
("both", "product_id", "version", "none"), damit eine Fehlklassifikation vor der Freigabe
am Panel und in status.json sichtbar ist statt erst an der Lampe.
""" """
import re import re
from collections import namedtuple from collections import namedtuple
Product = namedtuple("Product", "product_id target variant name") Product = namedtuple("Product", "product_id target name")
ParsedRequest = namedtuple("ParsedRequest", "product_id lamp_version product") ParsedRequest = namedtuple(
"ParsedRequest", "product_id lamp_version product variant variant_source")
# (Produkt-ID ohne "@", Ziel, Anzeigename) - die Dev-Variante wird darunter erzeugt. # (Produkt-ID ohne "@", Ziel, Anzeigename) - die "-dev"-Schreibweise wird darunter
# als zweiter Schluessel auf denselben Eintrag erzeugt.
_BASE = [ _BASE = [
("klyqa.lighting.rgb-cw-ww.e27", "e27_rgb", "Klyqa E27 RGB CW/WW"), ("klyqa.lighting.rgb-cw-ww.e27", "e27_rgb", "Klyqa E27 RGB CW/WW"),
("klyqa.lighting.rgb-cw-ww.e14", "e14_rgb", "Klyqa E14 RGB CW/WW"), ("klyqa.lighting.rgb-cw-ww.e14", "e14_rgb", "Klyqa E14 RGB CW/WW"),
@@ -25,6 +45,9 @@ _BASE = [
_UNSAFE = re.compile(r"[^0-9A-Za-z+\-_.]") _UNSAFE = re.compile(r"[^0-9A-Za-z+\-_.]")
# Die Produkt-ID darf kein "_" enthalten - daran trennen wir Version und ID. # Die Produkt-ID darf kein "_" enthalten - daran trennen wir Version und ID.
_PATH = re.compile(r"^/qcx_firmware_([^/_]+)_([^/_]+)_ota_light\.img$") _PATH = re.compile(r"^/qcx_firmware_([^/_]+)_([^/_]+)_ota_light\.img$")
# "-dev" in der Version, aber nicht als Anfang eines laengeren Wortes ("-devel", "-dev2"):
# erlaubt sind danach nur Ende oder ein Trennzeichen wie "." (z.B. "V1.1-dev.3").
_DEV_IN_VERSION = re.compile(r"-dev(?![0-9A-Za-z])")
def websafe(product_id): def websafe(product_id):
@@ -36,30 +59,54 @@ def websafe(product_id):
def _build(): def _build():
t = {} t = {}
for pid, target, name in _BASE: for pid, target, name in _BASE:
for suffix, variant in (("", "prod"), ("-dev", "dev")): base_key = websafe("@" + pid)
key = websafe("@" + pid + suffix) product = Product(base_key, target, name)
t[key] = Product(key, target, variant, name) # Beide Schreibweisen zeigen auf DENSELBEN Eintrag - "-dev" an der Produkt-ID
# sagt seit dem Hardwarebefund nichts mehr ueber das Ziel, nur noch (mit) etwas
# ueber die Variante, und die ermittelt parse_request().
t[base_key] = product
t[websafe("@" + pid + "-dev")] = product
# Das EVK hat nur eine Produktions-ID. Im Dev-Build meldet es sich laut # Das EVK hat nur eine Produktions-ID. Im Dev-Build meldet es sich laut
# qcl_product_info.c als "@klyqa.lighting.rgb-cw-ww.e27-dev" - die EVK-Zeile ist dort # qcl_product_info.c als "@klyqa.lighting.rgb-cw-ww.e27-dev" - die EVK-Zeile ist dort
# auskommentiert und durch die E27 ersetzt. Ein dev-signiertes EVK ist damit von einer # auskommentiert und durch die E27 ersetzt. Ein dev-signiertes EVK ist damit von einer
# dev-signierten E27 RGB NICHT unterscheidbar und bekommt deren Image. Das EVK wird # dev-signierten E27 RGB NICHT unterscheidbar und bekommt deren Image. Das EVK wird
# ohnehin seriell geflasht; die Zeile steht nur der Vollstaendigkeit halber hier. # ohnehin seriell geflasht; die Zeile steht nur der Vollstaendigkeit halber hier.
# Meldet es eine "-dev"-Version, gilt auch hier die Regel: dann wird evk_dev.img
# gesucht (und mangels Datei sauber mit 404 abgelehnt), nicht das falsch signierte.
key = websafe("@klyqa.lighting.w600.evk1.2") key = websafe("@klyqa.lighting.w600.evk1.2")
t[key] = Product(key, "evk", "prod", "W600 EVK v1.2") t[key] = Product(key, "evk", "W600 EVK v1.2")
return t return t
PRODUCTS = _build() PRODUCTS = _build()
def classify_variant(product_id, lamp_version):
"""(variant, variant_source) aus Produkt-ID UND Version.
"-dev" an einer der beiden Stellen bedeutet Entwicklungsgeraet; variant_source sagt,
an welcher es stand: "both", "product_id", "version" oder - bei prod - "none"."""
in_id = product_id.endswith("-dev")
in_version = bool(_DEV_IN_VERSION.search(lamp_version))
if in_id and in_version:
return "dev", "both"
if in_id:
return "dev", "product_id"
if in_version:
return "dev", "version"
return "prod", "none"
def parse_request(path): def parse_request(path):
"""Zerlegt einen angefragten Pfad. Gibt None zurueck, wenn er nicht dem Muster der """Zerlegt einen angefragten Pfad. Gibt None zurueck, wenn er nicht dem Muster der
alten Firmware entspricht, sonst ParsedRequest - dessen .product ist None, wenn die alten Firmware entspricht, sonst ParsedRequest - dessen .product ist None, wenn die
Produkt-ID unbekannt ist. Unbekannt heisst 404, nicht raten.""" Produkt-ID unbekannt ist. Unbekannt heisst 404, nicht raten. .variant/.variant_source
stehen auch bei unbekannter Produkt-ID, sie haengen nicht an der Tabelle."""
m = _PATH.match(path) m = _PATH.match(path)
if not m: if not m:
return None return None
pid, version = m.group(1), m.group(2) pid, version = m.group(1), m.group(2)
if pid != websafe(pid) or "/" in pid or ".." in pid.split("."): if pid != websafe(pid) or "/" in pid or ".." in pid.split("."):
return None return None
return ParsedRequest(pid, version, PRODUCTS.get(pid)) variant, source = classify_variant(pid, version)
return ParsedRequest(pid, version, PRODUCTS.get(pid), variant, source)
+14 -7
View File
@@ -259,7 +259,8 @@ class Station:
def record_migration(self, ip: str, target: str | None, variant: str | None, def record_migration(self, ip: str, target: str | None, variant: str | None,
product_id: str, lamp_version: str, product_name: str | None, product_id: str, lamp_version: str, product_name: str | None,
version: str, nbytes: int, status: str) -> None: version: str, nbytes: int, status: str,
variant_source: str | None = None) -> None:
"""Protokolliert eine Migrationsanfrage. Anders als record() oben wird hier auch """Protokolliert eine Migrationsanfrage. Anders als record() oben wird hier auch
dann ein Eintrag angelegt (Status "listen"/"unknown"/"notfound"), wenn nichts dann ein Eintrag angelegt (Status "listen"/"unknown"/"notfound"), wenn nichts
ausgeliefert wurde - das ist der ganze Zweck der Horchphase. last_seen wird nur bei ausgeliefert wurde - das ist der ganze Zweck der Horchphase. last_seen wird nur bei
@@ -269,13 +270,15 @@ class Station:
if status == "ok": if status == "ok":
self.served += 1 self.served += 1
entry = {"ts": now_iso(), "ip": ip, "target": target, "variant": variant, entry = {"ts": now_iso(), "ip": ip, "target": target, "variant": variant,
"variant_source": variant_source,
"product_id": product_id, "lamp_version": lamp_version, "product_id": product_id, "lamp_version": lamp_version,
"version": version, "bytes": nbytes, "status": status} "version": version, "bytes": nbytes, "status": status}
self.recent.append(entry) self.recent.append(entry)
del self.recent[:-RECENT_MAX] del self.recent[:-RECENT_MAX]
if target is not None: if target is not None:
self.last_seen = {"product_id": product_id, "target": target, self.last_seen = {"product_id": product_id, "target": target,
"variant": variant, "product_name": product_name, "variant": variant, "variant_source": variant_source,
"product_name": product_name,
"lamp_version": lamp_version, "ip": ip, "ts": entry["ts"]} "lamp_version": lamp_version, "ip": ip, "ts": entry["ts"]}
self.write_state() self.write_state()
@@ -395,7 +398,7 @@ class Handler(BaseHTTPRequestHandler):
self.reject(404, "not found") self.reject(404, "not found")
self.log(404, 0, started) self.log(404, 0, started)
return return
pid, lamp_version, product = parsed pid, lamp_version, product, variant, variant_source = parsed
ip = self.client_address[0] ip = self.client_address[0]
if product is None: if product is None:
self.station.record_migration(ip, None, None, pid, lamp_version, None, "?", 0, self.station.record_migration(ip, None, None, pid, lamp_version, None, "?", 0,
@@ -404,7 +407,11 @@ class Handler(BaseHTTPRequestHandler):
self.log(404, 0, started) self.log(404, 0, started)
return return
target, variant, name = product.target, product.variant, product.name # variant/variant_source kommen aus parse_request(), nicht mehr aus der
# Produkttabelle: das "-dev" kann in der Produkt-ID stehen, in der Version, oder in
# beiden (Hardwarebefund 2026-09-15, siehe products.py). Das entscheidet, welches
# der beiden unterschiedlich signierten Images ausgeliefert wird.
target, name = product.target, product.name
version = self.station.image_version(target, variant) version = self.station.image_version(target, variant)
# claim_delivery() liest die Phase UND verbraucht eine armed-Freigabe atomar unter # claim_delivery() liest die Phase UND verbraucht eine armed-Freigabe atomar unter
# station.lock (Fixrunde 1, C1) - zwischen dem Lesen und dem Verbrauchen darf keine # station.lock (Fixrunde 1, C1) - zwischen dem Lesen und dem Verbrauchen darf keine
@@ -415,7 +422,7 @@ class Handler(BaseHTTPRequestHandler):
# last_seen wird trotzdem gesetzt - das ist der ganze Zweck der Horchphase: # last_seen wird trotzdem gesetzt - das ist der ganze Zweck der Horchphase:
# die Anzeige soll zeigen, welches Modell da ist, auch wenn nichts rausgeht. # die Anzeige soll zeigen, welches Modell da ist, auch wenn nichts rausgeht.
self.station.record_migration(ip, target, variant, pid, lamp_version, name, self.station.record_migration(ip, target, variant, pid, lamp_version, name,
version, 0, "listen") version, 0, "listen", variant_source)
self.reject(404, "listening only") self.reject(404, "listening only")
self.log(404, 0, started) self.log(404, 0, started)
return return
@@ -427,7 +434,7 @@ class Handler(BaseHTTPRequestHandler):
# dass es gar kein Image gibt - das darf nicht auf Kosten der Freigabe gehen. # dass es gar kein Image gibt - das darf nicht auf Kosten der Freigabe gehen.
self.station.restore_armed(target) self.station.restore_armed(target)
self.station.record_migration(ip, target, variant, pid, lamp_version, name, self.station.record_migration(ip, target, variant, pid, lamp_version, name,
version, 0, "notfound") version, 0, "notfound", variant_source)
self.reject(404, "no image for this device") self.reject(404, "no image for this device")
self.log(404, 0, started) self.log(404, 0, started)
return return
@@ -440,7 +447,7 @@ class Handler(BaseHTTPRequestHandler):
# (das ist der Punkt von C1) - bei einem Abbruch wird das hier zurueckgenommen. # (das ist der Punkt von C1) - bei einem Abbruch wird das hier zurueckgenommen.
self.station.restore_armed(target) self.station.restore_armed(target)
self.station.record_migration(ip, target, variant, pid, lamp_version, name, version, self.station.record_migration(ip, target, variant, pid, lamp_version, name, version,
sent, status) sent, status, variant_source)
def _serve_file(self, path: Path, started: float) -> tuple[int, str]: def _serve_file(self, path: Path, started: float) -> tuple[int, str]:
"""Sendet `path` mit Range-Unterstuetzung; liefert (gesendete Bytes, Status). """Sendet `path` mit Range-Unterstuetzung; liefert (gesendete Bytes, Status).
+68 -13
View File
@@ -7,43 +7,98 @@ import products # noqa: E402
def test_table_is_complete(): def test_table_is_complete():
"""Acht Geraete mal zwei Varianten, minus das EVK, das keine eigene Dev-ID hat.""" """Sieben Geraete in beiden Schreibweisen der Produkt-ID, plus das EVK, das nur eine
Produktions-ID hat."""
assert len(products.PRODUCTS) == 15 assert len(products.PRODUCTS) == 15
targets = {p.target for p in products.PRODUCTS.values()} targets = {p.target for p in products.PRODUCTS.values()}
assert targets == {"evk", "e27_rgb", "e14_rgb", "gu10_rgb", assert targets == {"evk", "e27_rgb", "e14_rgb", "gu10_rgb",
"e27_cw", "e14_cw", "gu10_cw", "g95_cw"} "e27_cw", "e14_cw", "gu10_cw", "g95_cw"}
def test_production_id_maps_to_prod_variant(): def test_table_carries_no_variant():
"""Die Variante ist eine Eigenschaft der Anfrage, nicht der Produkt-ID - sie darf gar
nicht mehr aus der Tabelle zu holen sein (Hardwarebefund 2026-09-15)."""
assert not hasattr(products.PRODUCTS["-klyqa.lighting.cw-ww.g95"], "variant")
assert "variant" not in products.Product._fields
def test_both_spellings_map_to_the_same_device():
with_dev = products.PRODUCTS["-klyqa.lighting.cw-ww.g95-dev"]
without = products.PRODUCTS["-klyqa.lighting.cw-ww.g95"]
assert with_dev.target == without.target == "g95_cw"
assert with_dev.name == without.name == "Klyqa G95 Vintage CW/WW"
def test_production_id_maps_to_device_and_name():
p = products.PRODUCTS["-klyqa.lighting.rgb-cw-ww.gu10"] p = products.PRODUCTS["-klyqa.lighting.rgb-cw-ww.gu10"]
assert (p.target, p.variant) == ("gu10_rgb", "prod") assert p.target == "gu10_rgb"
assert p.name == "Klyqa GU10 RGB CW/WW" assert p.name == "Klyqa GU10 RGB CW/WW"
def test_dev_suffix_selects_dev_variant(): def test_parses_real_request_from_g95_dev_in_both_places():
p = products.PRODUCTS["-klyqa.lighting.cw-ww.g95-dev"] """Am 2026-09-13 an einer echten Entwicklungs-G95 protokolliert: "-dev" in Produkt-ID
assert (p.target, p.variant) == ("g95_cw", "dev") UND Version."""
def test_parses_real_request_from_g95():
"""Am 2026-09-13 an einer echten G95 protokolliert."""
r = products.parse_request( r = products.parse_request(
"/qcx_firmware_-klyqa.lighting.cw-ww.g95-dev_V1.1.1.1800-dev_ota_light.img") "/qcx_firmware_-klyqa.lighting.cw-ww.g95-dev_V1.1.1.1800-dev_ota_light.img")
assert r is not None assert r is not None
assert r.product_id == "-klyqa.lighting.cw-ww.g95-dev" assert r.product_id == "-klyqa.lighting.cw-ww.g95-dev"
assert r.lamp_version == "V1.1.1.1800-dev" assert r.lamp_version == "V1.1.1.1800-dev"
assert r.product.target == "g95_cw" assert r.product.target == "g95_cw"
assert r.product.variant == "dev" assert (r.variant, r.variant_source) == ("dev", "both")
def test_parses_real_request_from_gu10(): def test_parses_real_request_from_g95_dev_only_in_version():
"""Am 2026-09-15 an einer echten Entwicklungs-G95 protokolliert - die Anfrage, an der
die Station bisher scheiterte: Produkt-ID OHNE "-dev", Version MIT. Sie muss trotzdem
als Entwicklungsgeraet gelten, sonst geht das produktionssignierte Image raus."""
r = products.parse_request(
"/qcx_firmware_-klyqa.lighting.cw-ww.g95_V1.1.1.2357-dev_ota_light.img")
assert r is not None
assert r.product_id == "-klyqa.lighting.cw-ww.g95"
assert r.lamp_version == "V1.1.1.2357-dev"
assert r.product.target == "g95_cw"
assert (r.variant, r.variant_source) == ("dev", "version")
def test_parses_real_request_from_gu10_without_any_dev():
"""Die erfolgreich migrierte Produktionslampe: nirgends ein "-dev"."""
r = products.parse_request( r = products.parse_request(
"/qcx_firmware_-klyqa.lighting.rgb-cw-ww.gu10_V1.1.0.1138_ota_light.img") "/qcx_firmware_-klyqa.lighting.rgb-cw-ww.gu10_V1.1.0.1138_ota_light.img")
assert r.product.target == "gu10_rgb" assert r.product.target == "gu10_rgb"
assert r.product.variant == "prod" assert (r.variant, r.variant_source) == ("prod", "none")
assert r.lamp_version == "V1.1.0.1138" assert r.lamp_version == "V1.1.0.1138"
def test_dev_only_in_product_id_is_dev():
"""Die vierte denkbare Kombination - so noch nicht am Geraet gesehen, aber die
Machbarkeitsprobe zeigt, dass die Produkt-ID das "-dev" tragen kann, und nichts
garantiert, dass die Version es dann auch tut."""
r = products.parse_request(
"/qcx_firmware_-klyqa.lighting.cw-ww.g95-dev_V1.1.1.1800_ota_light.img")
assert r.product.target == "g95_cw"
assert (r.variant, r.variant_source) == ("dev", "product_id")
def test_dev_must_be_a_whole_word_in_the_version():
""""-devel"/"-dev2" sind kein Entwicklungskennzeichen - sonst wuerde eine
Produktionslampe mit so einer Version das dev-signierte Image bekommen."""
for version in ("V1.1.1.2357-devel", "V1.1.1.2357-dev2"):
r = products.parse_request(
"/qcx_firmware_-klyqa.lighting.cw-ww.g95_" + version + "_ota_light.img")
assert (r.variant, r.variant_source) == ("prod", "none"), version
# ... ein "-dev" vor einem Trennzeichen dagegen schon.
r = products.parse_request(
"/qcx_firmware_-klyqa.lighting.cw-ww.g95_V1.1-dev.3_ota_light.img")
assert (r.variant, r.variant_source) == ("dev", "version")
def test_variant_is_classified_even_for_unknown_product_ids():
r = products.parse_request(
"/qcx_firmware_-acme.lightbulb.xxl_V9.9-dev_ota_light.img")
assert r.product is None
assert (r.variant, r.variant_source) == ("dev", "version")
def test_unknown_product_id_parses_but_has_no_product(): def test_unknown_product_id_parses_but_has_no_product():
r = products.parse_request( r = products.parse_request(
"/qcx_firmware_-acme.lightbulb.xxl_V9.9_ota_light.img") "/qcx_firmware_-acme.lightbulb.xxl_V9.9_ota_light.img")
+20
View File
@@ -33,6 +33,26 @@ def test_migration_shows_detected_lamp():
assert "13:04" in text assert "13:04" in text
def test_migration_shows_where_the_dev_classification_came_from():
"""Der Hardwarebefund vom 2026-09-15: die Station lieferte ein falsch signiertes Image
aus, ohne dass am Panel etwas darauf hindeutete. Steht die Quelle der Einstufung
daneben, sieht der Bediener VOR der Freigabe, worauf sie beruht."""
def variant_line(last_seen):
state = {"mode": "migration", "phase": "listen", "served": 0, "recent": [],
"ssid": "qconnex-firmware", "last_seen": last_seen}
return [l for l in render.lines(state) if l.startswith("g95_cw (")][0]
base = {"product_name": "Klyqa G95 Vintage CW/WW", "target": "g95_cw",
"lamp_version": "V1.1.1.2357-dev"}
assert variant_line(dict(base, variant="dev", variant_source="version")) == "g95_cw (dev/Ver)"
assert variant_line(dict(base, variant="dev", variant_source="product_id")) == "g95_cw (dev/ID)"
assert variant_line(dict(base, variant="dev", variant_source="both")) == "g95_cw (dev/ID+Ver)"
# prod hat keine Quelle - da gibt es nirgends ein "-dev", das man anzeigen koennte.
assert variant_line(dict(base, variant="prod", variant_source="none")) == "g95_cw (prod)"
# Eine aeltere Serverfassung ohne das Feld bleibt darstellbar.
assert variant_line(dict(base, variant="dev")) == "g95_cw (dev)"
def test_recovery_mode_without_traffic(): def test_recovery_mode_without_traffic():
lines = render.lines({"mode": "recovery", "ssid": "ninlight-recovery", lines = render.lines({"mode": "recovery", "ssid": "ninlight-recovery",
"served": 0, "recent": []}) "served": 0, "recent": []})
+34
View File
@@ -594,6 +594,40 @@ def test_migration_auto_serves_known_product(migration_server):
assert st["last_seen"]["variant"] == "dev" assert st["last_seen"]["variant"] == "dev"
def test_migration_dev_only_in_version_serves_the_dev_image(migration_server):
"""Der Hardwarebefund vom 2026-09-15: Produkt-ID OHNE "-dev", Version MIT. Frueher
ging hier das produktionssignierte Image raus (200, und die Lampe lehnte es ab) -
jetzt muss das dev-signierte gehen, und status.json muss die Quelle mitfuehren."""
srv, imgdir = migration_server
(imgdir / "migrate").mkdir(exist_ok=True)
(imgdir / "migrate" / "g95_cw_dev.img").write_bytes(b"DEV" * 1000)
(imgdir / "migrate" / "g95_cw_prod.img").write_bytes(b"PROD" * 1000)
r = srv.get("/qcx_firmware_-klyqa.lighting.cw-ww.g95_V1.1.1.2357-dev_ota_light.img")
assert r.status == 200
assert r.body == b"DEV" * 1000
st = srv.status(lambda s: s.get("last_seen") is not None)
assert st["last_seen"]["target"] == "g95_cw"
assert st["last_seen"]["variant"] == "dev"
assert st["last_seen"]["variant_source"] == "version"
assert st["recent"][-1]["variant"] == "dev"
assert st["recent"][-1]["variant_source"] == "version"
def test_migration_production_lamp_still_gets_the_prod_image(migration_server):
"""Die erfolgreich migrierte GU10: nirgends ein "-dev", also prod - und die Quelle
"none", damit am Panel/in status.json steht, worauf die Einstufung beruht."""
srv, imgdir = migration_server
(imgdir / "migrate").mkdir(exist_ok=True)
(imgdir / "migrate" / "gu10_rgb_dev.img").write_bytes(b"DEV" * 1000)
(imgdir / "migrate" / "gu10_rgb_prod.img").write_bytes(b"PROD" * 1000)
r = srv.get("/qcx_firmware_-klyqa.lighting.rgb-cw-ww.gu10_V1.1.0.1138_ota_light.img")
assert r.status == 200
assert r.body == b"PROD" * 1000
st = srv.status(lambda s: s.get("last_seen") is not None)
assert st["last_seen"]["variant"] == "prod"
assert st["last_seen"]["variant_source"] == "none"
def test_migration_listen_records_but_serves_nothing(migration_server): def test_migration_listen_records_but_serves_nothing(migration_server):
srv, imgdir = migration_server srv, imgdir = migration_server
(imgdir / "migrate").mkdir(exist_ok=True) (imgdir / "migrate").mkdir(exist_ok=True)