generated from Grigo/AndroidTemplate
113 lines
2.9 KiB
Python
113 lines
2.9 KiB
Python
import sqlite3
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from core import storage
|
|
from core.schema import SCHEMA_VERSION, apply_migrations, check_db_ok, column_exists
|
|
|
|
|
|
@pytest.fixture
|
|
def temp_db(monkeypatch):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
path = str(Path(tmp) / "test.db")
|
|
monkeypatch.setattr(storage, "DATABASE_PATH", path)
|
|
yield path
|
|
|
|
|
|
def test_fresh_db_has_all_tables(temp_db):
|
|
storage.init_db()
|
|
status = storage.db_status()
|
|
assert status["db_ok"] is True
|
|
assert status["schema_version"] == SCHEMA_VERSION
|
|
|
|
|
|
def test_old_telemetry_without_meta_gets_migrated(temp_db):
|
|
conn = sqlite3.connect(temp_db)
|
|
conn.execute(
|
|
"""
|
|
CREATE TABLE telemetry (
|
|
id INTEGER PRIMARY KEY,
|
|
device_id TEXT,
|
|
lat REAL, lon REAL, rssi REAL, range_m REAL,
|
|
raw_frame TEXT, ts REAL
|
|
)
|
|
"""
|
|
)
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
storage.init_db()
|
|
conn = sqlite3.connect(temp_db)
|
|
assert column_exists(conn, "telemetry", "meta")
|
|
assert check_db_ok(conn)
|
|
conn.close()
|
|
|
|
|
|
def test_sync_track_offline_upload(temp_db, monkeypatch):
|
|
storage.init_db()
|
|
monkeypatch.setattr(storage, "fetch_elevation_m", lambda lat, lon: 100.0)
|
|
|
|
start = storage.start_track("android-12345678")
|
|
tid = start["track_id"]
|
|
|
|
result = storage.sync_track(
|
|
"android-12345678",
|
|
[
|
|
{"ts": 1.0, "lat": 55.75, "lon": 37.62, "role": "TX"},
|
|
{"ts": 2.0, "lat": 55.751, "lon": 37.621, "role": "TX"},
|
|
],
|
|
track_id=tid,
|
|
finish=True,
|
|
)
|
|
assert result["added"] == 2
|
|
assert result["finished"] is True
|
|
assert result["point_count"] == 2
|
|
|
|
track = storage.get_track(tid)
|
|
assert len(track["points"]) == 2
|
|
assert track["ended_at"] is not None
|
|
|
|
|
|
def test_sync_track_create_offline(temp_db, monkeypatch):
|
|
storage.init_db()
|
|
monkeypatch.setattr(storage, "fetch_elevation_m", lambda lat, lon: 50.0)
|
|
|
|
result = storage.sync_track(
|
|
"android-abcdef01",
|
|
[
|
|
{"ts": 10.0, "lat": 59.93, "lon": 30.33, "role": "RX"},
|
|
],
|
|
track_id=None,
|
|
started_at=10.0,
|
|
finish=True,
|
|
)
|
|
assert result["track_id"] > 0
|
|
assert result["point_count"] == 1
|
|
track = storage.get_track(result["track_id"])
|
|
assert track["started_at"] == 10.0
|
|
|
|
|
|
def test_tracks_crud(temp_db):
|
|
storage.init_db()
|
|
start = storage.start_track("android-12345678")
|
|
tid = start["track_id"]
|
|
storage.add_track_points(
|
|
tid,
|
|
[
|
|
{
|
|
"ts": 1.0,
|
|
"lat": 55.75,
|
|
"lon": 37.62,
|
|
"rssi": -70.0,
|
|
"role": "RX",
|
|
"meta": '{"packet":1}',
|
|
}
|
|
],
|
|
)
|
|
fin = storage.finish_track(tid)
|
|
assert fin["point_count"] == 1
|
|
track = storage.get_track(tid)
|
|
assert len(track["points"]) == 1
|