adam/Remote-Terminal-for-MeshCore
1
0
Fork 0
mirror of https://github.com/jkingsman/Remote-Terminal-for-MeshCore.git synced 2026-07-31 22:12:34 +02:00
Code Issues Packages Projects Releases Wiki Activity

Drop out crappy tests, and improve quality overall

Browse Source
This commit is contained in:
Jack Kingsman
2026-02-23 22:28:09 -08:00
parent 31bb1e7d22
commit ecb748b9e3
17 changed files with 226 additions and 1246 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_bot.py
+18 -11
View File
@@ -430,20 +430,27 @@ class TestRunBotForMessage:
"""Bot is triggered for outgoing messages (user can trigger their own bots)."""
with patch("app.repository.AppSettingsRepository") as mock_repo:
mock_settings = MagicMock()
mock_settings.bots = [] # No enabled bots, but settings ARE checked
mock_settings.bots = [
BotConfig(id="1", name="Echo", enabled=True, code="def bot(**k): return 'echo'")
]
mock_repo.get = AsyncMock(return_value=mock_settings)
await run_bot_for_message(
sender_name="Me",
sender_key="abc123",
message_text="Hello",
is_dm=True,
channel_key=None,
is_outgoing=True,
)
with (
patch("app.bot.asyncio.sleep", new_callable=AsyncMock),
patch("app.bot.execute_bot_code", return_value="echo") as mock_exec,
patch("app.bot.process_bot_response", new_callable=AsyncMock),
):
await run_bot_for_message(
sender_name="Me",
sender_key="abc123" + "0" * 58,
message_text="Hello",
is_dm=True,
channel_key=None,
is_outgoing=True,
)
# Should check settings (outgoing no longer skipped)
mock_repo.get.assert_called_once()
# Bot should actually execute for outgoing messages
mock_exec.assert_called_once()
@pytest.mark.asyncio
async def test_skips_when_no_enabled_bots(self):
tests/test_decoder.py
+21 -17
View File
@@ -33,7 +33,10 @@ class TestChannelKeyDerivation:
channel_name = "#test"
expected_key = hashlib.sha256(channel_name.encode("utf-8")).digest()[:16]
# This matches the meshcore_py implementation
# Verify the derived key produces the expected channel hash
result_hash = calculate_channel_hash(expected_key)
expected_hash = format(hashlib.sha256(expected_key).digest()[0], "02x")
assert result_hash == expected_hash
assert len(expected_key) == 16
def test_channel_hash_calculation(self):
@@ -342,7 +345,7 @@ class TestAdvertisementParsing:
def test_parse_advertisement_extracts_public_key(self):
"""Advertisement parsing extracts the public key correctly."""
from app.decoder import PayloadType, parse_packet
from app.decoder import parse_advertisement, parse_packet
packet_hex = (
"1100AE92564C5C9884854F04F469BBB2BAB8871A078053AF6CF4AA2C014B18CE8A83"
@@ -352,21 +355,27 @@ class TestAdvertisementParsing:
)
packet = bytes.fromhex(packet_hex)
# Verify packet is recognized as ADVERT type
info = parse_packet(packet)
assert info is not None
assert info.payload_type == PayloadType.ADVERT
result = parse_advertisement(info.payload)
assert result is not None
assert (
result.public_key == "ae92564c5c9884854f04f469bbb2bab8871a078053af6cf4aa2c014b18ce8a83"
)
def test_non_advertisement_returns_none(self):
"""Non-advertisement packets return None when parsed as advertisement."""
from app.decoder import PayloadType, parse_packet
"""Non-advertisement payload returns None when parsed as advertisement."""
from app.decoder import parse_advertisement, parse_packet
# GROUP_TEXT packet, not an advertisement
packet = bytes([0x15, 0x00]) + bytes(50)
info = parse_packet(packet)
assert info is not None
assert info.payload_type != PayloadType.ADVERT
result = parse_advertisement(info.payload)
assert result is None
class TestScalarClamping:
@@ -484,18 +493,13 @@ class TestSharedSecretDerivation:
def test_derive_shared_secret_different_keys_different_result(self):
"""Different key pairs produce different shared secrets."""
other_pub = bytes.fromhex(
"1234567890abcdef1234567890abcdef1234567890abcdef1234567890abcdef"
)
# Use the real FACE12 public key as a second peer key (valid curve point)
face12_pub = derive_public_key(self.FACE12_PRIV)
result1 = derive_shared_secret(self.FACE12_PRIV, self.A1B2C3_PUB)
# This may raise an exception for invalid public key, which is also acceptable
try:
result2 = derive_shared_secret(self.FACE12_PRIV, other_pub)
assert result1 != result2
except Exception:
# Invalid public keys may fail, which is fine
pass
result2 = derive_shared_secret(self.FACE12_PRIV, face12_pub)
assert result1 != result2
class TestDirectMessageDecryption:
tests/test_event_handlers.py
-50
View File
@@ -676,53 +676,3 @@ class TestOnNewContact:
contacts = await ContactRepository.get_all()
assert len(contacts) == 0
@pytest.mark.asyncio
async def test_sets_on_radio_true(self, test_db):
"""Contact data passed to upsert has on_radio=True."""
from app.event_handlers import on_new_contact
with (
patch("app.event_handlers.broadcast_event"),
patch("app.event_handlers.time") as mock_time,
):
mock_time.time.return_value = 1700000000
class MockEvent:
payload = {
"public_key": "dd" * 32,
"adv_name": "Delta",
"type": 0,
"flags": 0,
}
await on_new_contact(MockEvent())
contact = await ContactRepository.get_by_key("dd" * 32)
assert contact is not None
assert contact.on_radio is True
@pytest.mark.asyncio
async def test_sets_last_seen_to_current_timestamp(self, test_db):
"""Contact data includes last_seen set to current time."""
from app.event_handlers import on_new_contact
with (
patch("app.event_handlers.broadcast_event"),
patch("app.event_handlers.time") as mock_time,
):
mock_time.time.return_value = 1700099999
class MockEvent:
payload = {
"public_key": "ee" * 32,
"adv_name": "Echo",
"type": 0,
"flags": 0,
}
await on_new_contact(MockEvent())
contact = await ContactRepository.get_by_key("ee" * 32)
assert contact is not None
assert contact.last_seen == 1700099999
tests/test_radio_sync.py
-28
View File
@@ -130,17 +130,6 @@ class TestPollingPause:
# Now unpaused - all contexts exited
assert not is_polling_paused()
@pytest.mark.asyncio
async def test_triple_nested_pause(self):
"""Three levels of nesting work correctly."""
async with pause_polling():
async with pause_polling():
async with pause_polling():
assert is_polling_paused()
assert is_polling_paused()
assert is_polling_paused()
assert not is_polling_paused()
@pytest.mark.asyncio
async def test_pause_resumes_on_exception(self):
"""Polling resumes even if exception occurs in context."""
@@ -171,23 +160,6 @@ class TestPollingPause:
# All contexts exited
assert not is_polling_paused()
@pytest.mark.asyncio
async def test_counter_increments_and_decrements(self):
"""Counter correctly tracks pause depth."""
import app.radio_sync as radio_sync
assert radio_sync._polling_pause_count == 0
async with pause_polling():
assert radio_sync._polling_pause_count == 1
async with pause_polling():
assert radio_sync._polling_pause_count == 2
assert radio_sync._polling_pause_count == 1
assert radio_sync._polling_pause_count == 0
class TestSyncRadioTime:
"""Test the radio time sync function."""
tests/test_repository.py
+113 -189
View File
@@ -1,6 +1,5 @@
"""Tests for repository layer."""
import json
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
@@ -122,153 +121,98 @@ class TestMessageRepositoryAddPath:
class TestMessageRepositoryGetByContent:
"""Test MessageRepository.get_by_content method."""
"""Test MessageRepository.get_by_content against a real SQLite database."""
@pytest.mark.asyncio
async def test_get_by_content_finds_matching_message(self):
async def test_get_by_content_finds_matching_message(self, test_db):
"""Returns message when all content fields match."""
mock_conn = AsyncMock()
mock_cursor = AsyncMock()
mock_cursor.fetchone = AsyncMock(
return_value={
"id": 42,
"type": "CHAN",
"conversation_key": "ABCD1234",
"text": "Hello world",
"sender_timestamp": 1700000000,
"received_at": 1700000001,
"paths": None,
"txt_type": 0,
"signature": None,
"outgoing": 0,
"acked": 1,
}
msg_id = await _create_message(
test_db,
msg_type="CHAN",
conversation_key="ABCD1234ABCD1234ABCD1234ABCD1234",
text="Hello world",
sender_timestamp=1700000000,
)
mock_conn.execute = AsyncMock(return_value=mock_cursor)
mock_db = MagicMock()
mock_db.conn = mock_conn
with patch("app.repository.db", mock_db):
from app.repository import MessageRepository
result = await MessageRepository.get_by_content(
msg_type="CHAN",
conversation_key="ABCD1234",
text="Hello world",
sender_timestamp=1700000000,
)
result = await MessageRepository.get_by_content(
msg_type="CHAN",
conversation_key="ABCD1234ABCD1234ABCD1234ABCD1234",
text="Hello world",
sender_timestamp=1700000000,
)
assert result is not None
assert result.id == 42
assert result.id == msg_id
assert result.type == "CHAN"
assert result.conversation_key == "ABCD1234"
assert result.text == "Hello world"
assert result.acked == 1
@pytest.mark.asyncio
async def test_get_by_content_returns_none_when_not_found(self):
async def test_get_by_content_returns_none_when_not_found(self, test_db):
"""Returns None when no message matches."""
mock_conn = AsyncMock()
mock_cursor = AsyncMock()
mock_cursor.fetchone = AsyncMock(return_value=None)
mock_conn.execute = AsyncMock(return_value=mock_cursor)
await _create_message(test_db, text="Existing message")
mock_db = MagicMock()
mock_db.conn = mock_conn
with patch("app.repository.db", mock_db):
from app.repository import MessageRepository
result = await MessageRepository.get_by_content(
msg_type="CHAN",
conversation_key="NONEXISTENT",
text="Not found",
sender_timestamp=1700000000,
)
result = await MessageRepository.get_by_content(
msg_type="CHAN",
conversation_key="AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA0",
text="Not found",
sender_timestamp=1700000000,
)
assert result is None
@pytest.mark.asyncio
async def test_get_by_content_handles_null_sender_timestamp(self):
async def test_get_by_content_handles_null_sender_timestamp(self, test_db):
"""Handles messages with NULL sender_timestamp correctly."""
mock_conn = AsyncMock()
mock_cursor = AsyncMock()
mock_cursor.fetchone = AsyncMock(
return_value={
"id": 43,
"type": "PRIV",
"conversation_key": "abc123",
"text": "Test message",
"sender_timestamp": None,
"received_at": 1700000001,
"paths": None,
"txt_type": 0,
"signature": None,
"outgoing": 1,
"acked": 0,
}
msg_id = await _create_message(
test_db,
msg_type="PRIV",
conversation_key="abc123abc123abc123abc123abc12300",
text="Null timestamp msg",
sender_timestamp=None,
outgoing=True,
)
mock_conn.execute = AsyncMock(return_value=mock_cursor)
mock_db = MagicMock()
mock_db.conn = mock_conn
with patch("app.repository.db", mock_db):
from app.repository import MessageRepository
result = await MessageRepository.get_by_content(
msg_type="PRIV",
conversation_key="abc123",
text="Test message",
sender_timestamp=None,
)
result = await MessageRepository.get_by_content(
msg_type="PRIV",
conversation_key="abc123abc123abc123abc123abc12300",
text="Null timestamp msg",
sender_timestamp=None,
)
assert result is not None
assert result.id == msg_id
assert result.sender_timestamp is None
assert result.outgoing is True
@pytest.mark.asyncio
async def test_get_by_content_parses_paths_correctly(self):
"""Parses paths JSON into MessagePath objects."""
paths_json = json.dumps(
[
{"path": "1A2B", "received_at": 1700000000},
{"path": "3C4D", "received_at": 1700000001},
]
async def test_get_by_content_distinguishes_by_timestamp(self, test_db):
"""Different sender_timestamps are distinguished correctly."""
await _create_message(test_db, text="Same text", sender_timestamp=1700000000)
msg_id2 = await _create_message(test_db, text="Same text", sender_timestamp=1700000001)
result = await MessageRepository.get_by_content(
msg_type="CHAN",
conversation_key="AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA0",
text="Same text",
sender_timestamp=1700000001,
)
mock_conn = AsyncMock()
mock_cursor = AsyncMock()
mock_cursor.fetchone = AsyncMock(
return_value={
"id": 44,
"type": "CHAN",
"conversation_key": "ABCD1234",
"text": "Multi-path message",
"sender_timestamp": 1700000000,
"received_at": 1700000000,
"paths": paths_json,
"txt_type": 0,
"signature": None,
"outgoing": 0,
"acked": 2,
}
assert result is not None
assert result.id == msg_id2
@pytest.mark.asyncio
async def test_get_by_content_with_paths(self, test_db):
"""Returns message with paths correctly parsed."""
msg_id = await _create_message(test_db, text="Multi-path message")
await MessageRepository.add_path(msg_id, "1A2B", received_at=1700000000)
await MessageRepository.add_path(msg_id, "3C4D", received_at=1700000001)
result = await MessageRepository.get_by_content(
msg_type="CHAN",
conversation_key="AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA0",
text="Multi-path message",
sender_timestamp=1700000000,
)
mock_conn.execute = AsyncMock(return_value=mock_cursor)
mock_db = MagicMock()
mock_db.conn = mock_conn
with patch("app.repository.db", mock_db):
from app.repository import MessageRepository
result = await MessageRepository.get_by_content(
msg_type="CHAN",
conversation_key="ABCD1234",
text="Multi-path message",
sender_timestamp=1700000000,
)
assert result is not None
assert result.paths is not None
@@ -277,99 +221,79 @@ class TestMessageRepositoryGetByContent:
assert result.paths[1].path == "3C4D"
@pytest.mark.asyncio
async def test_get_by_content_handles_corrupted_paths_json(self):
"""Handles corrupted paths JSON gracefully."""
mock_conn = AsyncMock()
mock_cursor = AsyncMock()
mock_cursor.fetchone = AsyncMock(
return_value={
"id": 45,
"type": "CHAN",
"conversation_key": "ABCD1234",
"text": "Corrupted paths",
"sender_timestamp": 1700000000,
"received_at": 1700000000,
"paths": "not valid json {",
"txt_type": 0,
"signature": None,
"outgoing": 0,
"acked": 0,
}
async def test_get_by_content_recovers_from_corrupted_paths_json(self, test_db):
"""Malformed JSON in paths column returns message with paths=None."""
msg_id = await _create_message(test_db, text="Corrupted paths")
# Inject malformed JSON directly into the paths column
await test_db.conn.execute(
"UPDATE messages SET paths = ? WHERE id = ?",
("not valid json{{{", msg_id),
)
mock_conn.execute = AsyncMock(return_value=mock_cursor)
await test_db.conn.commit()
mock_db = MagicMock()
mock_db.conn = mock_conn
result = await MessageRepository.get_by_content(
msg_type="CHAN",
conversation_key="AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA0",
text="Corrupted paths",
sender_timestamp=1700000000,
)
with patch("app.repository.db", mock_db):
from app.repository import MessageRepository
result = await MessageRepository.get_by_content(
msg_type="CHAN",
conversation_key="ABCD1234",
text="Corrupted paths",
sender_timestamp=1700000000,
)
# Should return message with paths=None instead of raising
assert result is not None
assert result.id == msg_id
assert result.paths is None
@pytest.mark.asyncio
async def test_get_by_content_recovers_from_paths_missing_keys(self, test_db):
"""Valid JSON but missing expected keys returns message with paths=None."""
msg_id = await _create_message(test_db, text="Bad keys")
# Valid JSON but missing "path" / "received_at" keys
await test_db.conn.execute(
"UPDATE messages SET paths = ? WHERE id = ?",
('[{"wrong_key": "value"}]', msg_id),
)
await test_db.conn.commit()
result = await MessageRepository.get_by_content(
msg_type="CHAN",
conversation_key="AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA0",
text="Bad keys",
sender_timestamp=1700000000,
)
assert result is not None
assert result.id == msg_id
assert result.paths is None
class TestMessageRepositoryGetAckCount:
"""Test MessageRepository.get_ack_count method."""
"""Test MessageRepository.get_ack_count against a real SQLite database."""
@pytest.mark.asyncio
async def test_get_ack_count_returns_count(self):
async def test_get_ack_count_returns_count(self, test_db):
"""Returns ack count for existing message."""
mock_conn = AsyncMock()
mock_cursor = AsyncMock()
mock_cursor.fetchone = AsyncMock(return_value={"acked": 3})
mock_conn.execute = AsyncMock(return_value=mock_cursor)
msg_id = await _create_message(test_db)
# Simulate acking by directly updating
await test_db.conn.execute("UPDATE messages SET acked = ? WHERE id = ?", (3, msg_id))
mock_db = MagicMock()
mock_db.conn = mock_conn
with patch("app.repository.db", mock_db):
from app.repository import MessageRepository
result = await MessageRepository.get_ack_count(message_id=42)
result = await MessageRepository.get_ack_count(message_id=msg_id)
assert result == 3
@pytest.mark.asyncio
async def test_get_ack_count_returns_zero_for_nonexistent(self):
async def test_get_ack_count_returns_zero_for_nonexistent(self, test_db):
"""Returns 0 for nonexistent message."""
mock_conn = AsyncMock()
mock_cursor = AsyncMock()
mock_cursor.fetchone = AsyncMock(return_value=None)
mock_conn.execute = AsyncMock(return_value=mock_cursor)
mock_db = MagicMock()
mock_db.conn = mock_conn
with patch("app.repository.db", mock_db):
from app.repository import MessageRepository
result = await MessageRepository.get_ack_count(message_id=999)
result = await MessageRepository.get_ack_count(message_id=999999)
assert result == 0
@pytest.mark.asyncio
async def test_get_ack_count_returns_zero_for_unacked(self):
async def test_get_ack_count_returns_zero_for_unacked(self, test_db):
"""Returns 0 for message with no acks."""
mock_conn = AsyncMock()
mock_cursor = AsyncMock()
mock_cursor.fetchone = AsyncMock(return_value={"acked": 0})
mock_conn.execute = AsyncMock(return_value=mock_cursor)
msg_id = await _create_message(test_db)
mock_db = MagicMock()
mock_db.conn = mock_conn
with patch("app.repository.db", mock_db):
from app.repository import MessageRepository
result = await MessageRepository.get_ack_count(message_id=42)
result = await MessageRepository.get_ack_count(message_id=msg_id)
assert result == 0
tests/test_websocket_route.py
-55
View File
@@ -120,34 +120,6 @@ class TestWebSocketEndpoint:
assert pong == {"type": "pong"}
def test_multiple_pings_return_multiple_pongs(self):
"""Each ping gets its own pong response."""
with (
patch("app.routers.ws.radio_manager") as mock_ws_rm,
patch("app.routers.health.radio_manager") as mock_health_rm,
patch("app.routers.health.RawPacketRepository") as mock_repo,
patch("app.routers.health.settings") as mock_settings,
patch("app.routers.health.os.path.getsize", return_value=0),
):
mock_ws_rm.is_connected = True
mock_ws_rm.connection_info = "Serial: /dev/ttyUSB0"
mock_health_rm.is_connected = True
mock_health_rm.connection_info = "Serial: /dev/ttyUSB0"
mock_repo.get_oldest_undecrypted = AsyncMock(return_value=None)
mock_settings.database_path = "/tmp/test.db"
from app.main import app
client = TestClient(app)
with client.websocket_connect("/api/ws") as ws:
ws.receive_json() # consume health
for _ in range(3):
ws.send_text("ping")
pong = ws.receive_json()
assert pong == {"type": "pong"}
def test_non_ping_message_does_not_produce_response(self):
"""Messages other than 'ping' are silently ignored (no response sent)."""
with (
@@ -204,30 +176,3 @@ class TestWebSocketEndpoint:
# After context manager exits, the WebSocket is closed
assert len(ws_manager.active_connections) == 0
def test_disconnect_is_clean_no_error(self):
"""Normal client disconnect does not raise or leave dangling state."""
with (
patch("app.routers.ws.radio_manager") as mock_ws_rm,
patch("app.routers.health.radio_manager") as mock_health_rm,
patch("app.routers.health.RawPacketRepository") as mock_repo,
patch("app.routers.health.settings") as mock_settings,
patch("app.routers.health.os.path.getsize", return_value=0),
):
mock_ws_rm.is_connected = False
mock_ws_rm.connection_info = None
mock_health_rm.is_connected = False
mock_health_rm.connection_info = None
mock_repo.get_oldest_undecrypted = AsyncMock(return_value=None)
mock_settings.database_path = "/tmp/test.db"
from app.main import app
client = TestClient(app)
# Connect and immediately disconnect -- should not raise
with client.websocket_connect("/api/ws") as ws:
ws.receive_json() # consume health
# Verify clean state
assert len(ws_manager.active_connections) == 0