audi-app/tests/aktualisierung/test_aktualisierung.py

#!/usr/bin/env python3
"""Regressionstest für aktualisierung.py's entpacken_pruefen_tauschen().

Testet die Sicherheitslogik des Selbst-Updates komplett isoliert: kein
Gitea-Zugriff, kein Home Assistant, nur ein selbstgebautes Zip gegen einen
Temp-Ordner. Deckt genau die Fälle ab, die vor jedem Tausch abgefangen
werden müssen (siehe Moduldocstring in aktualisierung.py):
  - erfolgreicher Tausch inklusive Backup der alten Fassung
  - falsche Domain im heruntergeladenen Manifest
  - fehlendes version-Feld
  - fehlende manifest.json überhaupt
  - kaputtes Zip
  - der gesuchte Pfad fehlt im Archiv

Aufruf: python3 tests/aktualisierung/test_aktualisierung.py
(braucht das homeassistant-Paket, weil aktualisierung.py es importiert -
in der Integration selbst immer vorhanden, hier z. B. im Test-Container)
"""

import io
import os
import shutil
import sys
import tempfile
import unittest
import zipfile
from unittest import mock

_HIER = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(
    0, os.path.join(os.path.dirname(os.path.dirname(_HIER)), "custom_components", "audi_dashboard")
)

import aktualisierung as a  # noqa: E402


def _zip_bauen(dateien: dict[str, bytes], wurzel: str = "audi-app-main") -> bytes:
    """Baut ein Zip mit den gegebenen Dateien unter
    <wurzel>/custom_components/audi_dashboard/<pfad> - genau die Form, in
    der Gitea ein Repo-Archiv liefert."""
    puffer = io.BytesIO()
    with zipfile.ZipFile(puffer, "w") as archiv:
        for pfad, inhalt in dateien.items():
            archiv.writestr(f"{wurzel}/custom_components/audi_dashboard/{pfad}", inhalt)
    return puffer.getvalue()


_GUELTIGES_MANIFEST = b'{"domain": "audi_dashboard", "version": "2099.1.1.1"}'


class ErfolgreicherTausch(unittest.TestCase):
    def setUp(self):
        self.arbeitsordner = tempfile.mkdtemp()
        self.integration = os.path.join(self.arbeitsordner, "audi_dashboard")
        self.staging = os.path.join(self.arbeitsordner, "audi_dashboard_update_staging")
        self.backup = os.path.join(self.arbeitsordner, "audi_dashboard_backup")
        os.makedirs(self.integration)
        with open(os.path.join(self.integration, "manifest.json"), "wb") as f:
            f.write(b'{"domain": "audi_dashboard", "version": "2026.1.1.1"}')
        with open(os.path.join(self.integration, "alte_datei.py"), "wb") as f:
            f.write(b"# alte Fassung\n")

    def tearDown(self):
        shutil.rmtree(self.arbeitsordner, ignore_errors=True)

    def test_tauscht_und_sichert_die_alte_fassung(self):
        zip_bytes = _zip_bauen({
            "manifest.json": _GUELTIGES_MANIFEST,
            "neue_datei.py": b"# neue Fassung\n",
            "frontend/app.js": b"// panel\n",
        })
        ergebnis = a.entpacken_pruefen_tauschen(
            zip_bytes, self.integration, self.staging, self.backup
        )
        self.assertEqual(ergebnis["version"], "2099.1.1.1")
        # Das Testarchiv enthaelt .py-Dateien, also neustartpflichtig.
        self.assertTrue(ergebnis["neustart_noetig"])

        # Live-Ordner trägt jetzt die neue Fassung.
        with open(os.path.join(self.integration, "manifest.json"), "rb") as f:
            self.assertIn(b'"2099.1.1.1"', f.read())
        self.assertTrue(os.path.exists(os.path.join(self.integration, "neue_datei.py")))
        self.assertTrue(os.path.exists(os.path.join(self.integration, "frontend", "app.js")))
        self.assertFalse(os.path.exists(os.path.join(self.integration, "alte_datei.py")))

        # Alte Fassung bleibt vollständig und unverändert im Backup-Ordner.
        with open(os.path.join(self.backup, "manifest.json.bak"), "rb") as f:
            self.assertIn(b'"2026.1.1.1"', f.read())
        self.assertTrue(os.path.exists(os.path.join(self.backup, "alte_datei.py")))

        # manifest.json selbst existiert im Backup-Ordner NICHT mehr - sonst
        # würde HA ihn beim nächsten Neustart als zweite Integration mit
        # derselben Domain laden (siehe Moduldocstring in aktualisierung.py).
        self.assertFalse(os.path.exists(os.path.join(self.backup, "manifest.json")))

        # Kein Staging-Rest übrig.
        self.assertFalse(os.path.exists(self.staging))

    def test_rollback_stellt_manifest_der_alten_fassung_wieder_her(self):
        """Schlägt der zweite Rename fehl, muss der Backup-Ordner nicht nur
        zurückbenannt werden, sondern auch sein manifest.json - sonst stünde
        die wiederhergestellte alte Fassung ohne gültiges Manifest da."""
        zip_bytes = _zip_bauen({
            "manifest.json": _GUELTIGES_MANIFEST,
            "neue_datei.py": b"# neue Fassung\n",
        })
        echter_rename = os.rename

        def zweiten_rename_kaputt_machen(src, dst):
            if src == self.staging and dst == self.integration:
                raise OSError("simulierter Fehler")
            echter_rename(src, dst)

        with mock.patch("aktualisierung.os.rename", side_effect=zweiten_rename_kaputt_machen):
            with self.assertRaises(a.AktualisierungsFehler):
                a.entpacken_pruefen_tauschen(zip_bytes, self.integration, self.staging, self.backup)

        with open(os.path.join(self.integration, "manifest.json"), "rb") as f:
            self.assertIn(b'"2026.1.1.1"', f.read())
        self.assertTrue(os.path.exists(os.path.join(self.integration, "alte_datei.py")))
        self.assertFalse(os.path.exists(self.backup))

    def test_entfernt_pycache_aus_dem_staging_ordner(self):
        zip_bytes = _zip_bauen({
            "manifest.json": _GUELTIGES_MANIFEST,
            "__pycache__/alte.cpython-313.pyc": b"\x00\x00",
        })
        a.entpacken_pruefen_tauschen(zip_bytes, self.integration, self.staging, self.backup)
        self.assertFalse(os.path.exists(os.path.join(self.integration, "__pycache__")))


