admin
41f5eb1821
Am echten Broker aufgefallen: "hell_prozent = 100 uebernommen" stand im Log, die Anzeige blieb aber auf 50 %. Der Wert landete korrekt in settings.json und der Dimmer las ihn per reload() auch nach -- nur ging kein Helligkeitstelegramm raus. Grund: Ein Telegramm wird nur beim *Wechsel* der Stufe gesendet (so war es schon im alten MSP430-Programm gedacht, und fuer die LDR-Regelung ist das richtig). Aendert man die Helligkeit von aussen, waehrend die Stufe dieselbe bleibt, gibt es keinen Wechsel -- die Aenderung waere erst beim naechsten zufaelligen Stufenwechsel sichtbar geworden. run_esp32.main setzt nach dem reload() jetzt die Helligkeit der aktuellen Stufe neu. Ohne Dimmer (kein LDR) gilt weiterhin direkt hell_prozent. test_mqtt.TestHelligkeitWirdSofortSichtbar haelt das fest. Gegenprobe gemacht: mit der alten Fassung schlaegt der Test fehl (keine Helligkeit gesendet), mit der neuen geht genau ein Telegramm mit dem gesetzten Wert raus. Am Geraet bestaetigt, Broker "nuccy": Verbindung steht, Status (ra/dec/ldr/ helligkeit/link/online) kommt retained an, Einstellungen werden uebernommen, ungueltige Werte (500, Text) abgelehnt ohne settings.json anzufassen, und die Anzeige reagiert jetzt sichtbar auf 100 % -> 15 % -> 50 %. 145 Tests gruen. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
406 lines
14 KiB
Python
406 lines
14 KiB
Python
#!/usr/bin/env python3
|
|
"""Tests fuer die MQTT-Anbindung.
|
|
|
|
python3 -m unittest -v test_mqtt
|
|
|
|
Statt eines echten Brokers wird ein FakeClient eingehaengt, der dieselbe
|
|
Schnittstelle wie umqtt.simple.MQTTClient bietet. So laesst sich alles pruefen,
|
|
was im Betrieb schiefgehen kann -- auch die Faelle, die man mit einem echten
|
|
Broker nur schwer herbeifuehrt: Verbindungsabbruch beim Senden, Muell im Topic,
|
|
ein Broker, der gar nicht antwortet.
|
|
|
|
Schwerpunkt ist der Grundsatz aus mqtt.py: **MQTT darf die Anzeige nie
|
|
aufhalten.** Kein Test hier darf eine Ausnahme nach aussen durchlassen.
|
|
"""
|
|
|
|
import os
|
|
import unittest
|
|
|
|
import mqtt
|
|
import settings
|
|
|
|
|
|
class FakeClient:
|
|
"""Ersatz fuer umqtt.simple.MQTTClient."""
|
|
|
|
def __init__(self, **kw):
|
|
self.published = [] # (topic, payload, retain)
|
|
self.subscribed = []
|
|
self.last_will = None
|
|
self.callback = None
|
|
self.connected = False
|
|
self.disconnected = 0
|
|
self.eingang = [] # (topic, payload) -- werden bei check_msg zugestellt
|
|
self.fail_connect = False
|
|
self.fail_publish = False
|
|
self.fail_check = False
|
|
|
|
def set_callback(self, f):
|
|
self.callback = f
|
|
|
|
def set_last_will(self, topic, msg, retain=False):
|
|
self.last_will = (topic, msg, retain)
|
|
|
|
def connect(self):
|
|
if self.fail_connect:
|
|
raise OSError("Broker nicht erreichbar")
|
|
self.connected = True
|
|
|
|
def subscribe(self, topic):
|
|
self.subscribed.append(topic)
|
|
|
|
def publish(self, topic, msg, retain=False):
|
|
if self.fail_publish:
|
|
raise OSError("Verbindung weg")
|
|
self.published.append((topic, msg, retain))
|
|
|
|
def check_msg(self):
|
|
if self.fail_check:
|
|
raise OSError("Verbindung weg")
|
|
while self.eingang:
|
|
topic, payload = self.eingang.pop(0)
|
|
if self.callback:
|
|
self.callback(topic, payload)
|
|
|
|
def disconnect(self):
|
|
self.disconnected += 1
|
|
self.connected = False
|
|
|
|
|
|
class MqttTest(unittest.TestCase):
|
|
"""Basis: eine Bruecke mit FakeClient und eigener settings-Datei."""
|
|
|
|
def setUp(self):
|
|
self.pfad = "test_mqtt_settings_tmp.json"
|
|
self._weg()
|
|
# settings.PFAD umbiegen, damit die Tests nicht die echte Datei anfassen.
|
|
self._alt = settings.PFAD
|
|
settings.PFAD = self.pfad
|
|
settings.save(settings.DEFAULTS, self.pfad)
|
|
|
|
self.meldungen = []
|
|
self.geaendert = []
|
|
self.client = FakeClient()
|
|
self.bridge = mqtt.MqttBridge(
|
|
broker="test", prefix="ga", client=self.client,
|
|
log=self.meldungen.append,
|
|
on_change=self.geaendert.append)
|
|
|
|
def tearDown(self):
|
|
settings.PFAD = self._alt
|
|
self._weg()
|
|
|
|
def _weg(self):
|
|
try:
|
|
os.remove(self.pfad)
|
|
except OSError:
|
|
pass
|
|
|
|
def sende(self, topic, payload):
|
|
"""Eine Nachricht vom Broker zustellen lassen."""
|
|
self.client.eingang.append((topic.encode(), payload.encode()))
|
|
self.bridge.check()
|
|
|
|
def topics(self):
|
|
return [t.decode() if isinstance(t, bytes) else t
|
|
for t, _, _ in self.client.published]
|
|
|
|
|
|
class TestVerbinden(MqttTest):
|
|
|
|
def test_connect_abonniert_und_meldet_online(self):
|
|
self.assertTrue(self.bridge.connect())
|
|
self.assertTrue(self.bridge.connected)
|
|
self.assertEqual(self.client.subscribed, ["ga/set/+"])
|
|
self.assertIn(("ga/status/online", b"1", True), self.client.published)
|
|
|
|
def test_last_will_meldet_offline(self):
|
|
"""Bricht die Verbindung weg, muss der Broker selbst offline melden."""
|
|
self.bridge.connect()
|
|
self.assertEqual(self.client.last_will, ("ga/status/online", b"0", True))
|
|
|
|
def test_fehlgeschlagene_verbindung_wirft_nicht(self):
|
|
self.client.fail_connect = True
|
|
self.assertFalse(self.bridge.connect())
|
|
self.assertFalse(self.bridge.connected)
|
|
self.assertTrue(any("fehlgeschlagen" in m for m in self.meldungen))
|
|
|
|
def test_close_meldet_offline_ab(self):
|
|
self.bridge.connect()
|
|
self.bridge.close()
|
|
self.assertIn(("ga/status/online", b"0", True), self.client.published)
|
|
self.assertFalse(self.bridge.connected)
|
|
|
|
|
|
class TestEmpfang(MqttTest):
|
|
|
|
def setUp(self):
|
|
super().setUp()
|
|
self.bridge.connect()
|
|
|
|
def test_gueltiger_wert_wird_uebernommen(self):
|
|
self.sende("ga/set/hell_prozent", "60")
|
|
self.assertEqual(settings.load(self.pfad)["hell_prozent"], 60)
|
|
self.assertEqual(self.geaendert[-1]["hell_prozent"], 60)
|
|
|
|
def test_alle_setzbaren_werte(self):
|
|
for name in mqtt.SETZBAR:
|
|
self.assertIn(name, settings.DEFAULTS)
|
|
self.sende("ga/set/schwelle", "2000")
|
|
self.sende("ga/set/dunkel_prozent", "3")
|
|
self.sende("ga/set/hysterese", "100")
|
|
w = settings.load(self.pfad)
|
|
self.assertEqual((w["schwelle"], w["dunkel_prozent"], w["hysterese"]),
|
|
(2000, 3, 100))
|
|
|
|
def test_wert_ausserhalb_des_bereichs_wird_abgelehnt(self):
|
|
"""Der Kern der Pruefung: von aussen darf nichts Unsinniges durch."""
|
|
self.sende("ga/set/hell_prozent", "500")
|
|
self.assertEqual(settings.load(self.pfad)["hell_prozent"],
|
|
settings.DEFAULTS["hell_prozent"])
|
|
self.assertEqual(self.geaendert, [])
|
|
self.assertTrue(any("500" in m for m in self.meldungen))
|
|
|
|
def test_keine_zahl_wird_abgelehnt(self):
|
|
self.sende("ga/set/hell_prozent", "hell bitte")
|
|
self.assertEqual(settings.load(self.pfad)["hell_prozent"],
|
|
settings.DEFAULTS["hell_prozent"])
|
|
self.assertTrue(any("ganze Zahl" in m for m in self.meldungen))
|
|
|
|
def test_unbekanntes_topic_wird_verworfen(self):
|
|
self.sende("ga/set/gibtsnicht", "1")
|
|
self.assertEqual(self.geaendert, [])
|
|
self.assertTrue(any("unbekannt" in m for m in self.meldungen))
|
|
|
|
def test_leere_nachricht_wirft_nicht(self):
|
|
self.sende("ga/set/hell_prozent", "")
|
|
self.assertEqual(self.geaendert, [])
|
|
|
|
def test_kaputtes_on_change_bricht_nichts_ab(self):
|
|
"""Ein Fehler im Rueckruf darf den Empfang nicht toeten."""
|
|
def kaputt(_):
|
|
raise RuntimeError("absichtlich")
|
|
self.bridge._on_change = kaputt
|
|
self.sende("ga/set/hell_prozent", "60") # darf nicht werfen
|
|
self.assertEqual(settings.load(self.pfad)["hell_prozent"], 60)
|
|
self.assertTrue(any("on_change" in m for m in self.meldungen))
|
|
|
|
|
|
class TestSenden(MqttTest):
|
|
|
|
def setUp(self):
|
|
super().setUp()
|
|
self.bridge.connect()
|
|
self.client.published.clear()
|
|
|
|
def test_status_wird_veroeffentlicht(self):
|
|
self.bridge.publish("ra", "18h36m56s")
|
|
self.assertIn(("ga/status/ra", b"18h36m56s", True), self.client.published)
|
|
|
|
def test_gleicher_wert_wird_nicht_erneut_gesendet(self):
|
|
"""Spart Funkverkehr: LDR und Helligkeit aendern sich kaum."""
|
|
self.bridge.publish("ldr", 2453)
|
|
self.bridge.publish("ldr", 2453)
|
|
self.assertEqual(self.topics().count("ga/status/ldr"), 1)
|
|
|
|
def test_geaenderter_wert_wird_gesendet(self):
|
|
self.bridge.publish("ldr", 2453)
|
|
self.bridge.publish("ldr", 2600)
|
|
self.assertEqual(self.topics().count("ga/status/ldr"), 2)
|
|
|
|
def test_publish_many(self):
|
|
self.bridge.publish_many({"ra": "1", "dec": "2"})
|
|
self.assertIn("ga/status/ra", self.topics())
|
|
self.assertIn("ga/status/dec", self.topics())
|
|
|
|
def test_nach_reconnect_wird_alles_neu_gesendet(self):
|
|
"""Sonst fehlte dem Broker nach einem Abriss der aktuelle Stand."""
|
|
self.bridge.publish("ldr", 2453)
|
|
self.bridge._abwerfen()
|
|
self.bridge._client = self.client
|
|
self.bridge._naechster_versuch = None
|
|
self.bridge.connect()
|
|
self.client.published.clear()
|
|
self.bridge.publish("ldr", 2453)
|
|
self.assertIn("ga/status/ldr", self.topics())
|
|
|
|
|
|
class TestStoerungen(MqttTest):
|
|
"""Der eigentliche Zweck: die Anzeige darf nie stehenbleiben."""
|
|
|
|
def test_senden_bei_toter_verbindung_wirft_nicht(self):
|
|
self.bridge.connect()
|
|
self.client.fail_publish = True
|
|
self.assertFalse(self.bridge.publish("ra", "x"))
|
|
self.assertFalse(self.bridge.connected)
|
|
|
|
def test_empfang_bei_toter_verbindung_wirft_nicht(self):
|
|
self.bridge.connect()
|
|
self.client.fail_check = True
|
|
self.assertFalse(self.bridge.check())
|
|
self.assertFalse(self.bridge.connected)
|
|
|
|
def test_ohne_verbindung_wird_nicht_bei_jedem_aufruf_versucht(self):
|
|
"""Sonst kostet ein toter Broker in jedem Schleifendurchlauf Zeit."""
|
|
b = mqtt.MqttBridge(broker="test", prefix="ga", client=FakeClient(),
|
|
log=self.meldungen.append)
|
|
b._client.fail_connect = True
|
|
self.assertFalse(b.connect())
|
|
vorher = len(self.meldungen)
|
|
for _ in range(5):
|
|
self.assertFalse(b.check())
|
|
# Kein einziger neuer Versuch, solange die Wartezeit laeuft.
|
|
self.assertEqual(len(self.meldungen), vorher)
|
|
|
|
def test_wartezeit_waechst_und_ist_begrenzt(self):
|
|
b = mqtt.MqttBridge(broker="test", prefix="ga", client=FakeClient())
|
|
b._client.fail_connect = True
|
|
werte = []
|
|
for _ in range(12):
|
|
b._client = FakeClient()
|
|
b._client.fail_connect = True
|
|
b._naechster_versuch = None
|
|
b.connect()
|
|
werte.append(b._retry)
|
|
self.assertGreater(werte[1], werte[0]) # waechst
|
|
self.assertLessEqual(max(werte), mqtt.RETRY_MAX) # aber begrenzt
|
|
|
|
def test_publish_ohne_verbindung_meldet_nur_false(self):
|
|
b = mqtt.MqttBridge(broker="test", prefix="ga", client=FakeClient())
|
|
b._client.fail_connect = True
|
|
b.connect()
|
|
self.assertFalse(b.publish("ra", "x")) # wirft nicht
|
|
|
|
|
|
class TestHelligkeitWirdSofortSichtbar(MqttTest):
|
|
"""Am Geraet aufgefallen (2026-07-27): Einstellung uebernommen, Anzeige blieb.
|
|
|
|
Ein Helligkeitstelegramm geht nur beim *Wechsel* der Stufe raus. Wird per
|
|
MQTT hell_prozent geaendert, die Stufe bleibt aber dieselbe, muss die
|
|
Helligkeit trotzdem neu gesetzt werden -- sonst wirkt die Aenderung erst
|
|
beim naechsten zufaelligen Stufenwechsel.
|
|
"""
|
|
|
|
def _dimmer_und_display(self):
|
|
import brightness
|
|
from display import Display
|
|
from transport import FakeTransport
|
|
|
|
d = Display(FakeTransport(echo=False), response=True)
|
|
dimmer = brightness.BrightnessController(pfad=self.pfad)
|
|
dimmer.update(3000) # einschwingen auf HELL
|
|
return d, dimmer
|
|
|
|
def _uebernehmen_bauen(self, d, dimmer, log):
|
|
"""Dieselbe Logik wie in run_esp32.main."""
|
|
def uebernehmen(werte):
|
|
try:
|
|
if dimmer is not None:
|
|
dimmer.reload()
|
|
if dimmer.level is not None:
|
|
d.set_brightness(dimmer.percent(dimmer.level))
|
|
else:
|
|
d.set_brightness(werte["hell_prozent"])
|
|
except Exception as e:
|
|
log("Anzeige: %s" % e)
|
|
return uebernehmen
|
|
|
|
def _helligkeiten(self, transport):
|
|
"""Aus den Telegrammen die gesendeten Helligkeitswerte herausziehen."""
|
|
out = []
|
|
for frame in transport.frames:
|
|
i = frame.find(b"\x1bH")
|
|
if i >= 0:
|
|
out.append(int(frame[i + 3:i + 6]))
|
|
return out
|
|
|
|
def test_geaenderte_helligkeit_geht_an_die_anzeige(self):
|
|
d, dimmer = self._dimmer_und_display()
|
|
self.bridge._on_change = self._uebernehmen_bauen(d, dimmer,
|
|
self.meldungen.append)
|
|
self.bridge.connect()
|
|
d._t.frames.clear()
|
|
|
|
self.sende("ga/set/hell_prozent", "100")
|
|
|
|
self.assertEqual(self._helligkeiten(d._t), [100],
|
|
"Helligkeit wurde nicht an die Anzeige gesendet")
|
|
|
|
def test_ohne_dimmer_gilt_die_helle_stufe(self):
|
|
from display import Display
|
|
from transport import FakeTransport
|
|
d = Display(FakeTransport(echo=False), response=True)
|
|
self.bridge._on_change = self._uebernehmen_bauen(d, None,
|
|
self.meldungen.append)
|
|
self.bridge.connect()
|
|
d._t.frames.clear()
|
|
|
|
self.sende("ga/set/hell_prozent", "80")
|
|
|
|
self.assertEqual(self._helligkeiten(d._t), [80])
|
|
|
|
|
|
class TestOhneKonfiguration(unittest.TestCase):
|
|
|
|
def test_fehlende_mqtt_config_liefert_none(self):
|
|
"""Ohne mqtt_config.py laeuft die Anzeige ohne MQTT weiter."""
|
|
import sys
|
|
gemerkt = sys.modules.pop("mqtt_config", None)
|
|
pfad = sys.path[:]
|
|
try:
|
|
# Sicherstellen, dass mqtt_config wirklich nicht importierbar ist.
|
|
sys.path = [p for p in sys.path if p not in ("", ".", os.getcwd())]
|
|
meldungen = []
|
|
self.assertIsNone(mqtt.connect_from_config(log=meldungen.append))
|
|
self.assertTrue(any("mqtt_config" in m for m in meldungen))
|
|
finally:
|
|
sys.path = pfad
|
|
if gemerkt is not None:
|
|
sys.modules["mqtt_config"] = gemerkt
|
|
|
|
|
|
class TestSchleifenanbindung(MqttTest):
|
|
"""poll_loop mit Bruecke -- die Anzeige laeuft, auch wenn MQTT streikt."""
|
|
|
|
def _loop_einmal(self, bridge):
|
|
import displayloop
|
|
from display import Display
|
|
from transport import FakeTransport
|
|
|
|
class Client:
|
|
connected = True
|
|
|
|
def connect(self):
|
|
pass
|
|
|
|
def close(self):
|
|
pass
|
|
|
|
def get_coordinates(self):
|
|
return "18:36:56", "+38*47:01"
|
|
|
|
d = Display(FakeTransport(echo=False), response=True)
|
|
return displayloop.poll_loop(d, Client(), once=True, bridge=bridge,
|
|
log=lambda *a: None)
|
|
|
|
def test_koordinaten_gehen_raus(self):
|
|
self.bridge.connect()
|
|
self.client.published.clear()
|
|
self._loop_einmal(self.bridge)
|
|
self.assertIn("ga/status/ra", self.topics())
|
|
self.assertIn("ga/status/dec", self.topics())
|
|
self.assertIn("ga/status/link", self.topics())
|
|
|
|
def test_toter_broker_haelt_die_anzeige_nicht_auf(self):
|
|
"""Der wichtigste Test des Moduls."""
|
|
self.bridge.connect()
|
|
self.client.fail_publish = True
|
|
self.client.fail_check = True
|
|
gezeigt = self._loop_einmal(self.bridge)
|
|
self.assertIsNotNone(gezeigt) # Anzeige wurde trotzdem bedient
|
|
self.assertEqual(gezeigt[0], "18h36m56s")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main(verbosity=2)
|