4461035777
Zwischen Mock und echter GM4000 umschalten, ohne zu deployen und ohne Neustart:
mosquitto_pub -t grossanzeige/set/mount_host -m 192.168.1.115
mosquitto_pub -t grossanzeige/set/mount_host -m 192.168.178.26
mosquitto_pub -t grossanzeige/set/mount_host -m ''
Die Adresse liegt in settings.json auf dem Geraet und uebersteht damit
Stromausfall und Watchdog-Neustart. deploy.sh kopiert die Datei bewusst nicht
mit, die Einstellung ueberlebt also auch ein Deployen.
**Leer heisst: config.MOUNT_HOST gilt.** mount_client.resolve_host loest auf in
der Reihenfolge: ausdruecklich mitgegebener Host (REPL, run_display --host) ->
mount_host aus settings.json -> config.MOUNT_HOST. Ohne diesen Rueckweg wuerde
eine einmal per MQTT gesetzte Adresse die Zeile in config.py fuer immer
verdecken, auch nach einem Deployen mit geaendertem Code.
Pruefung: settings._pruefe unterscheidet jetzt Zahl und Text am Typ der Vorgabe
in DEFAULTS -- eine neue Einstellung wird weiterhin an genau einer Stelle
eingetragen. Fuer mount_host sind nur Buchstaben, Ziffern, Punkt und
Bindestrich erlaubt, hoechstens 63 Zeichen: Der Wert geht ungeprueft in
socket.getaddrinfo, und von MQTT kommt alles Moegliche. Ungueltiges wird
verworfen, settings.json bleibt unberuehrt.
Sofort wirksam: Der Rueckruf traegt die neue Adresse in den MountClient ein und
trennt nur die bestehende Verbindung -- neu verbindet die Poll-Schleife im
naechsten Durchlauf, mit der gewohnten Fehlerbehandlung. Antwortet dort niemand,
steht "KEIN LINK", und eine weitere Nachricht holt das Geraet zurueck.
grossanzeige/status/mount_host meldet, welche Adresse gerade abgefragt wird --
die Rueckmeldung steht direkt hinter bridge.check() und nicht im Erfolgspfad,
kommt also auch dann, wenn die Montierung schweigt.
Der on_change-Rueckruf ist als run_esp32.make_on_change aus main() herausgeloest.
test_mqtt.py prueft ihn bisher als Abschrift ("Dieselbe Logik wie in
run_esp32.main"); jetzt laeuft der echte Code durch die Tests.
201 Tests gruen. Neu, unter anderem:
- TestUmschaltenImLaufendenBetrieb faehrt die ganze Kette an echten Sockets:
erst "KEIN LINK" an einer toten Adresse, nach dem set/mount_host die
Koordinaten vom Mock.
- TestAdresswahl (test_lx200) haelt die Reihenfolge von resolve_host fest.
- test_esp32: main() ohne host nimmt nach einem Neustart, was gespeichert ist.
Nicht angefasst: run_display.py auf dem Mac behaelt config.MOUNT_HOST als
Vorgabe fuer --host. Dort gibt es keine MQTT-Bruecke, und settings.json ist
Zustand des Geraets, nicht des Macs.
Am Geraet bestaetigt (2026-07-29), am Broker mitgeschnitten. Umschalten im
laufenden Betrieb, mit dem Mock auf dem Mac als Ausgangslage:
0.0s status/mount_host 192.168.178.26 Ausgangslage, link = 1
-> set/mount_host = 192.168.178.99
5.1s status/mount_host 192.168.178.99 Rueckmeldung
5.9s status/link 0 dort antwortet niemand
-> set/mount_host = ''
17.8s status/mount_host 192.168.178.26 Vorgabe aus config.py
18.6s status/link 1 Koordinaten laufen wieder
Und ueber den Neustart hinweg -- der eigentliche Zweck des Speicherns:
-> set/mount_host = 192.168.178.99
1.6s status/mount_host 192.168.178.99
settings.json auf dem Geraet:
{"hysterese": 150, "mount_host": "192.168.178.99", "schwelle": 1800, ...}
-> Hardware-Reset
8.6s status/online 0 Last Will
15.5s status/online 1 Autostart durch
15.5s status/mount_host 192.168.178.99 aus der Datei, nicht aus config.py
Die alte settings.json ohne mount_host wird beim Laden einfach um die Vorgabe
ergaenzt; die eingestellten Helligkeiten bleiben unberuehrt. Eine Migration war
nicht noetig.
set/mount_host wird ohne Retain-Flag geschickt: Das ist ein Befehl, kein
Zustand. Retained laege er im Broker und wuerde bei jedem Reconnect erneut
zugestellt.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
172 lines
6.6 KiB
Python
172 lines
6.6 KiB
Python
#!/usr/bin/env python3
|
|
"""Tests fuer die zur Laufzeit aenderbaren Einstellungen (settings.json).
|
|
|
|
python3 -m unittest -v test_settings
|
|
|
|
Schwerpunkt ist die Pruefung der Werte: Sobald MQTT dazukommt, kommen sie von
|
|
aussen. Ein Helligkeitswert von 500 % oder eine kaputte Datei darf die Anzeige
|
|
nicht lahmlegen.
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import unittest
|
|
|
|
import settings
|
|
|
|
|
|
class Tempdatei(unittest.TestCase):
|
|
|
|
def setUp(self):
|
|
self.pfad = "test_settings_tmp.json"
|
|
self._weg()
|
|
|
|
def tearDown(self):
|
|
self._weg()
|
|
|
|
def _weg(self):
|
|
try:
|
|
os.remove(self.pfad)
|
|
except OSError:
|
|
pass
|
|
|
|
def _schreibe_roh(self, inhalt):
|
|
with open(self.pfad, "w") as f:
|
|
f.write(inhalt)
|
|
|
|
|
|
class TestLaden(Tempdatei):
|
|
|
|
def test_ohne_datei_gelten_die_vorgaben(self):
|
|
self.assertEqual(settings.load(self.pfad), settings.DEFAULTS)
|
|
|
|
def test_fehlende_werte_werden_ergaenzt(self):
|
|
self._schreibe_roh(json.dumps({"schwelle": 1234}))
|
|
w = settings.load(self.pfad)
|
|
self.assertEqual(w["schwelle"], 1234)
|
|
self.assertEqual(w["hell_prozent"], settings.DEFAULTS["hell_prozent"])
|
|
|
|
def test_kaputtes_json_faellt_auf_die_vorgaben_zurueck(self):
|
|
"""Ein Geraet in der Sternwarte darf daran nicht haengenbleiben."""
|
|
self._schreibe_roh("{das ist kein json")
|
|
self.assertEqual(settings.load(self.pfad), settings.DEFAULTS)
|
|
|
|
def test_json_das_kein_objekt_ist(self):
|
|
self._schreibe_roh("[1, 2, 3]")
|
|
self.assertEqual(settings.load(self.pfad), settings.DEFAULTS)
|
|
|
|
def test_einzelner_unbrauchbarer_wert_verwirft_nur_diesen(self):
|
|
self._schreibe_roh(json.dumps({"schwelle": 99999, "hell_prozent": 42}))
|
|
meldungen = []
|
|
w = settings.load(self.pfad, log=meldungen.append)
|
|
self.assertEqual(w["schwelle"], settings.DEFAULTS["schwelle"]) # verworfen
|
|
self.assertEqual(w["hell_prozent"], 42) # uebernommen
|
|
self.assertTrue(any("schwelle" in m for m in meldungen))
|
|
|
|
def test_kaputte_adresse_in_der_datei_verwirft_nur_diese(self):
|
|
self._schreibe_roh(json.dumps({"mount_host": "kein host!",
|
|
"hell_prozent": 42}))
|
|
meldungen = []
|
|
w = settings.load(self.pfad, log=meldungen.append)
|
|
self.assertEqual(w["mount_host"], settings.DEFAULTS["mount_host"])
|
|
self.assertEqual(w["hell_prozent"], 42)
|
|
|
|
def test_adresse_ueberlebt_das_speichern(self):
|
|
"""Der eigentliche Zweck: sie muss den Stromausfall ueberstehen."""
|
|
settings.update({"mount_host": "192.168.1.115"}, self.pfad)
|
|
self.assertEqual(settings.load(self.pfad)["mount_host"], "192.168.1.115")
|
|
|
|
def test_unbekannter_schluessel_wird_ignoriert(self):
|
|
self._schreibe_roh(json.dumps({"voellig_unbekannt": 1}))
|
|
w = settings.load(self.pfad)
|
|
self.assertEqual(w, settings.DEFAULTS)
|
|
self.assertNotIn("voellig_unbekannt", w)
|
|
|
|
|
|
class TestSpeichern(Tempdatei):
|
|
|
|
def test_speichern_und_wieder_lesen(self):
|
|
settings.save({"schwelle": 2000, "dunkel_prozent": 3}, self.pfad)
|
|
w = settings.load(self.pfad)
|
|
self.assertEqual(w["schwelle"], 2000)
|
|
self.assertEqual(w["dunkel_prozent"], 3)
|
|
|
|
def test_gespeichert_wird_immer_vollstaendig(self):
|
|
settings.save({"schwelle": 2000}, self.pfad)
|
|
with open(self.pfad) as f:
|
|
roh = json.load(f)
|
|
self.assertEqual(set(roh), set(settings.DEFAULTS))
|
|
|
|
def test_update_aendert_nur_das_angegebene(self):
|
|
settings.save({"schwelle": 1000, "hell_prozent": 30}, self.pfad)
|
|
settings.update({"hell_prozent": 70}, self.pfad)
|
|
w = settings.load(self.pfad)
|
|
self.assertEqual(w["hell_prozent"], 70)
|
|
self.assertEqual(w["schwelle"], 1000) # unveraendert
|
|
|
|
def test_ungueltiges_update_laesst_die_datei_unberuehrt(self):
|
|
settings.save({"schwelle": 1000}, self.pfad)
|
|
with self.assertRaises(settings.SettingsError):
|
|
settings.update({"hell_prozent": 500}, self.pfad)
|
|
self.assertEqual(settings.load(self.pfad)["schwelle"], 1000)
|
|
|
|
|
|
class TestPruefung(unittest.TestCase):
|
|
"""Die Grenzen, an denen spaeter MQTT-Eingaben haengenbleiben sollen."""
|
|
|
|
def test_prozent_ueber_100_wird_abgelehnt(self):
|
|
with self.assertRaises(settings.SettingsError):
|
|
settings.save({"hell_prozent": 101}, "/dev/null")
|
|
|
|
def test_negative_werte_werden_abgelehnt(self):
|
|
with self.assertRaises(settings.SettingsError):
|
|
settings.save({"schwelle": -1}, "/dev/null")
|
|
|
|
def test_schwelle_ueber_adc_bereich_wird_abgelehnt(self):
|
|
with self.assertRaises(settings.SettingsError):
|
|
settings.save({"schwelle": 4096}, "/dev/null")
|
|
|
|
def test_unbekannter_name_wird_abgelehnt(self):
|
|
with self.assertRaises(settings.SettingsError):
|
|
settings.save({"gibtsnicht": 1}, "/dev/null")
|
|
|
|
def test_nichtganzzahlige_werte_werden_abgelehnt(self):
|
|
for wert in (1.5, "50", None, True):
|
|
with self.assertRaises(settings.SettingsError):
|
|
settings.save({"hell_prozent": wert}, "/dev/null")
|
|
|
|
def test_adresse_wird_als_text_angenommen(self):
|
|
geprueft = settings._pruefe({"mount_host": "192.168.1.115"})
|
|
self.assertEqual(geprueft["mount_host"], "192.168.1.115")
|
|
|
|
def test_leere_adresse_ist_erlaubt(self):
|
|
"""Leer heisst 'Vorgabe aus config.py' -- der Weg zurueck zum Code."""
|
|
self.assertEqual(settings._pruefe({"mount_host": ""})["mount_host"], "")
|
|
|
|
def test_umgebende_leerzeichen_fallen_weg(self):
|
|
"""Bei MQTT-Nachrichten haengt schnell ein Zeilenende dran."""
|
|
self.assertEqual(settings._pruefe({"mount_host": " 10.0.0.5 "})["mount_host"],
|
|
"10.0.0.5")
|
|
|
|
def test_unerlaubte_zeichen_in_der_adresse(self):
|
|
"""Der Wert geht in socket.getaddrinfo -- nur Adressen und Namen."""
|
|
for wert in ("10.0.0.1 ; reboot", "hallo welt", "127.0.0.1:3490",
|
|
"192.168.1.5/24", "a" * 64):
|
|
with self.assertRaises(settings.SettingsError):
|
|
settings._pruefe({"mount_host": wert})
|
|
|
|
def test_adresse_muss_text_sein(self):
|
|
for wert in (42, None, True, ["10.0.0.1"]):
|
|
with self.assertRaises(settings.SettingsError):
|
|
settings._pruefe({"mount_host": wert})
|
|
|
|
def test_die_raender_sind_erlaubt(self):
|
|
geprueft = settings._pruefe({"hell_prozent": 100, "dunkel_prozent": 0,
|
|
"schwelle": 4095, "hysterese": 0})
|
|
self.assertEqual(geprueft["hell_prozent"], 100)
|
|
self.assertEqual(geprueft["schwelle"], 4095)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main(verbosity=2)
|