class AbgelehnteFaelle(unittest.TestCase):
    def setUp(self):
        self.arbeitsordner = tempfile.mkdtemp()
        self.integration = os.path.join(self.arbeitsordner, "audi_dashboard")
        self.staging = os.path.join(self.arbeitsordner, "audi_dashboard_update_staging")
        self.backup = os.path.join(self.arbeitsordner, "audi_dashboard_backup")
        os.makedirs(self.integration)
        with open(os.path.join(self.integration, "manifest.json"), "wb") as f:
            f.write(b'{"domain": "audi_dashboard", "version": "2026.1.1.1"}')

    def tearDown(self):
        shutil.rmtree(self.arbeitsordner, ignore_errors=True)

    def _unangetastet(self):
        """Nach jedem abgelehnten Fall muss der Live-Ordner exakt so
        dastehen wie vorher - keiner der Fehlerfälle darf vor dem Tausch
        etwas am Live-Ordner verändern (siehe Moduldocstring)."""
        with open(os.path.join(self.integration, "manifest.json"), "rb") as f:
            self.assertIn(b'"2026.1.1.1"', f.read())
        self.assertFalse(os.path.exists(self.backup))

    def test_lehnt_falsche_domain_ab(self):
        zip_bytes = _zip_bauen({
            "manifest.json": b'{"domain": "hacs", "version": "2099.1.1.1"}',
        })
        with self.assertRaises(a.AktualisierungsFehler):
            a.entpacken_pruefen_tauschen(zip_bytes, self.integration, self.staging, self.backup)
        self._unangetastet()

    def test_lehnt_fehlendes_version_feld_ab(self):
        zip_bytes = _zip_bauen({
            "manifest.json": b'{"domain": "audi_dashboard"}',
        })
        with self.assertRaises(a.AktualisierungsFehler):
            a.entpacken_pruefen_tauschen(zip_bytes, self.integration, self.staging, self.backup)
        self._unangetastet()

    def test_lehnt_fehlende_manifest_ab(self):
        zip_bytes = _zip_bauen({"irgendeine_datei.py": b"# leer\n"})
        with self.assertRaises(a.AktualisierungsFehler):
            a.entpacken_pruefen_tauschen(zip_bytes, self.integration, self.staging, self.backup)
        self._unangetastet()

    def test_lehnt_kaputtes_zip_ab(self):
        with self.assertRaises(a.AktualisierungsFehler):
            a.entpacken_pruefen_tauschen(
                b"das ist kein Zip", self.integration, self.staging, self.backup
            )
        self._unangetastet()

    def test_lehnt_archiv_ohne_gesuchten_pfad_ab(self):
        puffer = io.BytesIO()
        with zipfile.ZipFile(puffer, "w") as archiv:
            archiv.writestr("irgendein-anderes-repo-main/README.md", b"# anderes Projekt\n")
        with self.assertRaises(a.AktualisierungsFehler):
            a.entpacken_pruefen_tauschen(
                puffer.getvalue(), self.integration, self.staging, self.backup
            )
        self._unangetastet()


