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)
# Zwei verschiedene Dinge, seit die Pausenregel in GERAETEZEIT rechnet:
#
# PAUSE_S ist das Fenster, in dem eine zurueckkehrende Zuendung noch dieselbe
# Fahrt ist - gemessen an den Zeitstempeln des Geraets. Es muss realistisch
# sein, sonst prueft der Test etwas anderes als der Betrieb tut.
#
# WARTEN ist die Wanduhrzeit, die der Test tatsaechlich absitzt. Sie darf kurz
# sein; ob abgebrochen wird, haengt nicht an ihrer Laenge.
PAUSE_S = 900
WARTEN = 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
        self.ende_wartet = None

    def warte_ende_ab(self, koroutine, wartet=None):
        self.warte_ende_ab_abbrechen()
        self.ende_wartet = wartet
        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
        self.ende_wartet = 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_S

        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

        async def nach_pause(k, start_ts, ende_ts, pause_s):
            # Wie das Original, nur mit kurzer Wanduhrzeit - die echte
            # Pausenzeit steckt weiterhin in der Entscheidung, nicht im Warten.
            await asyncio.sleep(WARTEN)
            await asyncio.shield(f._jetzt_schliessen(k, start_ts, ende_ts))

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

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


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(WARTEN / 3)
        await f.zuendung_geaendert(k, "on", "off", AUS + datetime.timedelta(seconds=20))
        await asyncio.sleep(WARTEN * 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(WARTEN / 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 GepufferterStapel(Basis):
    """Der Fehler vom 04.09.2026: zwei Fahrten wurden zu einer.

    Das Geraet war fuenf Stunden ohne Verbindung. Als es sich meldete, kamen
    das Zuendungs-Aus der ersten Fahrt und das Zuendungs-Ein der zweiten in
    DERSELBEN Sekunde bei Home Assistant an (23:36:50). Ihre Geraetezeiten
    lagen 5 Stunden 12 auseinander: 18:23:32 und 23:35:43.

    Nach der Ankunft gerechnet sieht das aus wie eine Zuendung, die sofort
    zurueckkommt - also dieselbe Fahrt. Genau so wurde gerechnet, und dabei ist
    aus zwei Fahrten eine geworden: 18:10 bis 23:42, 5 Stunden 33, ohne
    Strecke. Die zweite Fahrt gab es gar nicht.

    Die Ankunftszeit spielt hier bewusst KEINE Rolle - beide Aufrufe folgen
    unmittelbar aufeinander, so wie sie es an dem Abend taten.
    """

    async def test_zwei_fahrten_bleiben_zwei(self):
        beginn1 = datetime.datetime(2026, 9, 4, 16, 10, 27, tzinfo=datetime.UTC)
        aus1 = datetime.datetime(2026, 9, 4, 18, 23, 32, tzinfo=datetime.UTC)
        an2 = datetime.datetime(2026, 9, 4, 21, 35, 43, tzinfo=datetime.UTC)
        k = FakeKoordinator(start_ts=beginn1)

        await f.zuendung_geaendert(k, "off", "on", aus1)
        await f.zuendung_geaendert(k, "on", "off", an2)

        self.assertEqual(
            len(k.beendet), 1, "die erste Fahrt muss geschlossen worden sein"
        )
        start, ende = k.beendet[0]
        self.assertEqual(start, beginn1)
        self.assertEqual(
            ende, aus1 - datetime.timedelta(seconds=f.ZUENDUNG_NACHLAUF_S),
            "geschlossen wird auf das Zuendungs-Aus minus Nachlauf",
        )
        self.assertEqual(
            k.fahrt_start_ts, an2, "und die zweite Fahrt beginnt mit ihrem eigenen an"
        )

    async def test_kurze_rueckkehr_bleibt_eine_fahrt(self):
        """Die Gegenprobe - sonst wuerde jeder Tankstopp die Fahrt zerreissen.

        Derselbe Ablauf, nur liegen die Geraetezeiten acht Minuten
        auseinander statt fuenf Stunden.
        """
        beginn = datetime.datetime(2026, 9, 4, 16, 10, 27, tzinfo=datetime.UTC)
        aus = datetime.datetime(2026, 9, 4, 18, 23, 32, tzinfo=datetime.UTC)
        an = aus + datetime.timedelta(minutes=8)
        k = FakeKoordinator(start_ts=beginn)

        await f.zuendung_geaendert(k, "off", "on", aus)
        await f.zuendung_geaendert(k, "on", "off", an)

        self.assertEqual(k.beendet, [], "innerhalb der Pause wird nichts geschlossen")
        self.assertEqual(k.fahrt_start_ts, beginn, "es bleibt dieselbe Fahrt")

    async def test_genau_an_der_grenze_bleibt_es_eine_fahrt(self):
        """Die Pausenzeit gehoert noch zur Fahrt, nicht schon zur naechsten.

        Gemessen wird ab dem ECHTEN Ende (Zuendungs-Aus minus Nachlauf), nicht
        ab dem Zuendungssignal - sonst waere die Grenze um 180 s verschoben.
        """
        beginn = datetime.datetime(2026, 9, 4, 16, 10, 27, tzinfo=datetime.UTC)
        aus = datetime.datetime(2026, 9, 4, 18, 23, 32, tzinfo=datetime.UTC)
        echtes_ende = aus - datetime.timedelta(seconds=f.ZUENDUNG_NACHLAUF_S)
        k = FakeKoordinator(start_ts=beginn)

        await f.zuendung_geaendert(k, "off", "on", aus)
        await f.zuendung_geaendert(
            k, "on", "off", echtes_ende + datetime.timedelta(seconds=PAUSE_S)
        )
        self.assertEqual(k.beendet, [], "genau auf der Grenze noch dieselbe Fahrt")

        k2 = FakeKoordinator(start_ts=beginn)
        await f.zuendung_geaendert(k2, "off", "on", aus)
        await f.zuendung_geaendert(
            k2, "on", "off", echtes_ende + datetime.timedelta(seconds=PAUSE_S + 1)
        )
        self.assertEqual(len(k2.beendet), 1, "eine Sekunde darueber sind es zwei")

    async def test_ohne_wartendes_ende_wird_nichts_zerschnitten(self):
        """Ein "an" ohne vorheriges "aus" darf keine Fahrt beenden.

        Fehlt das Aus ganz - weil es nie ankam -, waere ein Schnitt auf
        Verdacht schlimmer als eine zu lange Fahrt. Dann passiert nichts,
        wie bisher.
        """
        beginn = datetime.datetime(2026, 9, 4, 16, 10, 27, tzinfo=datetime.UTC)
        k = FakeKoordinator(start_ts=beginn)
        await f.zuendung_geaendert(
            k, "on", "off", beginn + datetime.timedelta(hours=5)
        )
        self.assertEqual(k.beendet, [])
        self.assertEqual(k.fahrt_start_ts, beginn)

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)