OTA-Update per MQTT

Neues Kommando <PREFIX>/cmd/ota loest ota.update() aus: laedt die
Projektdateien per HTTP, erst als temporaere .ota-Dateien, und benennt
sie nur bei vollem Erfolg um. Schlaegt ein Download fehl, bleiben die
bisherigen Dateien unveraendert -- die Anzeige laeuft mit der alten
Version weiter (Grundsatz wie beim Rest von MQTT: eine fehlgeschlagene
OTA darf das Geraet nicht unbrauchbar machen).

Am Gerät bestätigt, inkl. echtem Reboot: leere/ungueltige Payload,
unbekanntes Kommando und nicht erreichbarer Server werden abgefangen,
die Poll-Schleife laeuft unbeeinflusst weiter. Ein echtes Update mit
allen Projektdateien kam sauber ueber main.py-Autostart wieder hoch.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_017zTSXLK4w32PoCmFgSzp68

test_ota.py

@@ -0,0 +1,139 @@
+#!/usr/bin/env python3
+"""Tests fuer das OTA-Update (ota.py).
+
+    python3 -m unittest -v test_ota
+
+Ein echter Download wird nicht gebraucht: update() nimmt einen Getter
+entgegen (Vorgabe ota._get), der sich hier durch eine Attrappe ersetzen
+laesst -- genau wie mqtt.MqttBridge einen Client injiziert bekommt.
+
+Schwerpunkt ist derselbe Grundsatz wie bei MQTT: **Eine fehlgeschlagene OTA
+darf das Geraet nicht unbrauchbar machen.** Ein abgebrochener Download darf
+die bisherigen Dateien nicht antasten.
+"""
+
+import os
+import unittest
+
+import ota
+
+
+class FakeGetter:
+    """Ersatz fuer ota._get: liefert vordefinierte Inhalte oder wirft."""
+
+    def __init__(self, inhalte):
+        self.inhalte = dict(inhalte)   # url -> bytes (oder Exception-Instanz)
+        self.angefragt = []
+
+    def __call__(self, url):
+        self.angefragt.append(url)
+        wert = self.inhalte.get(url)
+        if wert is None:
+            raise ota.OtaError("nicht gefunden: %s" % url)
+        if isinstance(wert, Exception):
+            raise wert
+        return wert
+
+
+class OtaTest(unittest.TestCase):
+    """Arbeitet in einem eigenen Verzeichnis, um echte Projektdateien nicht
+    anzufassen."""
+
+    def setUp(self):
+        self._cwd = os.getcwd()
+        self.tmp = "test_ota_tmp"
+        os.makedirs(self.tmp, exist_ok=True)
+        os.chdir(self.tmp)
+        self.meldungen = []
+
+    def tearDown(self):
+        os.chdir(self._cwd)
+        import shutil
+        shutil.rmtree(self.tmp, ignore_errors=True)
+
+    def _bestehende_datei(self, name, inhalt):
+        with open(name, "w") as f:
+            f.write(inhalt)
+
+
+class TestErfolg(OtaTest):
+
+    def test_alle_dateien_werden_geschrieben(self):
+        files = ("a.py", "b.py")
+        getter = FakeGetter({
+            "http://srv/a.py": b"INHALT_A",
+            "http://srv/b.py": b"INHALT_B",
+        })
+        ota.update("http://srv", files=files, log=self.meldungen.append,
+                  getter=getter)
+        with open("a.py", "rb") as f:
+            self.assertEqual(f.read(), b"INHALT_A")
+        with open("b.py", "rb") as f:
+            self.assertEqual(f.read(), b"INHALT_B")
+        # Keine .ota-Reste nach erfolgreicher Uebernahme.
+        self.assertFalse(os.path.exists("a.py.ota"))
+        self.assertFalse(os.path.exists("b.py.ota"))
+
+    def test_bestehende_datei_wird_ersetzt(self):
+        self._bestehende_datei("a.py", "ALT")
+        getter = FakeGetter({"http://srv/a.py": b"NEU"})
+        ota.update("http://srv", files=("a.py",), getter=getter)
+        with open("a.py") as f:
+            self.assertEqual(f.read(), "NEU")
+
+    def test_url_mit_abschliessendem_slash(self):
+        getter = FakeGetter({"http://srv/a.py": b"X"})
+        ota.update("http://srv/", files=("a.py",), getter=getter)
+        self.assertEqual(getter.angefragt, ["http://srv/a.py"])
+
+    def test_feed_wird_vor_jeder_datei_gerufen(self):
+        gefuettert = []
+        getter = FakeGetter({
+            "http://srv/a.py": b"A", "http://srv/b.py": b"B",
+        })
+        ota.update("http://srv", files=("a.py", "b.py"), getter=getter,
+                  feed=lambda: gefuettert.append(1))
+        self.assertEqual(len(gefuettert), 2)
+
+
+class TestFehlschlag(OtaTest):
+
+    def test_abgebrochener_download_laesst_bestehende_datei_unangetastet(self):
+        self._bestehende_datei("a.py", "ALT")
+        self._bestehende_datei("b.py", "ALT_B")
+        # a.py laedt erfolgreich, b.py schlaegt fehl -- danach darf a.py nicht
+        # trotzdem auf dem neuen Stand stehen, sonst liefe ein halb
+        # aktualisiertes Geraet weiter.
+        getter = FakeGetter({"http://srv/a.py": b"NEU"})
+        with self.assertRaises(ota.OtaError):
+            ota.update("http://srv", files=("a.py", "b.py"), getter=getter,
+                      log=self.meldungen.append)
+        with open("a.py") as f:
+            self.assertEqual(f.read(), "ALT")
+        with open("b.py") as f:
+            self.assertEqual(f.read(), "ALT_B")
+        self.assertFalse(os.path.exists("a.py.ota"))
+        self.assertTrue(any("fehlgeschlagen" in m for m in self.meldungen))
+
+    def test_leere_antwort_gilt_als_fehler(self):
+        getter = FakeGetter({"http://srv/a.py": b""})
+        with self.assertRaises(ota.OtaError):
+            ota.update("http://srv", files=("a.py",), getter=getter)
+        self.assertFalse(os.path.exists("a.py"))
+
+    def test_http_fehler_bricht_ab(self):
+        getter = FakeGetter({"http://srv/a.py": ota.OtaError("HTTP 404 fuer /a.py")})
+        with self.assertRaises(ota.OtaError):
+            ota.update("http://srv", files=("a.py",), getter=getter)
+
+
+class TestHttpGet(unittest.TestCase):
+    """_get: nur der Teil, der ohne echten Netzwerkzugriff prüfbar ist."""
+
+    def test_nur_http_wird_unterstuetzt(self):
+        with self.assertRaises(ota.OtaError):
+            ota._get("https://srv/a.py")
+
+
+if __name__ == "__main__":
+    unittest.main()