class NeustartNoetig(unittest.TestCase):
    """Die Unterscheidung "braucht das einen Neustart?" ist bewusst schief.

    Sagt sie faelschlich "nein", laeuft neuer Python-Code nie an und der Fehler
    wird woanders gesucht. Ein ueberfluessiger Neustart kostet eine Minute.
    Deshalb: nur frontend/ und manifest.json duerfen ohne Neustart wechseln.
    """

    def setUp(self):
        self.basis = tempfile.mkdtemp()
        self.alt = os.path.join(self.basis, "alt")
        self.neu = os.path.join(self.basis, "neu")
        for ordner in (self.alt, self.neu):
            os.makedirs(os.path.join(ordner, "frontend"))
            self._schreiben(ordner, "manifest.json", '{"version": "1"}')
            self._schreiben(ordner, "__init__.py", "x = 1\n")
            self._schreiben(ordner, "frontend/app.js", "alert(1)\n")

    def tearDown(self):
        shutil.rmtree(self.basis, ignore_errors=True)

    def _schreiben(self, ordner, rel, inhalt):
        pfad = os.path.join(ordner, *rel.split("/"))
        os.makedirs(os.path.dirname(pfad), exist_ok=True)
        with open(pfad, "w", encoding="utf-8") as datei:
            datei.write(inhalt)

    def test_nichts_geaendert_braucht_keinen_neustart(self):
        self.assertFalse(a.neustart_noetig(self.alt, self.neu))

    def test_nur_frontend_braucht_keinen_neustart(self):
        self._schreiben(self.neu, "frontend/app.js", "alert(2)\n")
        self._schreiben(self.neu, "frontend/neu.css", "body{}\n")
        self.assertFalse(a.neustart_noetig(self.alt, self.neu))

    def test_manifest_allein_braucht_keinen_neustart(self):
        # Die Versionsnummer aendert sich bei JEDER Veroeffentlichung - ohne
        # diese Ausnahme waere jedes Update ein Neustart-Update.
        self._schreiben(self.neu, "manifest.json", '{"version": "2"}')
        self._schreiben(self.neu, "frontend/app.js", "alert(2)\n")
        self.assertFalse(a.neustart_noetig(self.alt, self.neu))

    def test_geaenderte_python_datei_braucht_neustart(self):
        self._schreiben(self.neu, "__init__.py", "x = 2\n")
        self.assertTrue(a.neustart_noetig(self.alt, self.neu))

    def test_neue_python_datei_braucht_neustart(self):
        self._schreiben(self.neu, "neu.py", "y = 1\n")
        self.assertTrue(a.neustart_noetig(self.alt, self.neu))

    def test_entfallene_python_datei_braucht_neustart(self):
        os.remove(os.path.join(self.neu, "__init__.py"))
        self.assertTrue(a.neustart_noetig(self.alt, self.neu))

    def test_services_yaml_braucht_neustart(self):
        # Nicht weil es zwingend waere, sondern weil die Regel pessimistisch ist.
        self._schreiben(self.neu, "services.yaml", "dienst:\n")
        self.assertTrue(a.neustart_noetig(self.alt, self.neu))

    def test_pycache_zaehlt_nicht(self):
        self._schreiben(self.neu, "__pycache__/x.cpython-312.pyc", "egal")
        self.assertFalse(a.neustart_noetig(self.alt, self.neu))

    def test_zeilenenden_zaehlen_nicht_als_aenderung(self):
        """Am 05.09.2026 am echten Update-Weg gemessen.

        Die installierte Fassung lag mit CRLF auf der Platte, das Gitea-Archiv
        liefert LF - drei .py-Dateien galten dadurch als geaendert, obwohl sie
        nach Entfernen der CR byteweise identisch waren, und die Erkennung
        verlangte einen Neustart fuer einen rein kosmetischen Unterschied."""
        with open(os.path.join(self.alt, "__init__.py"), "w", newline="\r\n", encoding="utf-8") as datei:
            datei.write("x = 1\ny = 2\n")
        with open(os.path.join(self.neu, "__init__.py"), "w", newline="\n", encoding="utf-8") as datei:
            datei.write("x = 1\ny = 2\n")
        self.assertFalse(a.neustart_noetig(self.alt, self.neu))

    def test_echte_aenderung_bleibt_erkannt_trotz_zeilenenden(self):
        with open(os.path.join(self.alt, "__init__.py"), "w", newline="\r\n", encoding="utf-8") as datei:
            datei.write("x = 1\n")
        with open(os.path.join(self.neu, "__init__.py"), "w", newline="\n", encoding="utf-8") as datei:
            datei.write("x = 2\n")
        self.assertTrue(a.neustart_noetig(self.alt, self.neu))

    def test_binaerdatei_wird_nicht_normalisiert(self):
        # Bei einer Schriftdatei ist 0x0D0A echter Inhalt - hier darf nicht
        # normalisiert werden, sonst verschwaende ein echter Unterschied.
        for ordner, inhalt in ((self.alt, b"\x00\r\n\x01"), (self.neu, b"\x00\n\x01")):
            with open(os.path.join(ordner, "schrift.woff2"), "wb") as datei:
                datei.write(inhalt)
        self.assertTrue(a.neustart_noetig(self.alt, self.neu))

    def test_unlesbarer_ordner_gilt_als_neustart(self):
        self.assertTrue(a.neustart_noetig(os.path.join(self.basis, "gibtsnicht"), self.neu))


