feat: load nodes from json for tests (#8)

* test: verify web query uses populated db

* test: add test script

data/nodes.py

@@ -1,12 +1,17 @@
-import json, sqlite3, time, threading
+import json, os, 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
 
-DB = "nodes.db"
+try:  # meshtastic is optional for tests
+    from meshtastic.serial_interface import SerialInterface
+    from meshtastic.mesh_interface import MeshInterface
+except ModuleNotFoundError:  # pragma: no cover - imported lazily for hardware usage
+    SerialInterface = None  # type: ignore
+    MeshInterface = None  # type: ignore
 
-schema = Path("nodes.sql").read_text()
+DB = os.environ.get("MESH_DB", "nodes.db")
+
+schema = Path(__file__).with_name("nodes.sql").read_text()
 conn = sqlite3.connect(DB, check_same_thread=False)
 conn.executescript(schema)
 conn.commit()
@@ -76,6 +81,14 @@ def upsert_node(node_id, n):
       altitude=excluded.altitude, node_json=excluded.node_json
     """, row)
 
+
+def load_nodes_from_file(path: str | Path):
+    """Populate the database from a nodes.json file."""
+    nodes = json.loads(Path(path).read_text())
+    for node_id, node in nodes.items():
+        upsert_node(node_id, node)
+    conn.commit()
+
 def snapshot_nodes_periodically(iface: MeshInterface, every_sec=30):
     time.sleep(5)  # let the library sync initial node DB
     while True:
@@ -88,6 +101,8 @@ def snapshot_nodes_periodically(iface: MeshInterface, every_sec=30):
         time.sleep(every_sec)
 
 def main():
+    if SerialInterface is None:
+        raise RuntimeError("meshtastic library not installed")
     iface = SerialInterface(devPath="/dev/ttyACM0")
     threading.Thread(target=snapshot_nodes_periodically, args=(iface, 30), daemon=True).start()
     print("Nodes ingestor running. Ctrl+C to stop.")

test/test.sh

@@ -0,0 +1,5 @@
+#!/usr/bin/env bash
+
+python -m venv .venv && source .venv/bin/activate
+pip install pytest
+pytest

test/test_web_app.py

@@ -0,0 +1,43 @@
+import json
+import os
+import subprocess
+import sys
+from pathlib import Path
+
+import pytest
+
+
+def test_query_nodes_from_web_app(tmp_path):
+    db_path = tmp_path / "nodes.db"
+    os.environ["MESH_DB"] = str(db_path)
+
+    # import nodes module after setting env var
+    sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+    from data import nodes  # type: ignore
+
+    nodes.load_nodes_from_file(Path(__file__).with_name("nodes.json"))
+
+    web_dir = Path(__file__).resolve().parents[1] / "web"
+    try:
+        subprocess.run(["bundle", "install"], cwd=web_dir, check=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
+        env = os.environ.copy()
+        env["MESH_DB"] = str(db_path)
+        out = subprocess.check_output(
+            [
+                "bundle",
+                "exec",
+                "ruby",
+                "-e",
+                'require_relative "app"; require "json"; puts query_nodes(5).to_json'
+            ],
+            cwd=web_dir,
+            env=env,
+            stderr=subprocess.DEVNULL,
+        )
+    except subprocess.CalledProcessError:
+        pytest.skip("ruby dependencies not installed")
+
+    data = json.loads(out)
+    assert len(data) == 5
+    last_heards = [item["last_heard"] for item in data]
+    assert last_heards == sorted(last_heards, reverse=True)