admin
bf39b75ac1
Topics (Praefix aus mqtt_config.py, Vorgabe "grossanzeige"):
<PREFIX>/set/hell_prozent 60 eingehend, schreibt settings.json
<PREFIX>/set/dunkel_prozent 5
<PREFIX>/set/schwelle 1800
<PREFIX>/set/hysterese 150
<PREFIX>/status/ra 18h36m56s ausgehend, retained
<PREFIX>/status/dec +38°47'01"
<PREFIX>/status/ldr 2453
<PREFIX>/status/helligkeit 50
<PREFIX>/status/link 1
<PREFIX>/status/online 1 mit Last Will auf 0
Eine empfangene Einstellung wird sofort wirksam: dieselbe Pruefung wie sonst
(settings.update), dann laedt der laufende BrightnessController sie per reload()
nach -- kein Neustart noetig.
Leitgedanke des Moduls: **MQTT darf die Anzeige nie aufhalten.** Die Anzeige ist
der Zweck des Geraets, MQTT ist Beiwerk. Deshalb faengt mqtt.py jeden Fehler
selbst ab und meldet ihn nur:
- Broker nicht erreichbar -> Versuch scheitert, Schleife laeuft weiter.
Wiederholung mit wachsendem Abstand (5..120 s), sonst kostet ein dauerhaft
toter Broker in jedem Schleifendurchlauf Zeit.
- Verbindungsabriss -> beim naechsten Senden/Empfangen bemerkt, Verbindung wird
verworfen und spaeter neu aufgebaut. Danach geht der gesamte Status erneut
raus, damit der Broker nicht auf veralteten Werten sitzenbleibt.
- Unsinniger Wert von aussen -> verworfen, settings.json bleibt unberuehrt.
- Geraet faellt aus -> Last Will meldet online=0. Ohne das bliebe online=1
stehen, obwohl niemand mehr da ist.
Der Empfang blockiert nicht (check_msg). Statuswerte gehen nur bei Aenderung
raus -- die Koordinaten aendern sich staendig, LDR und Helligkeit kaum.
mqtt_config.py ist optional und gitignored (Vorlage mqtt_config_example.py);
fehlt sie, laeuft alles wie bisher ohne MQTT. deploy.sh weist nur darauf hin.
test_mqtt.py haengt einen Fake-Broker ein und prueft auch die Faelle, die man
mit einem echten Server schwer herbeifuehrt: Abriss beim Senden, Muell im Topic,
Broker der nicht antwortet. 143 Tests gruen.
Am Geraet bestaetigt ist bisher der Fall OHNE Broker: mqtt.py importiert,
connect_from_config liefert None, ein unerreichbarer Broker wirft keine
Ausnahme. Der Test gegen einen echten Broker steht noch aus.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
338 lines
12 KiB
Python
338 lines
12 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 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)
|