grossanzeige/test_mqtt.py

#!/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_mount_host_wird_als_text_uebernommen(self):
        """Nicht jede Einstellung ist eine Zahl -- die Adresse ist Text."""
        self.sende("ga/set/mount_host", "192.168.1.115")
        self.assertEqual(settings.load(self.pfad)["mount_host"], "192.168.1.115")
        self.assertEqual(self.geaendert[-1]["mount_host"], "192.168.1.115")

    def test_mount_host_darf_ein_name_sein(self):
        self.sende("ga/set/mount_host", "gm4000.sternwarte.local")
        self.assertEqual(settings.load(self.pfad)["mount_host"],
                         "gm4000.sternwarte.local")

    def test_leerer_mount_host_setzt_auf_die_vorgabe_zurueck(self):
        """Leer heisst: wieder config.MOUNT_HOST -- der Weg zurueck."""
        self.sende("ga/set/mount_host", "10.0.0.9")
        self.sende("ga/set/mount_host", "")
        self.assertEqual(settings.load(self.pfad)["mount_host"], "")

    def test_unsinnige_adresse_wird_abgelehnt(self):
        """Der Wert geht in socket.getaddrinfo -- da darf kein Muell ankommen."""
        self.sende("ga/set/mount_host", "10.0.0.1; rm -rf /")
        self.assertEqual(settings.load(self.pfad)["mount_host"],
                         settings.DEFAULTS["mount_host"])
        self.assertEqual(self.geaendert, [])
        self.assertTrue(any("unerlaubte" in m for m in self.meldungen))

    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 FakeMountClient:
    """Nur so viel MountClient, wie der Rueckruf anfasst."""

    def __init__(self, host="192.0.2.7"):
        self.host = host
        self.geschlossen = 0

    def close(self):
        self.geschlossen += 1


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, client=None):
        """Der echte Rueckruf aus run_esp32 -- keine Abschrift davon."""
        import run_esp32
        return run_esp32.make_on_change(d, client or FakeMountClient(),
                                        dimmer, log=log)

    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 TestAdresswechselWirktSofort(MqttTest):
    """Der Zweck der Uebung: zwischen Mock und Montierung umschalten.

    Der Rueckruf traegt die neue Adresse in den Client ein und trennt die
    bestehende Verbindung -- neu verbunden wird in der Poll-Schleife.
    """

    def setUp(self):
        super().setUp()
        import config
        import run_esp32
        from display import Display
        from transport import FakeTransport

        # Ausgangslage wie nach dem Start: der Client fragt die Adresse ab, die
        # resolve_host geliefert hat -- hier die Vorgabe aus config.py.
        self.mount = FakeMountClient(config.MOUNT_HOST)
        self.display = Display(FakeTransport(echo=False), response=True)
        self.bridge._on_change = run_esp32.make_on_change(
            self.display, self.mount, None, log=self.meldungen.append)
        self.bridge.connect()

    def test_neue_adresse_landet_im_client(self):
        self.sende("ga/set/mount_host", "192.168.1.115")
        self.assertEqual(self.mount.host, "192.168.1.115")

    def test_bestehende_verbindung_wird_getrennt(self):
        """Sonst liefe die Abfrage weiter zur alten Adresse."""
        self.sende("ga/set/mount_host", "192.168.1.115")
        self.assertEqual(self.mount.geschlossen, 1)

    def test_dieselbe_adresse_trennt_nicht(self):
        """Eine wiederholte Nachricht darf die Abfrage nicht stoeren."""
        self.sende("ga/set/mount_host", self.mount.host)
        self.assertEqual(self.mount.geschlossen, 0)

    def test_helligkeit_laesst_die_verbindung_in_ruhe(self):
        import config
        self.sende("ga/set/hell_prozent", "60")
        self.assertEqual(self.mount.geschlossen, 0)
        self.assertEqual(self.mount.host, config.MOUNT_HOST)

    def test_leere_adresse_holt_die_vorgabe_aus_config(self):
        import config
        self.sende("ga/set/mount_host", "192.168.1.115")
        self.sende("ga/set/mount_host", "")
        self.assertEqual(self.mount.host, config.MOUNT_HOST)
        self.assertEqual(self.mount.geschlossen, 2)   # hin und zurueck


class TestUmschaltenImLaufendenBetrieb(MqttTest):
    """Dieselbe Kette wie am Geraet, nur mit echten Sockets:
    Nachricht -> settings.json -> MountClient -> naechste Abfrage.

    Angefangen wird an einer Adresse, an der niemand antwortet -- die Anzeige
    steht auf "KEIN LINK". Nach dem set/mount_host holt die Poll-Schleife die
    Koordinaten vom Mock, ohne dass jemand etwas neu startet.
    """

    def setUp(self):
        super().setUp()
        import run_esp32
        from display import Display
        from mount_client import MountClient
        from mount_mock import MountMock
        from transport import FakeTransport

        self.mock = MountMock(host="127.0.0.1", port=0)
        self.mock.start()
        # 127.0.0.2 ist nicht belegt -- dort antwortet niemand. Kurzer Timeout,
        # damit der erste Durchlauf den Test nicht aufhaelt.
        self.mount = MountClient("127.0.0.2", self.mock.port, timeout=0.3)
        self.display = Display(FakeTransport(echo=False), response=True)
        self.bridge._on_change = run_esp32.make_on_change(
            self.display, self.mount, None, log=self.meldungen.append)
        self.bridge.connect()

    def tearDown(self):
        self.mount.close()
        self.mock.stop()
        super().tearDown()

    def _durchlauf(self):
        import displayloop
        return displayloop.poll_loop(self.display, self.mount, once=True,
                                     bridge=self.bridge, log=lambda *a: None)

    def test_von_kein_link_zur_montierung(self):
        import config
        self.assertEqual(self._durchlauf(), config.NOLINK_LINES)

        self.sende("ga/set/mount_host", "127.0.0.1")

        self.assertEqual(self._durchlauf()[0], "RA 18h36.9m")
        self.assertIn(("ga/status/mount_host", b"127.0.0.1", True),
                      self.client.published)


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
            host = "192.0.2.7"        # wird als status/mount_host gemeldet

            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_abgefragte_adresse_wird_gemeldet(self):
        """Sonst bliebe offen, ob ein set/mount_host angekommen ist."""
        self.bridge.connect()
        self.client.published.clear()
        self._loop_einmal(self.bridge)
        self.assertIn(("ga/status/mount_host", b"192.0.2.7", True),
                      self.client.published)

    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
        # Bei Niedrigpraezision (config-Vorgabe) die beschriftete Zehntelform.
        self.assertEqual(gezeigt[0], "RA 18h36.9m")


if __name__ == "__main__":
    unittest.main(verbosity=2)