#!/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" 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)