126 lines
4.7 KiB
Python
126 lines
4.7 KiB
Python
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_clear_dtc_state_clears_active_and_mil(self):
|
|
app_sql.save_dtc_state(True, [
|
|
{"code": "P0300", "description": "Random misfire"},
|
|
{"code": "P0420", "description": "Catalyst efficiency"},
|
|
])
|
|
app_sql.clear_dtc_state()
|
|
state = app_sql.get_persisted_dtc_state()
|
|
self.assertIs(state["mil"], False)
|
|
self.assertEqual(state["codes"], [])
|
|
self.assertEqual(app_sql.count_active_dtcs(), 0)
|
|
# History rows are kept but no longer set.
|
|
rows = self.conn.execute(
|
|
"SELECT code, IS_SET FROM dtc ORDER BY code"
|
|
).fetchall()
|
|
self.assertEqual(rows, [("P0300", 0), ("P0420", 0)])
|
|
|
|
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()
|