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",
        })
        version = a.entpacken_pruefen_tauschen(
            zip_bytes, self.integration, self.staging, self.backup
        )
        self.assertEqual(version, "2099.1.1.1")

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


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