generated from Grigo/AndroidTemplate
69 lines
1.6 KiB
Python
69 lines
1.6 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_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
|