audi-app/tests/fahrterkennung/test_fahrtende_knopf.py

#!/usr/bin/env python3
"""Der Knopf "Fahrt beenden" und das Nachziehen des vorlaeufigen Endes.

Der Fall, fuer den es den Knopf gibt: das Fahrzeug steht in einer Tiefgarage
ohne Empfang. Das Geraet kann sein Zuendungs-Aus nicht melden, die Fahrt bliebe
offen, bis irgendwann gepufferte Datensaetze eintreffen. Der Knopf schliesst
sie auf das letzte Lebenszeichen - mehr weiss in dem Moment niemand.

Genau deshalb ist dieses Ende VORLAEUFIG: faehrt das Fahrzeug danach noch aus
der Garage heraus, kommt der Rest nach, und dann gehoert das Ende nachgezogen
statt verworfen. Geprueft wird beides - dass der Knopf das Richtige schliesst,
und dass die spaete Meldung die Fahrt korrigiert statt eine neue anzulegen.

Aufruf: python3 tests/fahrterkennung/test_fahrtende_knopf.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, 3, 8, 0, 0, tzinfo=datetime.UTC)
LEBENSZEICHEN = datetime.datetime(2026, 9, 3, 8, 20, 0, tzinfo=datetime.UTC)


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


class FakeZuordnung:
    werte = FakeWerte()


class FakeAblage:
    def __init__(self, fahrten=None):
        self.fahrten = fahrten or []
        self.aktualisiert = []

    async def profil_lesen(self):
        return {"einstellungen": {"fahrten_pausenzeit_min": 15}}

    async def fahrten_lesen(self):
        return self.fahrten

    async def fahrt_anhaengen(self, fahrt):
        self.fahrten.append(fahrt)

    async def fahrt_aktualisieren(self, trip_id, aenderungen):
        self.aktualisiert.append((trip_id, aenderungen))
        for fahrt in self.fahrten:
            if fahrt.get("trip_id") == trip_id:
                fahrt.update(aenderungen)
                return True
        return False


class FakeKoordinator:
    def __init__(self, start_ts=T0, fahrten=None):
        self.fahrt_start_ts = start_ts
        self.fahrt_sperre = asyncio.Lock()
        self.zuordnung = FakeZuordnung()
        self.ablage = FakeAblage(fahrten)
        self.hass = object()
        self.veroeffentlicht = 0
        self.abgebrochen = 0

    def warte_ende_ab_abbrechen(self):
        self.abgebrochen += 1

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

    async def fahrzeugstatus_veroeffentlichen(self):
        pass

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


def fahrt(trip_id, start, ende, vorlaeufig):
    return {
        "trip_id": trip_id,
        "ts_start": start.isoformat(),
        "ts_end": ende.isoformat(),
        "duration_s": int((ende - start).total_seconds()),
        "ende_vorlaeufig": vorlaeufig,
    }


class Basis(unittest.IsolatedAsyncioTestCase):
    async def asyncSetUp(self):
        self._echt = (
            f._letztes_lebenszeichen,
            f._geraetezeit,
            f.screening.durchfuehren,
            f.reifen.aktiver_satz,
        )

        async def lebenszeichen(k, sensor, seit):
            return LEBENSZEICHEN

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

        async def durchfuehren(k):
            return None

        def aktiver_satz(profil):
            return "sommer"

        f._letztes_lebenszeichen = lebenszeichen
        f._geraetezeit = geraetezeit
        f.screening.durchfuehren = durchfuehren
        f.reifen.aktiver_satz = aktiver_satz

    async def asyncTearDown(self):
        (
            f._letztes_lebenszeichen,
            f._geraetezeit,
            f.screening.durchfuehren,
            f.reifen.aktiver_satz,
        ) = self._echt


class Knopf(Basis):
    async def test_schliesst_auf_das_letzte_lebenszeichen(self):
        """Nicht auf 'jetzt': das Fahrzeug stand still, als es still stand -
        und genau dann liegen die beiden weit auseinander."""
        k = FakeKoordinator()
        self.assertTrue(await f.jetzt_beenden(k))
        self.assertEqual(len(k.ablage.fahrten), 1)
        gespeichert = k.ablage.fahrten[0]
        self.assertEqual(gespeichert["ts_end"], LEBENSZEICHEN.isoformat())
        self.assertTrue(gespeichert["ende_vorlaeufig"])
        self.assertIsNone(k.fahrt_start_ts)

    async def test_ohne_laufende_fahrt_passiert_nichts(self):
        k = FakeKoordinator(start_ts=None)
        self.assertFalse(await f.jetzt_beenden(k))
        self.assertEqual(k.ablage.fahrten, [])

    async def test_bricht_eine_wartende_ende_aufgabe_ab(self):
        """Sonst schloesse die Wartezeit die Fahrt ein zweites Mal."""
        k = FakeKoordinator()
        await f.jetzt_beenden(k)
        self.assertEqual(k.abgebrochen, 1)


class Nachziehen(Basis):
    async def _spaetes_aus(self, k, gemeldet):
        await f._spaetes_aus(k, gemeldet, gemeldet)

    async def test_spaeteres_aus_zieht_das_vorlaeufige_ende_nach(self):
        """Der eigentliche Zweck: nach der Tiefgarage kommt der Rest gepuffert
        nach, und die Fahrt endet dort, wo sie wirklich endete."""
        k = FakeKoordinator(
            start_ts=None,
            fahrten=[fahrt("t-1", T0, LEBENSZEICHEN, True)],
        )
        # Gemeldet wird das verzoegerte Zuendungs-Aus; echt ist es
        # ZUENDUNG_NACHLAUF_S frueher.
        echt = LEBENSZEICHEN + datetime.timedelta(minutes=12)
        gemeldet = echt + datetime.timedelta(seconds=f.ZUENDUNG_NACHLAUF_S)
        await self._spaetes_aus(k, gemeldet)

        self.assertEqual(len(k.ablage.aktualisiert), 1)
        _, aenderungen = k.ablage.aktualisiert[0]
        self.assertEqual(aenderungen["ts_end"], echt.isoformat())
        self.assertEqual(aenderungen["duration_s"], int((echt - T0).total_seconds()))
        self.assertFalse(aenderungen["ende_vorlaeufig"])
        self.assertNotIn(
            "edited_fields",
            aenderungen,
            "ts_end darf nicht als Handeingabe gelten - das blockierte die "
            "naechste Korrektur",
        )

    async def test_regulaer_geschlossene_fahrt_wird_nicht_angefasst(self):
        k = FakeKoordinator(
            start_ts=None,
            fahrten=[fahrt("t-1", T0, LEBENSZEICHEN, False)],
        )
        echt = LEBENSZEICHEN + datetime.timedelta(minutes=12)
        await self._spaetes_aus(
            k, echt + datetime.timedelta(seconds=f.ZUENDUNG_NACHLAUF_S)
        )
        self.assertEqual(k.ablage.aktualisiert, [])

    async def test_frueheres_aus_verkuerzt_die_fahrt_nicht(self):
        k = FakeKoordinator(
            start_ts=None,
            fahrten=[fahrt("t-1", T0, LEBENSZEICHEN, True)],
        )
        echt = LEBENSZEICHEN - datetime.timedelta(minutes=5)
        await self._spaetes_aus(
            k, echt + datetime.timedelta(seconds=f.ZUENDUNG_NACHLAUF_S)
        )
        self.assertEqual(k.ablage.aktualisiert, [])

    async def test_zu_weit_danach_wird_nicht_korrigiert(self):
        """Jenseits des Korrekturfensters gehoert das Aus wahrscheinlich zu
        einer ganz anderen Fahrt."""
        k = FakeKoordinator(
            start_ts=None,
            fahrten=[fahrt("t-1", T0, LEBENSZEICHEN, True)],
        )
        echt = LEBENSZEICHEN + datetime.timedelta(seconds=f.KORREKTUR_FENSTER_S + 60)
        with self.assertLogs(f._LOGGER, level="WARNING"):
            await self._spaetes_aus(
                k, echt + datetime.timedelta(seconds=f.ZUENDUNG_NACHLAUF_S)
            )
        self.assertEqual(k.ablage.aktualisiert, [])

    async def test_keine_verlaengerung_ueber_eine_andere_fahrt_hinweg(self):
        """Faengt zwischen dem vorlaeufigen Ende und dem Aus eine andere Fahrt
        an, gehoert das Aus zu ihr - zwei sich ueberlappende Fahrten waeren
        schlimmer als eine zu kurze."""
        alt_start = T0 - datetime.timedelta(hours=3)
        k = FakeKoordinator(
            start_ts=None,
            fahrten=[
                fahrt("t-alt", alt_start, alt_start + datetime.timedelta(minutes=10), True),
                fahrt("t-neu", T0, LEBENSZEICHEN, False),
            ],
        )
        echt = LEBENSZEICHEN + datetime.timedelta(minutes=5)
        await self._spaetes_aus(
            k, echt + datetime.timedelta(seconds=f.ZUENDUNG_NACHLAUF_S)
        )
        self.assertEqual(
            k.ablage.aktualisiert, [], "dazwischen liegt der Beginn von t-neu"
        )

    async def test_aeltere_vorlaeufige_fahrt_wird_gefunden(self):
        """Die Suche ist NICHT auf die juengste Fahrt beschraenkt: eine aeltere
        regulaer geschlossene Fahrt darf die Korrektur nicht verdecken."""
        frueher = T0 - datetime.timedelta(hours=5)
        k = FakeKoordinator(
            start_ts=None,
            fahrten=[
                fahrt("t-frueh", frueher, frueher + datetime.timedelta(minutes=20), False),
                fahrt("t-hand", T0, LEBENSZEICHEN, True),
            ],
        )
        echt = LEBENSZEICHEN + datetime.timedelta(minutes=8)
        await self._spaetes_aus(
            k, echt + datetime.timedelta(seconds=f.ZUENDUNG_NACHLAUF_S)
        )
        self.assertEqual(len(k.ablage.aktualisiert), 1)
        trip_id, aenderungen = k.ablage.aktualisiert[0]
        self.assertEqual(trip_id, "t-hand")
        self.assertEqual(aenderungen["ts_end"], echt.isoformat())


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