generated from Grigo/AndroidTemplate
263 lines
7.9 KiB
Python
263 lines
7.9 KiB
Python
import sqlite3
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from core import storage
|
|
from core.models import TelemetryIn
|
|
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
|
|
|
|
|
|
def test_reference_point_session_stats(temp_db):
|
|
storage.init_db()
|
|
start = storage.start_reference_point_session(
|
|
"android-12345678", "yard", 55.75, 37.62
|
|
)
|
|
sid = start["session_id"]
|
|
storage.record_telemetry(TelemetryIn(
|
|
device_id="android-11111111",
|
|
lat=55.751,
|
|
lon=37.621,
|
|
role="TX",
|
|
ts=start["started_at"],
|
|
meta='{"role":"TX","frequency_hz":433000000,"power_dbm":17,'
|
|
'"spreading_factor":7,"bandwidth_khz":125,"code_rate":"4/5"}',
|
|
))
|
|
storage.record_telemetry(TelemetryIn(
|
|
device_id="android-12345678",
|
|
lat=55.75,
|
|
lon=37.62,
|
|
role="RX",
|
|
ts=start["started_at"],
|
|
meta='{"role":"RX","rx_quality_percent":90}',
|
|
))
|
|
added = storage.add_reference_point_samples(
|
|
sid,
|
|
[
|
|
{
|
|
"ts": 1.0,
|
|
"lat": 55.75,
|
|
"lon": 37.62,
|
|
"role": "RX",
|
|
"meta": {"per_percent": 10, "rx_quality_percent": 80},
|
|
},
|
|
{
|
|
"ts": 2.0,
|
|
"lat": 55.75,
|
|
"lon": 37.62,
|
|
"role": "RX",
|
|
"meta": {"per_percent": 20, "rx_quality_percent": 90},
|
|
},
|
|
{
|
|
"ts": 3.0,
|
|
"lat": 55.75,
|
|
"lon": 37.62,
|
|
"role": "RX",
|
|
"meta": {"per_percent": 30, "rx_quality_percent": 100},
|
|
},
|
|
],
|
|
)
|
|
assert added["added"] == 3
|
|
storage.finish_reference_point_session(sid)
|
|
|
|
session = storage.get_reference_point_session(sid)
|
|
assert session["sample_count"] == 3
|
|
assert session["stats"]["per_percent"]["p50"] == 20
|
|
assert session["stats"]["per_percent"]["p75"] == 25
|
|
assert session["stats"]["rx_quality_percent"]["p90"] == 98
|
|
assert session["reference_context"]["tx_params"]["frequency_hz"] == 433000000
|
|
assert session["reference_context"]["tx_params"]["spreading_factor"] == 7
|
|
assert session["reference_context"]["rx_position"]["device_id"] == "android-12345678"
|
|
assert session["reference_context"]["rx_position"]["lat"] == 55.75
|
|
assert session["reference_context"]["tx_position"]["device_id"] == "android-11111111"
|
|
assert session["reference_context"]["tx_position"]["lat"] == 55.751
|
|
assert session["reference_context"]["distance_m"] is not None
|
|
assert session["reference_context"]["distance_m"] > 0
|
|
|
|
|
|
def test_reference_point_session_gfsk_tx_params(temp_db):
|
|
storage.init_db()
|
|
start = storage.start_reference_point_session(
|
|
"android-12345678", "gfsk-yard", 55.75, 37.62
|
|
)
|
|
sid = start["session_id"]
|
|
storage.record_telemetry(TelemetryIn(
|
|
device_id="android-11111111",
|
|
lat=55.751,
|
|
lon=37.621,
|
|
role="TX",
|
|
ts=start["started_at"],
|
|
meta='{"role":"TX","modem_mode":"GFSK","frequency_hz":467712500,'
|
|
'"power_dbm":-9,"bitrate_bps":20000,"fdev_hz":10000,'
|
|
'"modulation_index":1.0,"rx_bandwidth_hz":46900,"sync_word_length_bits":32}',
|
|
))
|
|
storage.add_reference_point_samples(
|
|
sid,
|
|
[{"ts": 1.0, "lat": 55.75, "lon": 37.62, "role": "RX", "meta": {"per_percent": 5}}],
|
|
)
|
|
storage.finish_reference_point_session(sid)
|
|
|
|
session = storage.get_reference_point_session(sid)
|
|
tx_params = session["reference_context"]["tx_params"]
|
|
assert tx_params["modem_mode"] == "GFSK"
|
|
assert tx_params["bitrate_bps"] == 20000
|
|
assert tx_params["fdev_hz"] == 10000
|
|
assert tx_params["modulation_index"] == 1.0
|
|
assert tx_params["rx_bandwidth_hz"] == 46900
|
|
assert tx_params["sync_word_length_bits"] == 32
|
|
|
|
|
|
def test_macro_presets_crud(temp_db):
|
|
storage.init_db()
|
|
result = storage.upsert_macro_preset(
|
|
"SF7 Test",
|
|
"SF7 Test",
|
|
"baseline",
|
|
params={"sf": 7, "bw_khz": "125"},
|
|
lines=["S", "AT+SF=7"],
|
|
device_role="RX",
|
|
)
|
|
preset = result["preset"]
|
|
assert preset["id"] == "sf7-test"
|
|
assert preset["params"]["sf"] == 7
|
|
|
|
listed = storage.list_macro_presets()
|
|
assert len(listed) == 1
|
|
assert listed[0]["lines"] == ["S", "AT+SF=7"]
|
|
|
|
storage.delete_macro_preset("sf7-test")
|
|
assert storage.list_macro_presets() == []
|
|
|
|
|
|
def test_macro_preset_gfsk_params(temp_db):
|
|
storage.init_db()
|
|
result = storage.upsert_macro_preset(
|
|
"GFSK RX",
|
|
"GFSK RX",
|
|
"gfsk baseline",
|
|
params={
|
|
"modem_mode": "GFSK",
|
|
"frequency_hz": 467712500,
|
|
"power_dbm": -9,
|
|
"bitrate_bps": 20000,
|
|
"fdev_hz": 10000,
|
|
"preamble": 48,
|
|
"send_timeout_ms": 0,
|
|
"role": "RX",
|
|
},
|
|
lines=[
|
|
"S", "AT+MD=GFSK", "AT+FQ=467712500", "AT+PW=-9",
|
|
"AT+BR=20000", "AT+FD=10000", "AT+PL=48", "AT+TM=0", "AT+RX",
|
|
],
|
|
device_role="RX",
|
|
)
|
|
preset = result["preset"]
|
|
assert preset["params"]["modem_mode"] == "GFSK"
|
|
assert preset["params"]["bitrate_bps"] == 20000
|
|
assert preset["lines"][1] == "AT+MD=GFSK"
|