adam/Remote-Terminal-for-MeshCore
1
0
Fork 0
mirror of https://github.com/jkingsman/Remote-Terminal-for-MeshCore.git synced 2026-08-09 18:23:20 +02:00
Code Issues Packages Projects Releases Wiki Activity

Rebase migrations, add some tests, API endpoint (whoops said we didn't need that and we did lol)

Browse Source
This commit is contained in:
Jack Kingsman
2026-04-02 12:12:35 -07:00
parent 5e1bdb2cc1
commit a2547d7f62
12 changed files with 631 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_migrations.py
+16 -16
View File
@@ -1249,8 +1249,8 @@ class TestMigration039:
applied = await run_migrations(conn)
assert applied == 11
assert await get_version(conn) == 49
assert applied == 12
assert await get_version(conn) == 50
cursor = await conn.execute(
"""
@@ -1321,8 +1321,8 @@ class TestMigration039:
applied = await run_migrations(conn)
assert applied == 11
assert await get_version(conn) == 49
assert applied == 12
assert await get_version(conn) == 50
cursor = await conn.execute(
"""
@@ -1388,8 +1388,8 @@ class TestMigration039:
applied = await run_migrations(conn)
assert applied == 5
assert await get_version(conn) == 49
assert applied == 6
assert await get_version(conn) == 50
cursor = await conn.execute(
"""
@@ -1441,8 +1441,8 @@ class TestMigration040:
applied = await run_migrations(conn)
assert applied == 10
assert await get_version(conn) == 49
assert applied == 11
assert await get_version(conn) == 50
await conn.execute(
"""
@@ -1503,8 +1503,8 @@ class TestMigration041:
applied = await run_migrations(conn)
assert applied == 9
assert await get_version(conn) == 49
assert applied == 10
assert await get_version(conn) == 50
await conn.execute(
"""
@@ -1556,8 +1556,8 @@ class TestMigration042:
applied = await run_migrations(conn)
assert applied == 8
assert await get_version(conn) == 49
assert applied == 9
assert await get_version(conn) == 50
await conn.execute(
"""
@@ -1696,8 +1696,8 @@ class TestMigration046:
applied = await run_migrations(conn)
assert applied == 4
assert await get_version(conn) == 49
assert applied == 5
assert await get_version(conn) == 50
cursor = await conn.execute(
"""
@@ -1790,8 +1790,8 @@ class TestMigration047:
applied = await run_migrations(conn)
assert applied == 3
assert await get_version(conn) == 49
assert applied == 4
assert await get_version(conn) == 50
cursor = await conn.execute(
"""
tests/test_repeater_telemetry.py
+189
View File
@@ -0,0 +1,189 @@
"""Tests for repeater telemetry history: repository CRUD and read-only endpoint."""
import time
import pytest
from app.models import CONTACT_TYPE_REPEATER
from app.repository import (
ContactRepository,
RepeaterTelemetryRepository,
)
KEY_A = "aa" * 32
KEY_B = "bb" * 32
SAMPLE_STATUS = {
"battery_volts": 4.15,
"tx_queue_len": 0,
"noise_floor_dbm": -100,
"last_rssi_dbm": -80,
"last_snr_db": 5.0,
"packets_received": 100,
"packets_sent": 50,
"airtime_seconds": 300,
"rx_airtime_seconds": 200,
"uptime_seconds": 1000,
"sent_flood": 10,
"sent_direct": 40,
"recv_flood": 60,
"recv_direct": 40,
"flood_dups": 5,
"direct_dups": 2,
"full_events": 0,
}
async def _insert_repeater(public_key: str, name: str = "Repeater"):
"""Insert a repeater contact into the test database."""
await ContactRepository.upsert(
{
"public_key": public_key,
"name": name,
"type": CONTACT_TYPE_REPEATER,
"flags": 0,
"direct_path": None,
"direct_path_len": -1,
"direct_path_hash_mode": -1,
"last_advert": None,
"lat": None,
"lon": None,
"last_seen": None,
"on_radio": False,
"last_contacted": None,
"first_seen": None,
}
)
@pytest.fixture
async def _db(test_db):
"""Set up test DB and patch the repeater_telemetry module's db reference."""
from app.repository import repeater_telemetry
original = repeater_telemetry.db
repeater_telemetry.db = test_db
try:
yield test_db
finally:
repeater_telemetry.db = original
class TestRepeaterTelemetryRepository:
"""Tests for RepeaterTelemetryRepository CRUD operations with JSON blob storage."""
@pytest.mark.asyncio
async def test_record_and_get_history(self, _db):
await _insert_repeater(KEY_A)
now = int(time.time())
await RepeaterTelemetryRepository.record(
public_key=KEY_A,
timestamp=now - 3600,
data={**SAMPLE_STATUS, "battery_volts": 4.15},
)
await RepeaterTelemetryRepository.record(
public_key=KEY_A,
timestamp=now,
data={**SAMPLE_STATUS, "battery_volts": 4.10},
)
history = await RepeaterTelemetryRepository.get_history(KEY_A, now - 7200)
assert len(history) == 2
assert history[0]["data"]["battery_volts"] == 4.15
assert history[1]["data"]["battery_volts"] == 4.10
assert history[0]["timestamp"] < history[1]["timestamp"]
@pytest.mark.asyncio
async def test_get_history_filters_by_time(self, _db):
await _insert_repeater(KEY_A)
now = int(time.time())
await RepeaterTelemetryRepository.record(KEY_A, now - 7200, SAMPLE_STATUS)
await RepeaterTelemetryRepository.record(KEY_A, now - 3600, SAMPLE_STATUS)
await RepeaterTelemetryRepository.record(KEY_A, now, SAMPLE_STATUS)
history = await RepeaterTelemetryRepository.get_history(KEY_A, now - 3601)
assert len(history) == 2
@pytest.mark.asyncio
async def test_get_history_isolates_by_key(self, _db):
await _insert_repeater(KEY_A)
await _insert_repeater(KEY_B)
now = int(time.time())
await RepeaterTelemetryRepository.record(
KEY_A, now, {**SAMPLE_STATUS, "battery_volts": 4.1}
)
await RepeaterTelemetryRepository.record(
KEY_B, now, {**SAMPLE_STATUS, "battery_volts": 3.9}
)
history_a = await RepeaterTelemetryRepository.get_history(KEY_A, 0)
history_b = await RepeaterTelemetryRepository.get_history(KEY_B, 0)
assert len(history_a) == 1
assert len(history_b) == 1
assert history_a[0]["data"]["battery_volts"] == 4.1
@pytest.mark.asyncio
async def test_data_stored_as_json(self, _db):
"""Verify the data column stores valid JSON that round-trips correctly."""
await _insert_repeater(KEY_A)
now = int(time.time())
await RepeaterTelemetryRepository.record(KEY_A, now, SAMPLE_STATUS)
history = await RepeaterTelemetryRepository.get_history(KEY_A, 0)
assert len(history) == 1
assert history[0]["data"] == SAMPLE_STATUS
class TestTelemetryHistoryEndpoint:
"""Tests for the read-only GET telemetry-history endpoint."""
@pytest.mark.asyncio
async def test_returns_history_for_repeater(self, _db, client):
await _insert_repeater(KEY_A)
now = int(time.time())
await RepeaterTelemetryRepository.record(KEY_A, now, SAMPLE_STATUS)
resp = await client.get(f"/api/contacts/{KEY_A}/repeater/telemetry-history")
assert resp.status_code == 200
data = resp.json()
assert len(data) == 1
assert data[0]["data"]["battery_volts"] == 4.15
@pytest.mark.asyncio
async def test_returns_empty_list_when_no_history(self, _db, client):
await _insert_repeater(KEY_A)
resp = await client.get(f"/api/contacts/{KEY_A}/repeater/telemetry-history")
assert resp.status_code == 200
assert resp.json() == []
@pytest.mark.asyncio
async def test_rejects_non_repeater(self, _db, client):
await ContactRepository.upsert(
{
"public_key": KEY_A,
"name": "Node",
"type": 0,
"flags": 0,
"direct_path": None,
"direct_path_len": -1,
"direct_path_hash_mode": -1,
"last_advert": None,
"lat": None,
"lon": None,
"last_seen": None,
"on_radio": False,
"last_contacted": None,
"first_seen": None,
}
)
resp = await client.get(f"/api/contacts/{KEY_A}/repeater/telemetry-history")
assert resp.status_code == 400
@pytest.mark.asyncio
async def test_returns_404_for_unknown_contact(self, _db, client):
resp = await client.get(f"/api/contacts/{KEY_A}/repeater/telemetry-history")
assert resp.status_code == 404