audi-app/tests/fahrterkennung/test_zuendungspause.py

#!/usr/bin/env python3
"""Die Wartezeit nach dem Zuendungs-Aus, und die beiden Nachlaeufe des Geraets.

Warum es das gibt (03.09.2026): das Trip-Signal des FMM003 endet erst nach
seinem "Ignition OFF Timeout" von 900 s - und der Schlaf-Timeout des Geraets
stand auf denselben 900 s. Beide Uhren starten beim Zuendungs-Aus und fallen in
derselben Sekunde. Am 02.09.2026 ist das zweimal beobachtet worden, mit
entgegengesetztem Ausgang: die Fahrt um 16:54 verlor ihr Trip-Ende (danach
1 h 48 min ohne einen einzigen Datensatz), die um 21:43 bekam es um 21:58:16,
genau 901 s spaeter. Ein Wettlauf, den unsere Seite nicht gewinnen kann.

Deshalb loest jetzt die Zuendung aus, und die Wartezeit laeuft bei uns - auf
einer Maschine, die nicht schlaeft. Geprueft wird genau das: dass eine
zurueckkehrende Zuendung die Fahrt fortsetzt statt sie zu teilen, dass die
Nachlaeufe des Geraets abgezogen werden, und dass ein spaet eintreffendes Aus
keine Phantomfahrt erzeugt.

Aufruf: python3 tests/fahrterkennung/test_zuendungspause.py
(braucht das homeassistant-Paket, weil die Integration es importiert - also
z. B. im Test-Container)
"""

import asyncio
import datetime
import os
import sys
import unittest

_HIER = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, os.path.dirname(os.path.dirname(_HIER)))

from custom_components.audi_dashboard import fahrterkennung as f  # noqa: E402

T0 = datetime.datetime(2026, 9, 2, 16, 32, 18, tzinfo=datetime.UTC)
AUS = datetime.datetime(2026, 9, 2, 16, 54, 21, tzinfo=datetime.UTC)
# So kurz, dass der Test schnell bleibt - die Laenge ist hier nicht der Punkt,
# sondern ob sie abgebrochen wird.
PAUSE = 0.15


class FakeWerte:
    def __init__(self, trip=""):
        self.TRIP_SENSOR = trip
        self.ZUENDUNG_SENSOR = "binary_sensor.zuendung"
        # Ohne Meldezeit-Sensor keine Geraetezeit und keine Klemme - die Tests
        # pruefen die Nachlaufrechnung, nicht die Zuordnung.
        self.MELDEZEIT_SENSOR = ""


class FakeZuordnung:
    def __init__(self, trip=""):
        self.werte = FakeWerte(trip)


class FakeAblage:
    def __init__(self, minuten=15, fahrten=None):
        self.profil = {"einstellungen": {"fahrten_pausenzeit_min": minuten}}
        self.fahrten = fahrten or []

    async def profil_lesen(self):
        return self.profil

    async def fahrten_lesen(self):
        return self.fahrten


class FakeKoordinator:
    def __init__(self, start_ts=T0, trip="", minuten=15, fahrten=None):
        self.fahrt_start_ts = start_ts
        self.fahrt_sperre = asyncio.Lock()
        self.zuordnung = FakeZuordnung(trip)
        self.ablage = FakeAblage(minuten, fahrten)
        self.beendet = []
        self._aufgabe = None

    def warte_ende_ab(self, koroutine):
        self.warte_ende_ab_abbrechen()
        self._aufgabe = asyncio.ensure_future(koroutine)

    def warte_ende_ab_abbrechen(self):
        if self._aufgabe is not None and not self._aufgabe.done():
            self._aufgabe.cancel()
        self._aufgabe = None

    async def fahrt_start_setzen(self, ts):
        self.fahrt_start_ts = ts

    async def fahrzeugstatus_veroeffentlichen(self):
        pass

    async def ausklingen(self):
        """Der wartenden Aufgabe Zeit geben, fertig zu werden."""
        if self._aufgabe is None:
            return
        try:
            await asyncio.wait_for(asyncio.shield(self._aufgabe), timeout=2)
        except (asyncio.CancelledError, asyncio.TimeoutError):
            pass


class Basis(unittest.IsolatedAsyncioTestCase):
    def _pause_kurz(self):
        async def pausenzeit(k):
            return 0 if k.zuordnung.werte.TRIP_SENSOR else PAUSE

        return pausenzeit

    async def asyncSetUp(self):
        self._echt = (
            f._geraetezeit, f.fahrt_beenden, f._pausenzeit_s, f._nachlauf_gegenpruefen,
        )

        async def geraetezeit(k, ereigniszeit, standard):
            return ereigniszeit or standard

        async def beenden(k, start_ts, ende_ts):
            k.beendet.append((start_ts, ende_ts))
            await k.fahrt_start_setzen(None)

        async def gegenpruefen(k, start_ts, signal_ende, gerechnet):
            # Liest den Verlauf und aendert nichts an der Rechnung (siehe dort);
            # ohne echtes hass-Objekt ist sie hier nur im Weg.
            return None

        f._geraetezeit = geraetezeit
        f.fahrt_beenden = beenden
        f._pausenzeit_s = self._pause_kurz()
        f._nachlauf_gegenpruefen = gegenpruefen

    async def asyncTearDown(self):
        (
            f._geraetezeit, f.fahrt_beenden, f._pausenzeit_s, f._nachlauf_gegenpruefen,
        ) = self._echt


