fix(phase26): persist DTC/MIL state + server speed_source (BUG-03, BUG-05)

BUG-03: DTC/MIL codes were never persisted; lost on OBD disconnect or reboot.
- Add new dtc table schema (code, description, MIL, first/last seen, IS_SET)
  and a migration that rebuilds the original unused NAME/STATUS/DTC_ID table.
- Persist MIL + codes to SQLite on each successful OBD poll (_persist_dtc_state).
- /api/dtc falls back to the last-persisted snapshot when OBD is offline/stale.
- Fix get_overview_stats to count the real dtc table (was querying non-existent
  dtc_codes table).
- Add cleared/since tracking via IS_SET + FIRST_SEEN_AT/LAST_SEEN_AT.

BUG-05: server trip replay hardcoded speed_source:'gps'; mirror Pi's
'obd if speed>0 else gps' logic in server get_trip_route.

Adds tests/test_dtc_persistence.py (7 tests).
This commit is contained in:
Mathieu Moran
2026-08-11 00:38:05 -04:00
parent cd0583ac84
commit e2f2314318
5 changed files with 251 additions and 15 deletions
+109
View File
@@ -0,0 +1,109 @@
import os
import tempfile
import unittest
from datetime import datetime
from modules import app_sql
class DtcPersistenceTest(unittest.TestCase):
"""Tests for persisted DTC/MIL state (BUG-03)."""
def setUp(self):
handle, self.path = tempfile.mkstemp(suffix=".db")
os.close(handle)
self.original_path = app_sql.sql_path
app_sql._conn = None
app_sql.sql_path = self.path
self.conn = app_sql.get_conn()
def tearDown(self):
if app_sql._conn is not None:
app_sql._conn.close()
app_sql._conn = None
app_sql.sql_path = self.original_path
os.unlink(self.path)
def test_save_and_read_back_dtc_state(self):
app_sql.save_dtc_state(True, [
{"code": "P0300", "description": "Random misfire"},
{"code": "P0420", "description": "Catalyst efficiency"},
])
state = app_sql.get_persisted_dtc_state()
self.assertIs(state["mil"], True)
self.assertEqual(
[c["code"] for c in state["codes"]],
["P0300", "P0420"],
)
self.assertEqual(state["codes"][0]["description"], "Random misfire")
self.assertEqual(app_sql.count_active_dtcs(), 2)
def test_cleared_code_removed_from_active(self):
app_sql.save_dtc_state(True, [
{"code": "P0300", "description": "Random misfire"},
{"code": "P0420", "description": "Catalyst efficiency"},
])
# A later poll reports only P0420 -> P0300 is cleared.
app_sql.save_dtc_state(True, [
{"code": "P0420", "description": "Catalyst efficiency"},
])
state = app_sql.get_persisted_dtc_state()
self.assertEqual([c["code"] for c in state["codes"]], ["P0420"])
self.assertEqual(app_sql.count_active_dtcs(), 1)
# The cleared code keeps its history row (IS_SET=0).
rows = self.conn.execute(
"SELECT code, IS_SET FROM dtc ORDER BY code"
).fetchall()
self.assertEqual(rows, [("P0300", 0), ("P0420", 1)])
def test_empty_poll_clears_all(self):
app_sql.save_dtc_state(True, [{"code": "P0300", "description": "Misfire"}])
app_sql.save_dtc_state(False, [])
self.assertEqual(app_sql.count_active_dtcs(), 0)
self.assertIs(app_sql.get_persisted_dtc_state()["mil"], False)
def test_first_seen_preserved_while_last_seen_bumps(self):
first = datetime(2026, 2, 1, 8, 0, 0)
later = datetime(2026, 2, 1, 8, 5, 0)
app_sql.save_dtc_state(True, [{"code": "P0300", "description": "M"}], first)
app_sql.save_dtc_state(True, [{"code": "P0300", "description": "M"}], later)
row = self.conn.execute(
"SELECT FIRST_SEEN_AT, LAST_SEEN_AT FROM dtc WHERE code = 'P0300'"
).fetchone()
self.assertEqual(row[0], first.isoformat())
self.assertEqual(row[1], later.isoformat())
def test_overview_stats_uses_dtc_table_not_broken_dtc_codes(self):
app_sql.save_dtc_state(True, [
{"code": "P0300", "description": "Random misfire"},
])
stats = app_sql.get_overview_stats()
self.assertEqual(stats["active_dtc"], 1)
def test_old_schema_dtc_table_migrated(self):
# Simulate the original (never-populated) dtc schema.
app_sql._conn.execute("DROP TABLE dtc")
app_sql._conn.execute(
"""CREATE TABLE dtc (
id integer PRIMARY KEY,
NAME text NOT NULL,
STATUS text,
DTC_ID text
)"""
)
app_sql._conn.commit()
app_sql._migrate_dtc_table(app_sql._conn)
columns = {
row[1] for row in app_sql._conn.execute("PRAGMA table_info(dtc)").fetchall()
}
self.assertIn("code", columns)
self.assertNotIn("DTC_ID", columns)
def test_no_persisted_state_returns_empty(self):
state = app_sql.get_persisted_dtc_state()
self.assertIsNone(state["mil"])
self.assertEqual(state["codes"], [])
if __name__ == "__main__":
unittest.main()