LoraMapTester/server/tests/test_schema.py

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_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() == []