Mathieu Moran
81c7ee25fd
- test_app_trip + test_trip_sql: mock was_car_ever_connected() so trip building runs (BUG-10 guard suppresses trips until car ECU reached) - test_app_settings: include odometer_baseline in valid payload + assert save - test_yaris_uplink: Hotspot -> Hotspot-local; restore_fallback no longer activates AP (reports no_connection); wait timeout writes fallback_restored - test_app_network: local_hotspot mode removed by WiFi swap -> wifi/no-internet
82 lines
2.6 KiB
Python
82 lines
2.6 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])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|