Audit, zweite Runde: der Knopf schrieb ein erfundenes Ende (2026.9.3.14)
_letztes_lebenszeichen() fiel auf "jetzt" zurueck, sobald verlauf_lesen() eine leere Liste lieferte - und die bedeutet sowohl "nichts aufgezeichnet" als auch "nicht lesbar". Am 03.09. trat der zweite Fall ein: der Knopf traf einen Neustart, der recorder war noch nicht bereit, die Abfrage lief 2:47 und kam leer zurueck. Die Fahrt bekam 18:43 statt 17:59 als Ende, 74 Minuten statt 30. Drei Quellen in fester Reihenfolge statt einer: Aufzeichnung, dann hass.states (last_reported/last_updated - weiss dasselbe ohne Datenbank), dann "jetzt", und das nur noch mit Warnung. Vier Regressionstests, drei davon gegen den alten Stand nachweislich rot. Dazu aus der Bestandspruefung: odometer_km liegt in drei alten Tankvorgaengen als Zeichenkette vor, ablage.distanz_seit_letzter_tankung() rechnete damit float - str. Laeuft jetzt durch als_kilometerstand(). Der Eingang war schon zu; das ist Altbestand. Und acht Dienste hatten keinen Klarnamen in translations/de.json - jetzt 28 Dienste, 28 Klarnamen. Co-Authored-By: Claude Opus 5 <[email protected]>
This commit is contained in:
@@ -63,6 +63,27 @@ class FakeAblage:
|
||||
return False
|
||||
|
||||
|
||||
class FakeZustand:
|
||||
"""Ein Eintrag in hass.states - nur das, was hier gelesen wird."""
|
||||
|
||||
def __init__(self, gemeldet):
|
||||
self.last_reported = gemeldet
|
||||
self.last_updated = gemeldet
|
||||
|
||||
|
||||
class FakeStates:
|
||||
def __init__(self, zustand):
|
||||
self._zustand = zustand
|
||||
|
||||
def get(self, entity_id):
|
||||
return self._zustand
|
||||
|
||||
|
||||
class FakeHass:
|
||||
def __init__(self, zustand=None):
|
||||
self.states = FakeStates(zustand)
|
||||
|
||||
|
||||
class FakeKoordinator:
|
||||
def __init__(self, start_ts=T0, fahrten=None):
|
||||
self.fahrt_start_ts = start_ts
|
||||
@@ -261,5 +282,56 @@ class Nachziehen(Basis):
|
||||
self.assertEqual(aenderungen["ts_end"], echt.isoformat())
|
||||
|
||||
|
||||
class Lebenszeichen(unittest.IsolatedAsyncioTestCase):
|
||||
"""Woher das Fahrtende kommt, wenn der Knopf gedrueckt wird.
|
||||
|
||||
Am 03.09.2026 hat genau das eine falsche Fahrt erzeugt: der Knopf traf
|
||||
einen Neustart, der recorder war noch nicht bereit, verlauf_lesen() kam
|
||||
nach 2:47 leer zurueck - und "leer" ist von "nichts aufgezeichnet" nicht zu
|
||||
unterscheiden. Das Ende wurde "jetzt": 74 Minuten statt 30.
|
||||
"""
|
||||
|
||||
def setUp(self):
|
||||
self._echt = f.verlauf_lesen
|
||||
|
||||
def tearDown(self):
|
||||
f.verlauf_lesen = self._echt
|
||||
|
||||
@staticmethod
|
||||
def _k(punkte, zustand):
|
||||
async def lesen(hass, sensor, start, ende):
|
||||
return punkte
|
||||
|
||||
f.verlauf_lesen = lesen
|
||||
k = FakeKoordinator()
|
||||
k.hass = FakeHass(zustand)
|
||||
return k
|
||||
|
||||
async def test_aufzeichnung_gewinnt(self):
|
||||
punkt = T0 + datetime.timedelta(minutes=30)
|
||||
k = self._k([(punkt, "on")], FakeZustand(T0 + datetime.timedelta(hours=2)))
|
||||
self.assertEqual(await f._letztes_lebenszeichen(k, "binary_sensor.x", T0), punkt)
|
||||
|
||||
async def test_ohne_aufzeichnung_zaehlt_der_zustandsspeicher(self):
|
||||
"""Der eigentliche Fix: nicht "jetzt", sondern wann die Entitaet
|
||||
zuletzt gemeldet hat - das weiss hass.states ohne Datenbank."""
|
||||
gemeldet = T0 + datetime.timedelta(minutes=30)
|
||||
k = self._k([], FakeZustand(gemeldet))
|
||||
self.assertEqual(await f._letztes_lebenszeichen(k, "binary_sensor.x", T0), gemeldet)
|
||||
|
||||
async def test_meldung_vor_dem_fahrtbeginn_zaehlt_nicht(self):
|
||||
"""Sie kann nicht das Ende sein - die Fahrt begann ja spaeter."""
|
||||
k = self._k([], FakeZustand(T0 - datetime.timedelta(hours=1)))
|
||||
with self.assertLogs(f._LOGGER, level="WARNING"):
|
||||
ende = await f._letztes_lebenszeichen(k, "binary_sensor.x", T0)
|
||||
self.assertGreater(ende, T0)
|
||||
|
||||
async def test_ohne_jede_quelle_wird_gewarnt(self):
|
||||
""""Jetzt" bleibt der letzte Ausweg - aber nicht mehr stillschweigend."""
|
||||
k = self._k([], None)
|
||||
with self.assertLogs(f._LOGGER, level="WARNING"):
|
||||
await f._letztes_lebenszeichen(k, "binary_sensor.x", T0)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main(verbosity=2)
|
||||
|
||||
Reference in New Issue
Block a user