audi-app/tests/fahrterkennung/test_zuendungspause.py

#!/usr/bin/env python3
"""Die Wartezeit nach dem Zuendungs-Aus und der Nachlauf 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):
        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):
        self.werte = FakeWerte()


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, minuten=15, fahrten=None):
        self.fahrt_start_ts = start_ts
        self.fahrt_sperre = asyncio.Lock()
        self.zuordnung = FakeZuordnung()
        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 PAUSE_S

        return pausenzeit

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

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



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_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 Neuschreibung(Basis):
    """Home Assistant schreibt denselben Wert auch neu.

    Am 04.09.2026 gezaehlt: von 22 Zustandszeilen der Zuendung waren nur 6
    echte Wechsel. Die uebrigen 16 waren Neuschreibungen - der Wert selbst
    hatte sich nicht geruehrt.

    Der Beobachter unterscheidet das nicht, er reicht neu und alt durch. Faellt
    so eine Neuschreibung von "aus" in die laufende Wartezeit, begann die
    Wartezeit von vorn - und bei genuegend vielen nie ein Ende.
    """

    async def test_neuschreibung_verlaengert_die_wartezeit_nicht(self):
        k = FakeKoordinator()
        await f.zuendung_geaendert(k, "off", "on", AUS)
        wartet = k.ende_wartet
        self.assertIsNotNone(wartet, "nach dem Aus wartet eine Bestaetigung")

        # Dieselbe Meldung noch einmal, ohne Wertwechsel.
        await f.zuendung_geaendert(k, "off", "off", AUS + datetime.timedelta(seconds=30))
        self.assertIs(
            k.ende_wartet, wartet, "es wartet weiterhin dieselbe Bestaetigung"
        )

        await k.ausklingen()
        self.assertEqual(len(k.beendet), 1, "und sie kommt zum Zug")
        self.assertEqual(
            k.beendet[0][1],
            AUS - datetime.timedelta(seconds=f.ZUENDUNG_NACHLAUF_S),
            "mit dem Ende aus dem ERSTEN Aus, nicht dem der Neuschreibung",
        )

    async def test_echtes_zweites_aus_zaehlt_weiterhin(self):
        """Die Gegenprobe: ein Wechsel darf die Bestaetigung sehr wohl ersetzen."""
        k = FakeKoordinator()
        await f.zuendung_geaendert(k, "off", "on", AUS)
        erstes = k.ende_wartet
        await f.zuendung_geaendert(k, "on", "off", AUS + datetime.timedelta(seconds=30))
        await f.zuendung_geaendert(k, "off", "on", AUS + datetime.timedelta(seconds=60))
        self.assertIsNotNone(k.ende_wartet)
        self.assertNotEqual(
            k.ende_wartet, erstes, "ein echtes zweites Aus setzt ein neues Ende"
        )

    async def test_neuschreibung_ohne_laufende_fahrt_geht_weiter_durch(self):
        """_spaetes_aus() braucht auch ein off -> off.

        Ein zweites, spaeter gepuffertes Aus traegt das Ende einer von Hand
        beendeten Fahrt nach. Ein Filter ganz oben haette genau das
        verschluckt - deshalb sitzt die Pruefung nur in Zweig 2.
        """
        gesehen = []

        async def spaetes_aus(k, ereigniszeit, jetzt):
            gesehen.append(ereigniszeit)

        echt = f._spaetes_aus
        f._spaetes_aus = spaetes_aus
        try:
            k = FakeKoordinator(start_ts=None)
            await f.zuendung_geaendert(k, "off", "off", AUS)
            self.assertEqual(
                len(gesehen), 1, "ohne laufende Fahrt muss das Aus ankommen"
            )
        finally:
            f._spaetes_aus = echt

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_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)