tobias
9ac69cdf44
Home Assistant schreibt denselben Zustandswert auch neu, wenn sich nur ein Attribut geruehrt hat. Am 04.09.2026 gezaehlt: von 22 Zustandszeilen der Zuendung waren nur 6 echte Wechsel, die uebrigen 16 Neuschreibungen - und alle 6 Wechsel lagen ausserhalb der Fahrten. Das Zuendungssignal ist unterwegs stabil. Der Beobachter unterschied das nicht. Fiel eine Neuschreibung von "aus" in die laufende Wartezeit, brach Zweig 2 die wartende Bestaetigung ab und begann die 15 Minuten von vorn. Am 04.09. ist es nicht passiert - die Neuschreibungen lagen hinter dem Fenster -, aber das Loch war da. _signalwechsel() bekommt jetzt mit, ob es ein echter Wechsel war. Eine Neuschreibung, waehrend fuer diese Fahrt schon eine Bestaetigung wartet, laesst sie stehen. Bewusst NICHT als Filter ganz oben: _spaetes_aus() verlaesst sich darauf, ein Zuendungs-Aus auch als off -> off zu sehen - daran haengt das Nachtragen vorlaeufiger Enden. Die Pruefung sitzt nur in Zweig 2, und ein eigener Test belegt, dass der andere Weg weiter offen ist. 21 Tests (von 18), Gegenprobe mit ausgeschalteter Pruefung scheitert wie erwartet an einem. py_compile sauber, audi_ha_test auf 2026.9.4.25 ohne Traceback. AGENTS.md haelt ausserdem fest: die Beschreibung des Beenden-Knopfes war zu weit gefasst (die Korrektur zieht nur nach hinten, nie nach vorn), und der flespi-Cache laesst sich ueber DELETE auf die Einstellung zum Neulesen zwingen - mit dem Haken, dass dabei ausstehende Werte verloren gingen.
436 lines
17 KiB
Python
436 lines
17 KiB
Python
#!/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 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_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)
|