class Wartezeit(Basis):
    async def test_aus_schliesst_nicht_sofort(self):
        k = FakeKoordinator()
        await f.zuendung_geaendert(k, "off", "on", AUS)
        self.assertEqual(
            k.beendet, [], "waehrend der Wartezeit darf nichts geschrieben sein"
        )
        self.assertIsNotNone(k.fahrt_start_ts)
        await k.ausklingen()
        self.assertEqual(len(k.beendet), 1, "nach der Wartezeit muss die Fahrt enden")

    async def test_zuendung_kommt_zurueck_und_die_fahrt_laeuft_weiter(self):
        """Der eigentliche Zweck der Wartezeit."""
        k = FakeKoordinator()
        await f.zuendung_geaendert(k, "off", "on", AUS)
        await asyncio.sleep(PAUSE / 3)
        await f.zuendung_geaendert(k, "on", "off", AUS + datetime.timedelta(seconds=20))
        await asyncio.sleep(PAUSE * 2)
        self.assertEqual(k.beendet, [], "eine zurueckgekehrte Zuendung beendet nichts")
        self.assertEqual(k.fahrt_start_ts, T0, "es bleibt dieselbe Fahrt")

    async def test_start_stopp_zerlegt_die_fahrt_nicht(self):
        """Der RS4 stellt an der Ampel den Motor ab - aus einer Drehzahlquelle
        ist das nicht vom Abstellen zu unterscheiden."""
        k = FakeKoordinator()
        t = AUS
        for _ in range(4):
            await f.zuendung_geaendert(k, "off", "on", t)
            await asyncio.sleep(PAUSE / 5)
            t += datetime.timedelta(seconds=45)
            await f.zuendung_geaendert(k, "on", "off", t)
            t += datetime.timedelta(seconds=200)
        self.assertEqual(k.beendet, [])
        self.assertEqual(k.fahrt_start_ts, T0)

    async def test_ohne_wartezeit_bleibt_es_wie_bisher(self):
        """Mit zugeordnetem Trip-Signal wartet das Geraet - wir nicht."""
        k = FakeKoordinator(trip="binary_sensor.trip")
        await f.zuendung_geaendert(k, "off", "on", AUS)
        self.assertEqual(len(k.beendet), 1, "ohne Pause wird sofort geschlossen")


class Nachlauf(Basis):
    async def test_zuendungsweg_zieht_180_s_ab(self):
        k = FakeKoordinator()
        await f.zuendung_geaendert(k, "off", "on", AUS)
        await k.ausklingen()
        _, ende = k.beendet[0]
        self.assertEqual(ende, AUS - datetime.timedelta(seconds=180))

    async def test_tripweg_zieht_beide_nachlaeufe_ab(self):
        """900 s Trip-Nachlauf plus 180 s des Zuendungselements: das Trip-Ende
        haengt selbst an der bereits verzoegerten Zuendung."""
        k = FakeKoordinator(trip="binary_sensor.trip")
        await f.zuendung_geaendert(k, "off", "on", AUS)
        _, ende = k.beendet[0]
        self.assertEqual(ende, AUS - datetime.timedelta(seconds=1080))

    async def test_zu_kurz_wird_auf_den_beginn_geklemmt(self):
        k = FakeKoordinator(start_ts=AUS - datetime.timedelta(seconds=60))
        await f.zuendung_geaendert(k, "off", "on", AUS)
        await k.ausklingen()
        start, ende = k.beendet[0]
        self.assertEqual(ende, start, "kuerzer als der Nachlauf heisst: nicht gefahren")


class SpaetesAus(Basis):
    async def test_aus_zur_geschlossenen_fahrt_wird_verworfen(self):
        """Gepuffert eintreffendes Aus, dessen Geraetezeit in die schon
        geschlossene Fahrt faellt - es darf nichts ausloesen."""
        ende = AUS - datetime.timedelta(seconds=180)
        k = FakeKoordinator(
            start_ts=None,
            fahrten=[{"ts_start": T0.isoformat(), "ts_end": ende.isoformat()}],
        )
        await f.zuendung_geaendert(k, "off", "on", AUS)
        self.assertEqual(k.beendet, [])
        self.assertIsNone(k.fahrt_start_ts, "es darf keine Fahrt entstehen")

    async def test_aus_ohne_jede_fahrt_meldet_sich(self):
        k = FakeKoordinator(start_ts=None, fahrten=[])
        with self.assertLogs(f._LOGGER, level="WARNING") as protokoll:
            await f.zuendung_geaendert(k, "off", "on", AUS)
        self.assertTrue(
            any("entgangen" in z for z in protokoll.output),
            "ein Aus ohne passende Fahrt gehoert ins Protokoll",
        )


class Einstellung(unittest.IsolatedAsyncioTestCase):
    """_pausenzeit_s selbst - hier ungepatcht."""

    async def test_wert_aus_dem_profil(self):
        k = FakeKoordinator(minuten=25)
        self.assertEqual(await f._pausenzeit_s(k), 25 * 60)

    async def test_trip_signal_schlaegt_die_einstellung(self):
        k = FakeKoordinator(minuten=25, trip="binary_sensor.trip")
        self.assertEqual(await f._pausenzeit_s(k), 0)

    async def test_unsinn_faellt_auf_den_standard_zurueck(self):
        k = FakeKoordinator()
        k.ablage.profil = {"einstellungen": {"fahrten_pausenzeit_min": "viel"}}
        self.assertEqual(await f._pausenzeit_s(k), f.STANDARD_PAUSENZEIT_MIN * 60)

    async def test_obergrenze_greift(self):
        k = FakeKoordinator(minuten=9999)
        self.assertEqual(await f._pausenzeit_s(k), f.PAUSENZEIT_MAX_MIN * 60)

    async def test_null_ist_erlaubt(self):
        """Wer keine Pause will, bekommt keine - dann endet die Fahrt sofort."""
        k = FakeKoordinator(minuten=0)
        self.assertEqual(await f._pausenzeit_s(k), 0)


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