grossanzeige/test_ota.py

#!/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()