test: cover Position serialization in node snapshot (#6)

data/nodes.py

@@ -1,4 +1,5 @@
 import json, sqlite3, time, threading
+from dataclasses import asdict, is_dataclass
 from pathlib import Path
 from meshtastic.serial_interface import SerialInterface
 from meshtastic.mesh_interface import MeshInterface
@@ -10,35 +11,55 @@ conn = sqlite3.connect(DB, check_same_thread=False)
 conn.executescript(schema)
 conn.commit()
 
+def _get(obj, key, default=None):
+    """Return value for key/attribute from dicts or objects."""
+    if isinstance(obj, dict):
+        return obj.get(key, default)
+    return getattr(obj, key, default)
+
+
+def _jsonable(obj):
+    """Recursively convert dataclasses and objects into JSON-serialisable types."""
+    if is_dataclass(obj):
+        return _jsonable(asdict(obj))
+    if isinstance(obj, dict):
+        return {k: _jsonable(v) for k, v in obj.items()}
+    if isinstance(obj, (list, tuple)):
+        return [_jsonable(v) for v in obj]
+    if hasattr(obj, "__dict__"):
+        return _jsonable(vars(obj))
+    return obj
+
+
 def upsert_node(node_id, n):
-    user = (n.get("user") or {})
-    met  = (n.get("deviceMetrics") or {})
-    pos  = (n.get("position") or {})
+    user = _get(n, "user") or {}
+    met = _get(n, "deviceMetrics") or {}
+    pos = _get(n, "position") or {}
     row = (
         node_id,
-        n.get("num"),
-        user.get("shortName"),
-        user.get("longName"),
-        user.get("macaddr"),
-        user.get("hwModel") or n.get("hwModel"),
-        user.get("role"),
-        user.get("publicKey"),
-        user.get("isUnmessagable"),
-        n.get("isFavorite"),
-        n.get("hopsAway"),
-        n.get("snr"),
-        n.get("lastHeard"),
-        met.get("batteryLevel"),
-        met.get("voltage"),
-        met.get("channelUtilization"),
-        met.get("airUtilTx"),
-        met.get("uptimeSeconds"),
-        pos.get("time"),
-        pos.get("locationSource"),
-        pos.get("latitude"),
-        pos.get("longitude"),
-        pos.get("altitude"),
-        json.dumps(n, ensure_ascii=False)
+        _get(n, "num"),
+        _get(user, "shortName"),
+        _get(user, "longName"),
+        _get(user, "macaddr"),
+        _get(user, "hwModel") or _get(n, "hwModel"),
+        _get(user, "role"),
+        _get(user, "publicKey"),
+        _get(user, "isUnmessagable"),
+        _get(n, "isFavorite"),
+        _get(n, "hopsAway"),
+        _get(n, "snr"),
+        _get(n, "lastHeard"),
+        _get(met, "batteryLevel"),
+        _get(met, "voltage"),
+        _get(met, "channelUtilization"),
+        _get(met, "airUtilTx"),
+        _get(met, "uptimeSeconds"),
+        _get(pos, "time"),
+        _get(pos, "locationSource"),
+        _get(pos, "latitude"),
+        _get(pos, "longitude"),
+        _get(pos, "altitude"),
+        json.dumps(_jsonable(n), ensure_ascii=False),
     )
     conn.execute("""
     INSERT INTO nodes(node_id,num,short_name,long_name,macaddr,hw_model,role,public_key,is_unmessagable,is_favorite,

test/test_nodes_serialization.py

@@ -0,0 +1,56 @@
+import json
+import os
+import sqlite3
+import sys
+import types
+from dataclasses import dataclass
+from pathlib import Path
+
+
+def test_upsert_node_handles_position(tmp_path):
+    data_dir = Path(__file__).resolve().parent.parent / "data"
+    cwd = os.getcwd()
+    os.chdir(data_dir)
+    try:
+        # Provide minimal stubs for the meshtastic modules imported by nodes.py
+        meshtastic = types.ModuleType("meshtastic")
+        serial_module = types.ModuleType("serial_interface")
+        serial_module.SerialInterface = object
+        mesh_module = types.ModuleType("mesh_interface")
+        mesh_module.MeshInterface = object
+        meshtastic.serial_interface = serial_module
+        meshtastic.mesh_interface = mesh_module
+        sys.modules.setdefault("meshtastic", meshtastic)
+        sys.modules.setdefault("meshtastic.serial_interface", serial_module)
+        sys.modules.setdefault("meshtastic.mesh_interface", mesh_module)
+
+        sys.path.insert(0, str(data_dir))
+        import nodes
+        # Close original on-disk connection and use in-memory DB for testing
+        nodes.conn.close()
+        dbfile = Path("nodes.db")
+        if dbfile.exists():
+            dbfile.unlink()
+        nodes.conn = sqlite3.connect(":memory:", check_same_thread=False)
+        nodes.conn.executescript(nodes.schema)
+        nodes.conn.commit()
+
+        @dataclass
+        class Position:
+            time: int = 123
+            locationSource: str = "GPS"
+            latitude: float = 52.5
+            longitude: float = 13.4
+            altitude: float = 34.0
+
+        n = {"num": 7, "position": Position()}
+        nodes.upsert_node("node1", n)
+        nodes.conn.commit()
+        row = nodes.conn.execute(
+            "SELECT node_json FROM nodes WHERE node_id=?", ("node1",)
+        ).fetchone()
+        assert row is not None
+        data = json.loads(row[0])
+        assert data["position"]["latitude"] == 52.5
+    finally:
+        os.chdir(cwd)