Mathieu Moran
99035e86ee
- Pi history page gets a select checkbox column + 'Delete selected' button (mirrors the server UI) - POST /api/trips/delete batch endpoint - app_sql.delete_trips_batch() + deleted_trip_markers table so a later trip rebuild can't resurrect locally-deleted trips - Deletion is local-only: raw tracking rows are kept and the remote server is never notified (it keeps its archival copy recomputed from uploaded rows)
112 lines
3.7 KiB
Python
112 lines
3.7 KiB
Python
import os
|
|
import tempfile
|
|
import unittest
|
|
from datetime import datetime, timedelta
|
|
from unittest.mock import patch
|
|
|
|
from modules import app_sql
|
|
|
|
|
|
class TripSqlTest(unittest.TestCase):
|
|
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._last_trip_refresh = 0
|
|
app_sql.sql_path = self.path
|
|
self.conn = app_sql.get_conn()
|
|
# refresh_trips() and build_trip_summaries() suppress trips until the
|
|
# car ECU has been reached once (BUG-10 guard). Simulate a connected car.
|
|
self._car_connected = patch(
|
|
"modules.app_obd.was_car_ever_connected", return_value=True
|
|
)
|
|
self._car_connected.start()
|
|
|
|
def tearDown(self):
|
|
self._car_connected.stop()
|
|
if app_sql._conn is not None:
|
|
app_sql._conn.close()
|
|
app_sql._conn = None
|
|
app_sql._last_trip_refresh = 0
|
|
app_sql.sql_path = self.original_path
|
|
os.unlink(self.path)
|
|
|
|
def insert_motion(self, start_id, count, started, latitude):
|
|
rows = [
|
|
(
|
|
start_id + index,
|
|
"Yaris",
|
|
"test",
|
|
started + timedelta(seconds=index * 5),
|
|
latitude + index * 0.0001,
|
|
-73.5,
|
|
20,
|
|
None,
|
|
0,
|
|
)
|
|
for index in range(count)
|
|
]
|
|
self.conn.executemany(
|
|
"""INSERT INTO tracking (
|
|
id, VN, NAME, DATETIME, GPSX, GPSY, SPEED, ODOMETER, UPLOADED
|
|
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)""",
|
|
rows,
|
|
)
|
|
self.conn.commit()
|
|
|
|
def test_incremental_rebuild_preserves_trip_before_overlap_boundary(self):
|
|
started = datetime(2026, 1, 1, 8)
|
|
self.insert_motion(1, 160, started, 45.5)
|
|
self.insert_motion(161, 10, started + timedelta(minutes=30), 45.6)
|
|
app_sql.refresh_trips(force=True)
|
|
original = self.conn.execute(
|
|
"""SELECT START_TRACKING_ID, END_TRACKING_ID, DISTANCE_KM
|
|
FROM trips ORDER BY START_TRACKING_ID"""
|
|
).fetchall()
|
|
self.assertEqual(len(original), 2)
|
|
|
|
self.insert_motion(171, 1, started + timedelta(minutes=31), 45.601)
|
|
app_sql._last_trip_refresh = 0
|
|
app_sql.refresh_trips()
|
|
rebuilt = self.conn.execute(
|
|
"""SELECT START_TRACKING_ID, END_TRACKING_ID, DISTANCE_KM
|
|
FROM trips ORDER BY START_TRACKING_ID"""
|
|
).fetchall()
|
|
|
|
self.assertEqual(rebuilt[0], original[0])
|
|
|
|
def test_deleted_trip_is_not_resurrected_by_rebuild(self):
|
|
started = datetime(2026, 1, 1, 8)
|
|
self.insert_motion(1, 160, started, 45.5)
|
|
self.insert_motion(161, 10, started + timedelta(minutes=30), 45.6)
|
|
app_sql.refresh_trips(force=True)
|
|
self.assertEqual(
|
|
self.conn.execute("SELECT COUNT(*) FROM trips").fetchone()[0], 2
|
|
)
|
|
|
|
trip_ids = [
|
|
row[0]
|
|
for row in self.conn.execute("SELECT id FROM trips ORDER BY id").fetchall()
|
|
]
|
|
self.assertEqual(app_sql.delete_trips_batch(trip_ids), 2)
|
|
self.assertEqual(
|
|
self.conn.execute("SELECT COUNT(*) FROM trips").fetchone()[0], 0
|
|
)
|
|
self.assertEqual(
|
|
self.conn.execute(
|
|
"SELECT COUNT(*) FROM deleted_trip_markers"
|
|
).fetchone()[0],
|
|
2,
|
|
)
|
|
|
|
# A full rebuild must not resurrect locally-deleted trips.
|
|
app_sql.refresh_trips(force=True)
|
|
self.assertEqual(
|
|
self.conn.execute("SELECT COUNT(*) FROM trips").fetchone()[0], 0
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|