class VersionsOrdnung(unittest.TestCase):
    """Nur eine NACHWEISLICH neuere Fassung ist ein Update.

    Bis zum 05.09.2026 stand in version_pruefen() `remote_version !=
    eigene_version` - reine Ungleichheit ohne Richtung. Lief die Instanz einer
    Veroeffentlichung voraus, bot die Integration an, sich auf die AELTERE
    Fassung zu aktualisieren (vom Eigentuemer gemeldet: .16 wurde angeboten,
    waehrend .18 lief)."""

    def test_neuere_fassung_ist_ein_update(self):
        self.assertIs(a.ist_neuer("2026.9.5.18", "2026.9.5.16"), True)
        self.assertIs(a.ist_neuer("2026.10.1.1", "2026.9.5.99"), True)
        self.assertIs(a.ist_neuer("2027.1.1.1", "2026.12.31.9"), True)

    def test_aeltere_fassung_ist_keins(self):
        self.assertIs(a.ist_neuer("2026.9.5.16", "2026.9.5.18"), False)
        self.assertIs(a.ist_neuer("2026.9.4.26", "2026.9.5.1"), False)

    def test_gleichstand_ist_keins(self):
        self.assertIs(a.ist_neuer("2026.9.5.18", "2026.9.5.18"), False)

    def test_fehlende_stellen_zaehlen_als_null(self):
        self.assertIs(a.ist_neuer("2026.9.4", "2026.9.4.0"), False)
        self.assertIs(a.ist_neuer("2026.9.4.1", "2026.9.4"), True)

    def test_nicht_sortierbar_bleibt_offen(self):
        # Formatwechsel: lieber "nicht entscheidbar" als eine erfundene
        # Reihenfolge. "4b" darf nicht klaglos als 4 gelesen werden.
        self.assertIsNone(a.ist_neuer("2026.9.4b", "2026.9.4.1"))
        self.assertIsNone(a.ist_neuer("v2", "2026.9.5.1"))
        self.assertIsNone(a.ist_neuer("", "2026.9.5.1"))

    def test_zeichengleich_ist_gleich_auch_bei_fremdem_format(self):
        self.assertIs(a.ist_neuer("beliebig", "beliebig"), False)


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