audi-app/tests/fahrterkennung/test_signalwechsel.py

#!/usr/bin/env python3
"""Regressionstest: zwei Wechsel des Fahrtsignals in derselben Sekunde.

Der Fall, der ihn nötig gemacht hat (02.09.2026, im Verlauf nachgemessen):
das Gerät hatte gepuffert, deshalb kamen "Fahrt zu Ende" und "Fahrt begonnen"
gemeinsam um 21:22:12 bei Home Assistant an. Home Assistant startet für jeden
Zustandswechsel eine eigene Aufgabe, und der Beende-Zweig wartet mehrfach
(Meldezeit, Verlauf, Schreiben, Screening). Das "on" überholte ihn, sah
fahrt_start_ts noch gesetzt und fiel durch beide Zweige - die anschliessende
36-Minuten-Fahrt wurde live nie erfasst und tauchte erst über den Rückblick auf.

Geprüft wird deshalb genau die Verschränkung, nicht die Fahrterkennung als
Ganzes: die langsamen Teile (_geraetezeit, _echtes_ende, fahrt_beenden) sind
ersetzt, übrig bleibt die Frage, ob der zweite Wechsel den ersten abwarten muss.

Aufruf: python3 tests/fahrterkennung/test_signalwechsel.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)
T_WECHSEL = datetime.datetime(2026, 9, 2, 21, 22, 12, tzinfo=datetime.UTC)


class FakeWerte:
    TRIP_SENSOR = "binary_sensor.trip"
    ZUENDUNG_SENSOR = "binary_sensor.zuendung"


class FakeZuordnung:
    werte = FakeWerte()


class FakeKoordinator:
    """Nur das, was der Signalpfad anfasst."""

    def __init__(self, start_ts):
        self.fahrt_start_ts = start_ts
        self.fahrt_sperre = asyncio.Lock()
        self.beendet = []
        self.veroeffentlicht = 0
        # Dieser Test beschreibt die Welt MIT Trip-Signal: dort bringt das
        # Geraet seine Pausentoleranz mit, _pausenzeit_s() gibt 0 zurueck und
        # es wird sofort geschlossen. Gegenstand ist hier die Sperre, nicht
        # die Wartezeit - die hat test_zuendungspause.py.
        self.zuordnung = FakeZuordnung()

    def warte_ende_ab_abbrechen(self):
        pass

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

    async def fahrzeugstatus_veroeffentlichen(self):
        self.veroeffentlicht += 1


class SignalwechselTest(unittest.IsolatedAsyncioTestCase):
    async def asyncSetUp(self):
        self._echt = (f._geraetezeit, f._echtes_ende, f.fahrt_beenden)

        async def geraetezeit(k, ereigniszeit, standard):
            # Der eigentliche Punkt: dieser Schritt wartet wirklich (in echt
            # bis zu MELDEZEIT_FRIST_S = 2 s auf die Meldezeit des Geräts).
            await asyncio.sleep(0.05)
            return ereigniszeit or standard

        async def echtes_ende(k, start, signal_ende):
            return signal_ende

        async def beenden(k, start_ts, ende_ts):
            await asyncio.sleep(0.05)  # Schreiben und Screening brauchen Zeit
            k.beendet.append((start_ts, ende_ts))
            await k.fahrt_start_setzen(None)

        f._geraetezeit, f._echtes_ende, f.fahrt_beenden = (
            geraetezeit, echtes_ende, beenden,
        )

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

    async def test_off_und_on_in_derselben_sekunde(self):
        """Der genaue Fall vom 02.09.2026: die neue Fahrt darf nicht verloren
        gehen, während die alte noch geschlossen wird."""
        k = FakeKoordinator(T0)

        aus = asyncio.create_task(f.zuendung_geaendert(k, "off", "on", T_WECHSEL))
        # Dem "off" gerade so viel Vorlauf, dass es die Sperre hält und im
        # ersten await steht - genau die Lage, in der das "on" eintrifft.
        await asyncio.sleep(0)
        await asyncio.sleep(0)
        an = asyncio.create_task(f.zuendung_geaendert(k, "on", "off", T_WECHSEL))
        await asyncio.gather(aus, an)

        self.assertEqual(len(k.beendet), 1, "die alte Fahrt muss genau einmal enden")
        self.assertEqual(k.beendet[0][0], T0)
        self.assertIsNotNone(
            k.fahrt_start_ts,
            "nach dem Paar muss die neue Fahrt laufen - sonst ist sie verloren",
        )
        self.assertEqual(k.fahrt_start_ts, T_WECHSEL)

    async def test_nacheinander_unveraendert(self):
        """Der gewöhnliche Ablauf darf sich durch die Sperre nicht ändern."""
        k = FakeKoordinator(T0)
        await f.zuendung_geaendert(k, "off", "on", T_WECHSEL)
        self.assertEqual(len(k.beendet), 1)
        self.assertIsNone(k.fahrt_start_ts)

        await f.zuendung_geaendert(k, "on", "off", T_WECHSEL)
        self.assertEqual(k.fahrt_start_ts, T_WECHSEL)

    async def test_zweites_on_startet_keine_zweite_fahrt(self):
        k = FakeKoordinator(T0)
        await f.zuendung_geaendert(k, "on", "off", T_WECHSEL)
        self.assertEqual(k.fahrt_start_ts, T0, "die laufende Fahrt bleibt stehen")
        self.assertEqual(k.beendet, [])

    async def test_ohne_vorzustand_passiert_nichts(self):
        """Registrierung beim Start ist kein Wechsel - sonst Phantomfahrten."""
        k = FakeKoordinator(None)
        await f.zuendung_geaendert(k, "on", None, T_WECHSEL)
        self.assertIsNone(k.fahrt_start_ts)

    async def test_funkstille_wird_ignoriert(self):
        k = FakeKoordinator(T0)
        await f.zuendung_geaendert(k, "unavailable", "on", T_WECHSEL)
        self.assertEqual(k.fahrt_start_ts, T0)
        self.assertEqual(k.beendet, [])


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