test: expanded test suite for v0.9.0 modules

Command tests:
- tests/commands/: test_base_command, test_cmd_command, test_dice_command,
  test_hello_command, test_help_command, test_magic8_command,
  test_ping_command, test_roll_command
- tests/test_bridge_bot_responses, test_channel_manager_logic,
  test_checkin_service, test_command_manager, test_command_prefix,
  test_config_merge, test_config_validation, test_db_manager,
  test_plugin_loader, test_profanity_filter, test_security_utils,
  test_service_plugin_loader, test_utils

Integration and unit:
- tests/integration/: test_path_graph_integration, test_path_resolution
- tests/regression/: test_keyword_escapes
- tests/unit/: test_mesh_graph, test_mesh_graph_edges,
  test_mesh_graph_multihop, test_mesh_graph_optimizations,
  test_mesh_graph_scoring, test_mesh_graph_validation,
  test_path_command_graph, test_path_command_graph_selection,
  test_path_command_multibyte

Helpers: tests/conftest.py, tests/helpers.py
This commit is contained in:
Stacy Olivas
2026-03-17 17:45:21 -07:00
parent 6d9d01ee4f
commit 2c4daa1720
42 changed files with 2002 additions and 620 deletions
+7 -9
View File
@@ -1,16 +1,14 @@
"""Tests for modules.commands.base_command."""
import pytest
from unittest.mock import MagicMock
from modules.commands.base_command import BaseCommand
from modules.commands.ping_command import PingCommand
from modules.commands.dadjoke_command import DadJokeCommand
from modules.commands.joke_command import JokeCommand
from modules.commands.stats_command import StatsCommand
from modules.commands.hacker_command import HackerCommand
from modules.commands.sports_command import SportsCommand
from modules.commands.alert_command import AlertCommand
from modules.commands.base_command import BaseCommand
from modules.commands.dadjoke_command import DadJokeCommand
from modules.commands.hacker_command import HackerCommand
from modules.commands.joke_command import JokeCommand
from modules.commands.ping_command import PingCommand
from modules.commands.sports_command import SportsCommand
from modules.commands.stats_command import StatsCommand
from modules.models import MeshMessage
from tests.conftest import mock_message
+1 -1
View File
@@ -3,7 +3,7 @@
import pytest
from modules.commands.cmd_command import CmdCommand
from tests.conftest import command_mock_bot, mock_message
from tests.conftest import mock_message
class TestCmdCommand:
+2 -2
View File
@@ -1,10 +1,10 @@
"""Tests for modules.commands.dice_command."""
import pytest
from unittest.mock import patch
from modules.commands.dice_command import DiceCommand
from tests.conftest import command_mock_bot, mock_message
from tests.conftest import mock_message
class TestParseDiceNotation:
+3 -2
View File
@@ -1,10 +1,11 @@
"""Tests for modules.commands.hello_command."""
import pytest
from unittest.mock import patch
import pytest
from modules.commands.hello_command import HelloCommand
from tests.conftest import command_mock_bot, mock_message
from tests.conftest import mock_message
class TestHelloCommand:
+1 -1
View File
@@ -3,7 +3,7 @@
import pytest
from modules.commands.help_command import HelpCommand
from tests.conftest import command_mock_bot, mock_message
from tests.conftest import mock_message
class TestHelpCommand:
+1 -1
View File
@@ -3,7 +3,7 @@
import pytest
from modules.commands.magic8_command import Magic8Command, magic8_responses
from tests.conftest import command_mock_bot, mock_message
from tests.conftest import mock_message
class TestMagic8Command:
+1 -1
View File
@@ -3,7 +3,7 @@
import pytest
from modules.commands.ping_command import PingCommand
from tests.conftest import command_mock_bot, mock_message
from tests.conftest import mock_message
class TestPingCommand:
+2 -1
View File
@@ -1,10 +1,11 @@
"""Tests for modules.commands.roll_command."""
import re
import pytest
from modules.commands.roll_command import RollCommand
from tests.conftest import command_mock_bot, mock_message
from tests.conftest import mock_message
class TestParseRollNotation:
+19 -18
View File
@@ -3,19 +3,20 @@
Pytest fixtures for meshcore-bot tests
"""
import pytest
import sqlite3
import configparser
import sqlite3
from contextlib import closing
from pathlib import Path
from unittest.mock import Mock, MagicMock, AsyncMock
from datetime import datetime, timedelta
from pathlib import Path
from typing import Any, Optional
from unittest.mock import AsyncMock, MagicMock, Mock
import pytest
from modules.db_manager import DBManager
from modules.mesh_graph import MeshGraph
from modules.models import MeshMessage
from tests.helpers import create_test_repeater, create_test_edge, populate_test_graph
from tests.helpers import create_test_edge, populate_test_graph
def mock_message(
@@ -115,7 +116,7 @@ def mock_logger():
def test_config():
"""Create a test configuration with Path_Command settings."""
config = configparser.ConfigParser()
# Add Path_Command section with graph-related settings
config.add_section('Path_Command')
config.set('Path_Command', 'enabled', 'true')
@@ -135,12 +136,12 @@ def test_config():
config.set('Path_Command', 'graph_geographic_weight', '0.7')
config.set('Path_Command', 'graph_prefer_stored_keys', 'true')
config.set('Path_Command', 'star_bias_multiplier', '2.5')
# Add Bot section (for location if needed)
config.add_section('Bot')
config.set('Bot', 'bot_latitude', '47.6062')
config.set('Bot', 'bot_longitude', '-122.3321')
return config
@@ -152,14 +153,14 @@ def test_db(mock_logger, tmp_path):
SQLite :memory: creates a new empty DB per connection, causing isolation issues.
"""
db_path = str(tmp_path / "test.db")
# Create a minimal bot mock for DBManager
mock_bot = Mock()
mock_bot.logger = mock_logger
# Create DBManager with file-based database
db_manager = DBManager(mock_bot, db_path)
# Initialize mesh_connections table schema
db_manager.create_table('mesh_connections', '''
id INTEGER PRIMARY KEY AUTOINCREMENT,
@@ -174,7 +175,7 @@ def test_db(mock_logger, tmp_path):
geographic_distance REAL,
UNIQUE(from_prefix, to_prefix)
''')
# Initialize complete_contact_tracking table schema (for repeater lookups)
db_manager.create_table('complete_contact_tracking', '''
id INTEGER PRIMARY KEY AUTOINCREMENT,
@@ -203,7 +204,7 @@ def test_db(mock_logger, tmp_path):
out_bytes_per_hop INTEGER,
is_starred INTEGER DEFAULT 0
''')
# Create indexes (after tables are created)
# Create indexes (db_manager created tables in same db_path)
try:
@@ -224,9 +225,9 @@ def test_db(mock_logger, tmp_path):
except Exception:
# Indexes are optional, continue if they fail
pass
yield db_manager
# Cleanup (tmp_path is automatically cleaned up by pytest)
@@ -241,14 +242,14 @@ def mock_bot(mock_logger, test_config, test_db):
bot._local_root = None # Use bot_root / local / commands in CommandManager
bot.prefix_hex_chars = 2 # For path/prefix logic (PR #77)
bot.key_prefix = lambda pk: (pk or '')[: getattr(bot, 'prefix_hex_chars', 2)] # For path_command graph selection
# Mock repeater_manager if needed
bot.repeater_manager = Mock()
bot.repeater_manager.get_repeater_devices = Mock(return_value=[])
# Mock web_viewer_integration (optional, for edge notifications)
bot.web_viewer_integration = None
return bot
+17 -17
View File
@@ -3,8 +3,8 @@
Test helper functions and factories for creating test data
"""
from datetime import datetime, timedelta
from typing import Dict, Any, List, Optional, Tuple
from datetime import datetime
from typing import Any, Optional
def create_test_repeater(
@@ -17,9 +17,9 @@ def create_test_repeater(
last_heard: Optional[datetime] = None,
last_advert_timestamp: Optional[datetime] = None,
role: str = "repeater"
) -> Dict[str, Any]:
) -> dict[str, Any]:
"""Factory function to create test repeater data.
Args:
prefix: Two-character hex prefix (default: "01")
name: Repeater name
@@ -30,20 +30,20 @@ def create_test_repeater(
last_heard: Last heard timestamp (default: now)
last_advert_timestamp: Last advert timestamp (default: now)
role: Device role (default: "repeater")
Returns:
Dictionary with repeater data matching database schema
"""
if public_key is None:
# Generate a realistic-looking public key from prefix
public_key = (prefix.lower() * 16)[:64] # 64 hex chars = 32 bytes
now = datetime.now()
if last_heard is None:
last_heard = now
if last_advert_timestamp is None:
last_advert_timestamp = now
return {
'name': name,
'public_key': public_key,
@@ -76,9 +76,9 @@ def create_test_edge(
avg_hop_position: Optional[float] = None,
geographic_distance: Optional[float] = None,
prefix_hex_chars: int = 2
) -> Dict[str, Any]:
) -> dict[str, Any]:
"""Factory function to create test edge data.
Args:
from_prefix: Source node prefix
to_prefix: Destination node prefix
@@ -90,7 +90,7 @@ def create_test_edge(
avg_hop_position: Average hop position in paths
geographic_distance: Distance in km
prefix_hex_chars: Number of hex chars per prefix (default 2). Use bot.prefix_hex_chars when testing with a bot.
Returns:
Dictionary with edge data matching MeshGraph edge structure
"""
@@ -99,12 +99,12 @@ def create_test_edge(
first_seen = now
if last_seen is None:
last_seen = now
if from_public_key is None:
from_public_key = (from_prefix.lower() * 16)[:64]
if to_public_key is None:
to_public_key = (to_prefix.lower() * 16)[:64]
return {
'from_prefix': from_prefix.lower()[:prefix_hex_chars],
'to_prefix': to_prefix.lower()[:prefix_hex_chars],
@@ -118,22 +118,22 @@ def create_test_edge(
}
def create_test_path(node_ids: List[str], prefix_hex_chars: int = 2) -> List[str]:
def create_test_path(node_ids: list[str], prefix_hex_chars: int = 2) -> list[str]:
"""Factory function to create test path data.
Args:
node_ids: List of node prefixes in path order
prefix_hex_chars: Number of hex chars per node (default 2). Use bot.prefix_hex_chars when testing with a bot.
Returns:
List of node IDs (normalized to lowercase)
"""
return [node_id.lower()[:prefix_hex_chars] for node_id in node_ids]
def populate_test_graph(mesh_graph, edges: List[Dict[str, Any]], prefix_hex_chars: int = 2):
def populate_test_graph(mesh_graph, edges: list[dict[str, Any]], prefix_hex_chars: int = 2):
"""Helper to populate a MeshGraph instance with test edges.
Args:
mesh_graph: MeshGraph instance to populate
edges: List of edge dictionaries (from create_test_edge)
@@ -3,69 +3,71 @@
Integration tests for path resolution with graph-based validation
"""
import pytest
from datetime import datetime, timedelta
from datetime import datetime
from unittest.mock import Mock
import pytest
from modules.commands.path_command import PathCommand
from modules.mesh_graph import MeshGraph
from tests.helpers import create_test_repeater, create_test_edge, populate_test_graph
from tests.helpers import create_test_repeater
@pytest.mark.integration
class TestPathResolutionIntegration:
"""Integration tests for full path resolution."""
@pytest.mark.asyncio
async def test_path_resolution_with_graph_data(self, mock_bot, test_db, mesh_graph):
"""Test complete path resolution using real database."""
mock_bot.mesh_graph = mesh_graph
# Populate database with repeater data
test_db.execute_update('''
INSERT INTO complete_contact_tracking
INSERT INTO complete_contact_tracking
(public_key, name, role, last_heard, latitude, longitude, is_starred)
VALUES (?, ?, ?, ?, ?, ?, ?)
''', ('0101010101010101010101010101010101010101010101010101010101010101',
''', ('0101010101010101010101010101010101010101010101010101010101010101',
'Repeater 01', 'repeater', datetime.now().isoformat(), 47.6062, -122.3321, 0))
test_db.execute_update('''
INSERT INTO complete_contact_tracking
INSERT INTO complete_contact_tracking
(public_key, name, role, last_heard, latitude, longitude, is_starred)
VALUES (?, ?, ?, ?, ?, ?, ?)
''', ('7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e',
'Repeater 7e', 'repeater', datetime.now().isoformat(), 47.5, -122.3, 0))
# Create graph edge
mesh_graph.add_edge('01', '7e')
for _ in range(5):
mesh_graph.add_edge('01', '7e')
path_cmd = PathCommand(mock_bot)
# Mock the lookup function to return our test data
def mock_lookup(node_id):
if node_id == '01':
return [create_test_repeater('01', 'Repeater 01',
return [create_test_repeater('01', 'Repeater 01',
public_key='0101010101010101010101010101010101010101010101010101010101010101')]
elif node_id == '7e':
return [create_test_repeater('7e', 'Repeater 7e',
public_key='7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e')]
return []
# Test path resolution
path = ['01', '7e']
result = await path_cmd._lookup_repeater_names(path, lookup_func=mock_lookup)
assert len(result) > 0
@pytest.mark.asyncio
async def test_path_resolution_prefix_collision(self, mock_bot, test_db, mesh_graph):
"""Test path with prefix collisions using graph-based disambiguation."""
mock_bot.mesh_graph = mesh_graph
key1 = '7e1111111111111111111111111111111111111111111111111111111111111111'
key2 = '7e2222222222222222222222222222222222222222222222222222222222222222'
# Mock repeater_manager to return two repeaters with same prefix
async def mock_get_repeater_devices(include_historical=True):
return [
@@ -124,37 +126,37 @@ class TestPathResolutionIntegration:
'is_starred': 0
}
]
mock_bot.repeater_manager = Mock()
mock_bot.repeater_manager.get_repeater_devices = mock_get_repeater_devices
# Create graph edge to local repeater
mesh_graph.add_edge('01', '7e', to_public_key=key1)
for _ in range(10):
mesh_graph.add_edge('01', '7e')
path_cmd = PathCommand(mock_bot)
path = ['01', '7e']
result = await path_cmd._lookup_repeater_names(path)
# Should select local starred repeater with graph edge
assert len(result) > 0
if '7e' in result:
# Verify it selected the correct one (should be Local 7e)
assert result['7e']['name'] == 'Local 7e'
@pytest.mark.asyncio
async def test_path_resolution_starred_preference(self, mock_bot, test_db, mesh_graph):
"""Test starred repeater preference in collisions."""
mock_bot.mesh_graph = mesh_graph
# Create edges for both repeaters
mesh_graph.add_edge('01', '7e')
mesh_graph.add_edge('01', '7a')
path_cmd = PathCommand(mock_bot)
def mock_lookup(node_id):
if node_id == '01':
return [create_test_repeater('01', 'Repeater 01')]
@@ -164,25 +166,25 @@ class TestPathResolutionIntegration:
create_test_repeater('7e', 'Regular 7e', is_starred=False)
]
return []
path = ['01', '7e']
result = await path_cmd._lookup_repeater_names(path, lookup_func=mock_lookup)
# Should prefer starred repeater
assert len(result) > 0
@pytest.mark.asyncio
async def test_path_resolution_stored_keys_priority(self, mock_bot, test_db, mesh_graph):
"""Test stored public key priority."""
mock_bot.mesh_graph = mesh_graph
# Create edge with stored public key
stored_key = '7e1111111111111111111111111111111111111111111111111111111111111111'
other_key = '7e2222222222222222222222222222222222222222222222222222222222222222'
mesh_graph.add_edge('01', '7e', to_public_key=stored_key)
for _ in range(5):
mesh_graph.add_edge('01', '7e')
async def mock_get_repeater_devices(include_historical=True):
return [
{
@@ -240,32 +242,32 @@ class TestPathResolutionIntegration:
'is_starred': 0
}
]
mock_bot.repeater_manager = Mock()
mock_bot.repeater_manager.get_repeater_devices = mock_get_repeater_devices
path_cmd = PathCommand(mock_bot)
path = ['01', '7e']
result = await path_cmd._lookup_repeater_names(path)
# Should select repeater with matching stored key
assert len(result) > 0
if '7e' in result:
assert result['7e']['name'] == 'Matching Key'
@pytest.mark.asyncio
async def test_path_resolution_multi_hop_inference(self, mock_bot, test_db, mesh_graph):
"""Test multi-hop path inference in real scenario."""
mock_bot.mesh_graph = mesh_graph
# Create 2-hop path: 01 -> 7e -> 86
mesh_graph.add_edge('01', '7e')
mesh_graph.add_edge('7e', '86')
path_cmd = PathCommand(mock_bot)
path_cmd.graph_multi_hop_enabled = True
def mock_lookup(node_id):
if node_id == '01':
return [create_test_repeater('01', 'Repeater 01')]
@@ -274,12 +276,12 @@ class TestPathResolutionIntegration:
elif node_id == '86':
return [create_test_repeater('86', 'Repeater 86')]
return []
path = ['01', '7e', '86']
result = await path_cmd._lookup_repeater_names(path, lookup_func=mock_lookup)
assert len(result) > 0
def test_path_resolution_edge_persistence(self, mock_bot, test_db):
"""Test edge persistence across operations."""
# Create graph and add edge
@@ -287,52 +289,52 @@ class TestPathResolutionIntegration:
graph1.add_edge('01', '7e')
for _ in range(5):
graph1.add_edge('01', '7e')
# Verify in database
results = test_db.execute_query('SELECT * FROM mesh_connections WHERE from_prefix = ? AND to_prefix = ?',
results = test_db.execute_query('SELECT * FROM mesh_connections WHERE from_prefix = ? AND to_prefix = ?',
('01', '7e'))
assert len(results) == 1
assert results[0]['observation_count'] == 6
# Create new graph instance (simulates restart)
graph2 = MeshGraph(mock_bot)
# Edge should be loaded from database
edge = graph2.get_edge('01', '7e')
assert edge is not None
assert edge['observation_count'] == 6
@pytest.mark.asyncio
async def test_path_resolution_real_world_scenario(self, mock_bot, test_db, mesh_graph):
"""Test with realistic path data."""
mock_bot.mesh_graph = mesh_graph
# Create realistic path: 01 -> 7e -> 86 -> e0 -> 09
path_nodes = ['01', '7e', '86', 'e0', '09']
# Add edges with varying strengths
mesh_graph.add_edge('01', '7e')
for _ in range(10):
mesh_graph.add_edge('01', '7e') # Strong
mesh_graph.add_edge('7e', '86')
for _ in range(5):
mesh_graph.add_edge('7e', '86') # Medium
mesh_graph.add_edge('86', 'e0')
for _ in range(3):
mesh_graph.add_edge('86', 'e0') # Weak
mesh_graph.add_edge('e0', '09')
for _ in range(8):
mesh_graph.add_edge('e0', '09') # Strong
path_cmd = PathCommand(mock_bot)
def mock_lookup(node_id):
return [create_test_repeater(node_id, f'Repeater {node_id}')]
result = await path_cmd._lookup_repeater_names(path_nodes, lookup_func=mock_lookup)
# Should resolve all nodes
assert len(result) == len(path_nodes)
+31 -31
View File
@@ -3,32 +3,33 @@
Integration tests for full path resolution
"""
import pytest
import sqlite3
from contextlib import closing
from datetime import datetime, timedelta
import pytest
from modules.commands.path_command import PathCommand
from modules.mesh_graph import MeshGraph
from tests.helpers import create_test_repeater, create_test_edge
from tests.helpers import create_test_repeater
@pytest.mark.integration
class TestPathResolutionIntegration:
"""Integration tests for full path resolution with real database."""
def test_path_resolution_with_graph_edges(self, mock_bot, test_db, populated_mesh_graph):
"""Test path resolution using graph edges from database."""
# Add repeaters to database
repeater1 = create_test_repeater('01', 'Repeater 01', latitude=47.6062, longitude=-122.3321)
repeater2 = create_test_repeater('7e', 'Repeater 7e', latitude=47.6100, longitude=-122.3400)
repeater3 = create_test_repeater('86', 'Repeater 86', latitude=47.6200, longitude=-122.3500)
# Insert into database
with closing(sqlite3.connect(test_db.db_path)) as conn:
cursor = conn.cursor()
for r in [repeater1, repeater2, repeater3]:
cursor.execute('''
INSERT INTO complete_contact_tracking
INSERT INTO complete_contact_tracking
(public_key, name, role, latitude, longitude, last_heard, is_starred)
VALUES (?, ?, ?, ?, ?, ?, ?)
''', (
@@ -37,38 +38,37 @@ class TestPathResolutionIntegration:
1 if r['is_starred'] else 0
))
conn.commit()
# Create path command
command = PathCommand(mock_bot)
command.graph_based_validation = True
# Test path resolution
path_hex = "017e86"
# This would normally be called via handle_command, but we test the core logic
# For integration, we verify the graph edges are used correctly
# Verify edges exist in graph
assert populated_mesh_graph.has_edge('01', '7e')
assert populated_mesh_graph.has_edge('7e', '86')
def test_path_resolution_prefix_collision(self, mock_bot, test_db, mesh_graph):
"""Test path resolution with prefix collisions resolved by graph."""
# Create two repeaters with same prefix but different public keys
key1 = '7e' * 32
key2 = '7f' * 32
repeater1 = create_test_repeater('7e', 'Repeater 7e A', public_key=key1, latitude=47.6062, longitude=-122.3321)
repeater2 = create_test_repeater('7e', 'Repeater 7e B', public_key=key2, latitude=47.7000, longitude=-122.4000)
# Add edge with stored public key matching repeater1
mesh_graph.add_edge('01', '7e', to_public_key=key1)
# Insert into database
with closing(sqlite3.connect(test_db.db_path)) as conn:
cursor = conn.cursor()
for r in [repeater1, repeater2]:
cursor.execute('''
INSERT INTO complete_contact_tracking
INSERT INTO complete_contact_tracking
(public_key, name, role, latitude, longitude, last_heard, is_starred)
VALUES (?, ?, ?, ?, ?, ?, ?)
''', (
@@ -77,55 +77,55 @@ class TestPathResolutionIntegration:
1 if r['is_starred'] else 0
))
conn.commit()
# Graph should prefer repeater1 due to stored key match
edge = mesh_graph.get_edge('01', '7e')
assert edge is not None
assert edge.get('to_public_key') == key1
def test_edge_persistence_across_restarts(self, mock_bot, test_db, populated_mesh_graph):
"""Test that edges persist in database across graph restarts."""
# Add edge to existing graph
graph1 = populated_mesh_graph
graph1.add_edge('01', '7e', from_public_key='01' * 32, to_public_key='7e' * 32)
# Verify edge in database
with closing(sqlite3.connect(test_db.db_path)) as conn:
cursor = conn.cursor()
cursor.execute('SELECT * FROM mesh_connections WHERE from_prefix = ? AND to_prefix = ?', ('01', '7e'))
row = cursor.fetchone()
assert row is not None
# Create new graph instance (simulating restart) - loads from same db via mock_bot.db_manager
graph2 = MeshGraph(mock_bot)
# Edge should be loaded from database
assert graph2.has_edge('01', '7e')
edge = graph2.get_edge('01', '7e')
assert edge['from_public_key'] == '01' * 32
assert edge['to_public_key'] == '7e' * 32
def test_graph_vs_geographic_selection(self, mock_bot, test_db, populated_mesh_graph):
"""Test that graph selection can override geographic when graph evidence is strong."""
# Create two repeaters with same prefix
# Repeater1: closer geographically
# Repeater2: has strong graph evidence
repeater1 = create_test_repeater('7e', 'Close Repeater', latitude=47.6062, longitude=-122.3321)
repeater2 = create_test_repeater('7e', 'Graph Repeater', latitude=47.7000, longitude=-122.4000)
# Add strong graph edge for repeater2
key2 = repeater2['public_key']
populated_mesh_graph.add_edge('01', '7e', to_public_key=key2)
populated_mesh_graph.add_edge('01', '7e', to_public_key=key2) # Multiple observations
populated_mesh_graph.add_edge('01', '7e', to_public_key=key2)
# Insert into database
with closing(sqlite3.connect(test_db.db_path)) as conn:
cursor = conn.cursor()
for r in [repeater1, repeater2]:
cursor.execute('''
INSERT INTO complete_contact_tracking
INSERT INTO complete_contact_tracking
(public_key, name, role, latitude, longitude, last_heard, is_starred)
VALUES (?, ?, ?, ?, ?, ?, ?)
''', (
@@ -134,18 +134,18 @@ class TestPathResolutionIntegration:
1 if r['is_starred'] else 0
))
conn.commit()
# Graph should prefer repeater2 due to stored key and multiple observations
edge = populated_mesh_graph.get_edge('01', '7e')
assert edge is not None
assert edge.get('to_public_key') == key2
assert edge['observation_count'] >= 3
def test_real_world_path_scenario(self, mock_bot, test_db, populated_mesh_graph):
"""Test a realistic multi-hop path scenario."""
# Create a realistic path: 01 -> 7e -> 86 -> e0 -> 09
path_nodes = ['01', '7e', '86', 'e0', '09']
# Add repeaters
repeaters = []
for i, node_id in enumerate(path_nodes):
@@ -156,13 +156,13 @@ class TestPathResolutionIntegration:
latitude=lat, longitude=lon
)
repeaters.append(repeater)
# Insert into database
with closing(sqlite3.connect(test_db.db_path)) as conn:
cursor = conn.cursor()
for r in repeaters:
cursor.execute('''
INSERT INTO complete_contact_tracking
INSERT INTO complete_contact_tracking
(public_key, name, role, latitude, longitude, last_heard, is_starred)
VALUES (?, ?, ?, ?, ?, ?, ?)
''', (
@@ -171,7 +171,7 @@ class TestPathResolutionIntegration:
1 if r['is_starred'] else 0
))
conn.commit()
# Verify path is valid in graph
is_valid, confidence = populated_mesh_graph.validate_path(path_nodes)
assert is_valid is True
-1
View File
@@ -1,6 +1,5 @@
"""Regression tests for keyword escape sequences."""
import pytest
from modules.utils import decode_escape_sequences
+3 -2
View File
@@ -1,8 +1,9 @@
"""Tests for bridge bot-responses: channel_sent_listeners registration and cleanup."""
import pytest
from configparser import ConfigParser
from unittest.mock import Mock, MagicMock, AsyncMock, patch
from unittest.mock import MagicMock, patch
import pytest
from modules.service_plugins.discord_bridge_service import DiscordBridgeService
from modules.service_plugins.telegram_bridge_service import TelegramBridgeService
+124 -1
View File
@@ -1,9 +1,9 @@
"""Tests for ChannelManager pure logic (no meshcore device calls)."""
import hashlib
from unittest.mock import AsyncMock, Mock, patch
import pytest
from unittest.mock import Mock
from modules.channel_manager import ChannelManager
@@ -78,3 +78,126 @@ class TestCacheManagement:
cm._cache_valid = True
cm.invalidate_cache()
assert cm._cache_valid is False
# ---------------------------------------------------------------------------
# TestGetCachedChannels
# ---------------------------------------------------------------------------
class TestGetCachedChannels:
"""Tests for _get_cached_channels()."""
def test_returns_channels_sorted_by_index(self, cm):
cm._channels_cache = {
2: {"channel_name": "third", "channel_idx": 2},
0: {"channel_name": "first", "channel_idx": 0},
1: {"channel_name": "second", "channel_idx": 1},
}
result = cm._get_cached_channels()
assert [c["channel_name"] for c in result] == ["first", "second", "third"]
def test_empty_cache_returns_empty_list(self, cm):
cm._channels_cache = {}
assert cm._get_cached_channels() == []
def test_single_channel_in_cache(self, cm):
cm._channels_cache = {0: {"channel_name": "solo", "channel_idx": 0}}
result = cm._get_cached_channels()
assert len(result) == 1
assert result[0]["channel_name"] == "solo"
# ---------------------------------------------------------------------------
# TestFetchAllChannelsCacheLifecycle
# ---------------------------------------------------------------------------
class TestFetchAllChannelsCacheLifecycle:
"""Tests for fetch_all_channels() cache validity transitions."""
@pytest.mark.asyncio
async def test_cache_valid_returns_cached_without_device_call(self, cm):
cm._channels_cache = {0: {"channel_name": "general", "channel_idx": 0}}
cm._cache_valid = True
cm.bot.connected = True
channels = await cm.fetch_all_channels(force_refresh=False)
assert len(channels) == 1
assert channels[0]["channel_name"] == "general"
@pytest.mark.asyncio
async def test_device_not_connected_returns_empty_list(self, cm):
cm.bot.connected = False
channels = await cm.fetch_all_channels(force_refresh=True)
assert channels == []
@pytest.mark.asyncio
async def test_force_refresh_when_disconnected_preserves_existing_cache(self, cm):
"""When device is not connected, the connectivity check fires before cache clear,
so the existing cache is preserved (early return before clear)."""
cm._channels_cache = {0: {"channel_name": "existing", "channel_idx": 0}}
cm._cache_valid = True
cm.bot.connected = False
channels = await cm.fetch_all_channels(force_refresh=True)
# Returns empty list (device not connected)
assert channels == []
# Cache not cleared because early return before clear_cache step
assert 0 in cm._channels_cache
@pytest.mark.asyncio
async def test_successful_fetch_marks_cache_valid(self, cm):
"""After a successful fetch, _cache_valid should be True."""
cm.bot.connected = True
async def fake_fetch_single(idx):
if idx == 0:
return {"channel_name": "general", "channel_idx": 0, "channel_key_hex": ""}
return None
cm._fetch_single_channel = fake_fetch_single
cm._store_channels_in_db = Mock()
channels = await cm.fetch_all_channels(force_refresh=True)
assert cm._cache_valid is True
assert any(c["channel_name"] == "general" for c in channels)
@pytest.mark.asyncio
async def test_three_consecutive_timeouts_aborts_fetch(self, cm):
"""If first 3 channels all return None, the fetch should abort early."""
cm.bot.connected = True
call_count = 0
async def always_none(idx):
nonlocal call_count
call_count += 1
return None
cm._fetch_single_channel = always_none
cm._store_channels_in_db = Mock()
with patch("asyncio.sleep", new_callable=AsyncMock):
channels = await cm.fetch_all_channels(force_refresh=True)
# Should have aborted before fetching all 8 channels
assert call_count < cm.max_channels
assert channels == []
# ---------------------------------------------------------------------------
# TestGetChannelByName
# ---------------------------------------------------------------------------
class TestGetChannelByName:
"""Tests for get_channel_number() with various cache states."""
def test_returns_none_when_cache_empty(self, cm):
cm._channels_cache = {}
cm._cache_valid = True
assert cm.get_channel_number("nonexistent") is None
def test_exact_name_lookup(self, cm):
"""get_channel_number does a case-insensitive exact match (no # stripping)."""
cm._channels_cache = {0: {"channel_name": "#general"}}
cm._cache_valid = True
assert cm.get_channel_number("#general") == 0
+5 -3
View File
@@ -4,13 +4,15 @@ Skipped when the checkin_service local plugin is not present (it does not ship w
"""
import configparser
import pytest
# Import from local plugin (repo root is on path when running tests)
import sys
from datetime import datetime
from pathlib import Path
from unittest.mock import MagicMock, Mock, patch
# Import from local plugin (repo root is on path when running tests)
import sys
import pytest
_root = Path(__file__).resolve().parent.parent
if str(_root) not in sys.path:
sys.path.insert(0, str(_root))
+83 -5
View File
@@ -1,14 +1,13 @@
"""Tests for modules.command_manager."""
import time
import pytest
from configparser import ConfigParser
from pathlib import Path
from unittest.mock import Mock, MagicMock, patch, AsyncMock
from unittest.mock import AsyncMock, MagicMock, Mock, patch
import pytest
from modules.command_manager import CommandManager, InternetStatusCache
from modules.models import MeshMessage
from tests.conftest import mock_message
@@ -256,7 +255,7 @@ class TestGetHelpForCommand:
def test_unknown_command_returns_error(self, cm_bot):
manager = make_manager(cm_bot)
result = manager.get_help_for_command("nonexistent")
manager.get_help_for_command("nonexistent")
# Translator receives 'commands.help.unknown' key with command name
cm_bot.translator.translate.assert_called()
call_args = cm_bot.translator.translate.call_args
@@ -290,6 +289,7 @@ class TestSendChannelMessageListeners:
async def test_successful_send_invokes_listeners_with_synthetic_event(self, cm_bot, mock_logger):
"""When send_channel_message succeeds, each channel_sent_listener is called with event.payload shape (channel_idx, text)."""
import asyncio
from meshcore import EventType
cm_bot.connected = True
@@ -449,3 +449,81 @@ class TestSendChannelMessagesChunked:
assert result is False
manager.send_channel_message.assert_called_once()
# ---------------------------------------------------------------------------
# TestLoadAliases
# ---------------------------------------------------------------------------
class TestLoadAliases:
"""Tests for load_aliases() config parsing."""
def test_empty_when_no_section(self, cm_bot):
manager = make_manager(cm_bot)
assert manager.aliases == {}
def test_reads_alias_entries(self, cm_bot):
cm_bot.config.add_section("Aliases")
cm_bot.config.set("Aliases", "s", "schedule")
cm_bot.config.set("Aliases", "wx", "weather")
manager = make_manager(cm_bot)
assert manager.aliases == {"s": "schedule", "wx": "weather"}
def test_aliases_are_lowercased(self, cm_bot):
cm_bot.config.add_section("Aliases")
cm_bot.config.set("Aliases", "S", "Schedule")
manager = make_manager(cm_bot)
assert "s" in manager.aliases
assert manager.aliases["s"] == "schedule"
def test_empty_alias_key_ignored(self, cm_bot):
# ConfigParser won't allow a truly empty key, so this tests whitespace values
cm_bot.config.add_section("Aliases")
cm_bot.config.set("Aliases", "wx", "") # empty canonical
manager = make_manager(cm_bot)
assert "wx" not in manager.aliases
# ---------------------------------------------------------------------------
# TestApplyAliases
# ---------------------------------------------------------------------------
class TestApplyAliases:
"""Tests for _apply_aliases() keyword injection."""
def _make_mock_command(self, name, keywords):
cmd = Mock()
cmd.name = name
cmd.keywords = list(keywords)
return cmd
def test_alias_injected_into_command_keywords(self, cm_bot):
sched_cmd = self._make_mock_command("schedule", ["schedule"])
cm_bot.config.add_section("Aliases")
cm_bot.config.set("Aliases", "s", "schedule")
make_manager(cm_bot, commands={"schedule": sched_cmd})
assert "s" in sched_cmd.keywords
def test_unknown_alias_logs_warning_and_skipped(self, cm_bot):
cm_bot.config.add_section("Aliases")
cm_bot.config.set("Aliases", "x", "nonexistent")
make_manager(cm_bot)
cm_bot.logger.warning.assert_called()
def test_duplicate_alias_not_added_twice(self, cm_bot):
sched_cmd = self._make_mock_command("schedule", ["schedule", "s"])
cm_bot.config.add_section("Aliases")
cm_bot.config.set("Aliases", "s", "schedule")
make_manager(cm_bot, commands={"schedule": sched_cmd})
assert sched_cmd.keywords.count("s") == 1
def test_multiple_aliases_for_same_command(self, cm_bot):
wx_cmd = self._make_mock_command("weather", ["weather"])
cm_bot.config.add_section("Aliases")
cm_bot.config.set("Aliases", "wx", "weather")
cm_bot.config.set("Aliases", "w", "weather")
make_manager(cm_bot, commands={"weather": wx_cmd})
assert "wx" in wx_cmd.keywords
assert "w" in wx_cmd.keywords
+56 -56
View File
@@ -4,17 +4,17 @@ Unit tests for command prefix functionality
Tests that all commands properly handle command prefixes when enabled
"""
import pytest
from pathlib import Path
from unittest.mock import Mock, MagicMock, patch
from configparser import ConfigParser
from pathlib import Path
from unittest.mock import Mock, patch
from modules.models import MeshMessage
import pytest
from modules.command_manager import CommandManager
from modules.commands.base_command import BaseCommand
from modules.commands.hello_command import HelloCommand
from modules.commands.ping_command import PingCommand
from modules.commands.help_command import HelpCommand
from modules.command_manager import CommandManager
from modules.models import MeshMessage
class MockTestCommand(BaseCommand):
@@ -23,7 +23,7 @@ class MockTestCommand(BaseCommand):
keywords = ['test', 't']
description = "Test command"
category = "test"
async def execute(self, message: MeshMessage) -> bool:
"""Execute the test command (required by abstract base class)"""
return True
@@ -68,105 +68,105 @@ def mock_message():
class TestCommandPrefix:
"""Test command prefix functionality"""
def test_no_prefix_allows_commands(self, mock_bot, mock_message):
"""Test that without prefix configured, commands work normally"""
mock_bot.config.set('Bot', 'command_prefix', '')
command = MockTestCommand(mock_bot)
# Should match without prefix
assert command.matches_keyword(mock_message) is True
# Should also match with legacy ! prefix
mock_message.content = "!test"
assert command.matches_keyword(mock_message) is True
def test_prefix_required_when_configured(self, mock_bot, mock_message):
"""Test that when prefix is configured, it's required"""
mock_bot.config.set('Bot', 'command_prefix', '!')
command = MockTestCommand(mock_bot)
# Should match with prefix
mock_message.content = "!test"
assert command.matches_keyword(mock_message) is True
# Should NOT match without prefix
mock_message.content = "test"
assert command.matches_keyword(mock_message) is False
def test_dot_prefix(self, mock_bot, mock_message):
"""Test dot prefix (e.g., .ping)"""
mock_bot.config.set('Bot', 'command_prefix', '.')
command = MockTestCommand(mock_bot)
# Should match with dot prefix
mock_message.content = ".test"
assert command.matches_keyword(mock_message) is True
# Should NOT match without prefix
mock_message.content = "test"
assert command.matches_keyword(mock_message) is False
def test_single_char_prefix(self, mock_bot, mock_message):
"""Test single character prefix (e.g., bping)"""
mock_bot.config.set('Bot', 'command_prefix', 'b')
command = MockTestCommand(mock_bot)
# Should match with 'b' prefix
mock_message.content = "btest"
assert command.matches_keyword(mock_message) is True
# Should NOT match without prefix
mock_message.content = "test"
assert command.matches_keyword(mock_message) is False
def test_multi_char_prefix(self, mock_bot, mock_message):
"""Test multi-character prefix (e.g., abcping)"""
mock_bot.config.set('Bot', 'command_prefix', 'abc')
command = MockTestCommand(mock_bot)
# Should match with 'abc' prefix
mock_message.content = "abctest"
assert command.matches_keyword(mock_message) is True
# Should NOT match without prefix
mock_message.content = "test"
assert command.matches_keyword(mock_message) is False
# Should NOT match with partial prefix
mock_message.content = "abtest"
assert command.matches_keyword(mock_message) is False
def test_prefix_with_whitespace(self, mock_bot, mock_message):
"""Test that prefix works with whitespace after it"""
mock_bot.config.set('Bot', 'command_prefix', '!')
command = MockTestCommand(mock_bot)
# Should match with prefix and space
mock_message.content = "! test"
assert command.matches_keyword(mock_message) is True
# Should match with prefix and no space
mock_message.content = "!test"
assert command.matches_keyword(mock_message) is True
def test_prefix_with_keyword_variations(self, mock_bot, mock_message):
"""Test prefix with different keyword variations"""
mock_bot.config.set('Bot', 'command_prefix', '!')
command = MockTestCommand(mock_bot)
# Test first keyword
mock_message.content = "!test"
assert command.matches_keyword(mock_message) is True
# Test second keyword
mock_message.content = "!t"
assert command.matches_keyword(mock_message) is True
# Test keyword with arguments
mock_message.content = "!test arg1 arg2"
assert command.matches_keyword(mock_message) is True
def test_hello_command_with_prefix(self, mock_bot, mock_message):
"""Test hello command specifically with prefix"""
mock_bot.config.set('Bot', 'command_prefix', '!')
@@ -174,30 +174,30 @@ class TestCommandPrefix:
mock_bot.config.add_section('Hello_Command')
mock_bot.config.set('Hello_Command', 'enabled', 'true')
command = HelloCommand(mock_bot)
# Should match with prefix
mock_message.content = "!hello"
assert command.matches_keyword(mock_message) is True
# Should NOT match without prefix
mock_message.content = "hello"
assert command.matches_keyword(mock_message) is False
def test_ping_command_with_prefix(self, mock_bot, mock_message):
"""Test ping command with prefix"""
mock_bot.config.set('Bot', 'command_prefix', '.')
mock_bot.config.add_section('Ping_Command')
mock_bot.config.set('Ping_Command', 'enabled', 'true')
command = PingCommand(mock_bot)
# Should match with dot prefix
mock_message.content = ".ping"
assert command.matches_keyword(mock_message) is True
# Should NOT match without prefix
mock_message.content = "ping"
assert command.matches_keyword(mock_message) is False
def test_command_manager_with_prefix(self, mock_bot, mock_message):
"""Test CommandManager handles prefix correctly"""
mock_bot.config.set('Bot', 'command_prefix', '!')
@@ -205,88 +205,88 @@ class TestCommandPrefix:
mock_bot.config.set('Keywords', 'keywords', '')
mock_bot.config.add_section('Custom_Syntax')
mock_bot.config.set('Custom_Syntax', 'custom_syntax', '')
# Mock plugin loader to return empty commands for simplicity
with patch('modules.command_manager.PluginLoader') as mock_loader_class:
mock_loader = Mock()
mock_loader.load_all_plugins = Mock(return_value={})
mock_loader_class.return_value = mock_loader
manager = CommandManager(mock_bot)
# Should return empty matches for message without prefix
mock_message.content = "test"
matches = manager.check_keywords(mock_message)
assert matches == []
# Should process message with prefix
mock_message.content = "!test"
matches = manager.check_keywords(mock_message)
# Will be empty because no commands loaded, but should process without error
assert isinstance(matches, list)
def test_prefix_with_mentions(self, mock_bot, mock_message):
"""Test that prefix works correctly with @[username] mentions"""
mock_bot.config.set('Bot', 'command_prefix', '!')
mock_bot.config.set('Bot', 'bot_name', 'TestBot')
command = MockTestCommand(mock_bot)
# Mock self_info to return bot name
mock_bot.meshcore = Mock()
mock_bot.meshcore.self_info = {'name': 'TestBot'}
# Should match with prefix and bot mention
mock_message.content = "!@[TestBot] test"
assert command.matches_keyword(mock_message) is True
# Should NOT match with prefix but other user mention
mock_message.content = "!@[OtherUser] test"
assert command.matches_keyword(mock_message) is False
def test_different_prefixes_dont_match(self, mock_bot, mock_message):
"""Test that wrong prefix doesn't match"""
mock_bot.config.set('Bot', 'command_prefix', '!')
command = MockTestCommand(mock_bot)
# Should NOT match with wrong prefix
mock_message.content = ".test"
assert command.matches_keyword(mock_message) is False
mock_message.content = "btest"
assert command.matches_keyword(mock_message) is False
mock_message.content = "abctest"
assert command.matches_keyword(mock_message) is False
def test_prefix_case_sensitive(self, mock_bot, mock_message):
"""Test that prefix matching is case-sensitive"""
mock_bot.config.set('Bot', 'command_prefix', '!')
command = MockTestCommand(mock_bot)
# Should match with exact prefix
mock_message.content = "!test"
assert command.matches_keyword(mock_message) is True
# Prefix matching is case-sensitive, so different case shouldn't match
# (This tests the actual behavior - prefixes are case-sensitive)
mock_message.content = "!TEST" # Prefix is still '!', so this should match
assert command.matches_keyword(mock_message) is True # '!' is same case
# But if prefix is lowercase, uppercase shouldn't match
mock_bot.config.set('Bot', 'command_prefix', 'b')
command = MockTestCommand(mock_bot)
mock_message.content = "Btest" # Uppercase B
assert command.matches_keyword(mock_message) is False # Should not match lowercase 'b'
def test_empty_prefix_string(self, mock_bot, mock_message):
"""Test that empty string prefix means no prefix required"""
mock_bot.config.set('Bot', 'command_prefix', '')
command = MockTestCommand(mock_bot)
# Should match without prefix
mock_message.content = "test"
assert command.matches_keyword(mock_message) is True
# Should also match with legacy ! prefix
mock_message.content = "!test"
assert command.matches_keyword(mock_message) is True
-1
View File
@@ -1,6 +1,5 @@
"""Tests for config loading and merging of local/config.ini."""
import pytest
from pathlib import Path
from modules.core import MeshCoreBot
+5 -6
View File
@@ -1,18 +1,17 @@
"""Tests for modules.config_validation."""
import pytest
from pathlib import Path
from modules.config_validation import (
SEVERITY_ERROR,
SEVERITY_INFO,
SEVERITY_WARNING,
_check_path_writable,
_get_command_prefix_to_section,
_resolve_path,
_suggest_similar_command,
strip_optional_quotes,
validate_config,
_resolve_path,
_check_path_writable,
_suggest_similar_command,
_get_command_prefix_to_section,
)
@@ -392,5 +391,5 @@ class TestGetCommandPrefixToSection:
def test_contains_known_commands(self):
result = _get_command_prefix_to_section()
assert "stats" in result or "ping" in result
for k, v in result.items():
for _k, v in result.items():
assert v.endswith("_Command")
+97 -2
View File
@@ -1,11 +1,14 @@
"""Tests for MeshCoreBot logic (config loading, radio settings, helpers)."""
import pytest
import asyncio
import struct
from pathlib import Path
from unittest.mock import MagicMock, patch
import pytest
from modules.core import MeshCoreBot
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
@@ -185,3 +188,95 @@ monitor_channels = #general
assert b.prefix_hex_chars == 4
assert b.is_valid_prefix("dead") is True
assert b.is_valid_prefix("de") is False
# ---------------------------------------------------------------------------
# Loop exception handler (TASK-00 / BUG-022)
# ---------------------------------------------------------------------------
class TestLoopExceptionHandler:
"""Verify the custom asyncio exception handler installed by start()."""
def _make_bot(self, tmp_path: Path) -> MeshCoreBot:
config_file = tmp_path / "config.ini"
db_path = tmp_path / "bot.db"
_write_config(config_file, db_path)
return MeshCoreBot(config_file=str(config_file))
def _extract_handler(self, bot: MeshCoreBot) -> object:
"""Run a fake start() up to the set_exception_handler call and return the handler."""
mock_loop = MagicMock(spec=asyncio.AbstractEventLoop)
mock_loop.get_exception_handler.return_value = None
captured: list = []
def capture_handler(h):
captured.append(h)
mock_loop.set_exception_handler.side_effect = capture_handler
with patch.object(bot, "connect", return_value=False):
with patch("asyncio.get_running_loop", return_value=mock_loop):
asyncio.run(bot.start())
assert captured, "set_exception_handler was never called"
return captured[0], mock_loop
def test_handler_is_installed_on_start(self, tmp_path):
bot = self._make_bot(tmp_path)
handler, mock_loop = self._extract_handler(bot)
mock_loop.set_exception_handler.assert_called_once()
assert callable(handler)
def test_index_error_logged_at_debug_not_propagated(self, tmp_path):
bot = self._make_bot(tmp_path)
handler, mock_loop = self._extract_handler(bot)
with patch.object(bot.logger, "debug") as mock_debug:
handler(mock_loop, {"exception": IndexError("index out of range")})
mock_debug.assert_called_once()
assert "IndexError" in mock_debug.call_args[0][1]
# default handler must NOT be invoked for IndexError
mock_loop.default_exception_handler.assert_not_called()
def test_struct_error_logged_at_debug_not_propagated(self, tmp_path):
bot = self._make_bot(tmp_path)
handler, mock_loop = self._extract_handler(bot)
with patch.object(bot.logger, "debug") as mock_debug:
handler(mock_loop, {"exception": struct.error("unpack requires")})
mock_debug.assert_called_once()
mock_loop.default_exception_handler.assert_not_called()
def test_other_exception_passes_to_default_handler(self, tmp_path):
bot = self._make_bot(tmp_path)
# Use a real previous handler to verify passthrough
previous_handler = MagicMock()
mock_loop = MagicMock(spec=asyncio.AbstractEventLoop)
mock_loop.get_exception_handler.return_value = previous_handler
captured: list = []
mock_loop.set_exception_handler.side_effect = lambda h: captured.append(h)
with patch.object(bot, "connect", return_value=False):
with patch("asyncio.get_running_loop", return_value=mock_loop):
asyncio.run(bot.start())
handler = captured[0]
ctx = {"exception": RuntimeError("something else")}
handler(mock_loop, ctx)
previous_handler.assert_called_once_with(mock_loop, ctx)
mock_loop.default_exception_handler.assert_not_called()
def test_no_exception_key_passes_to_default_handler(self, tmp_path):
bot = self._make_bot(tmp_path)
handler, mock_loop = self._extract_handler(bot)
ctx = {"message": "Task destroyed but pending"}
handler(mock_loop, ctx)
mock_loop.default_exception_handler.assert_called_once_with(ctx)
+1 -2
View File
@@ -1,11 +1,10 @@
"""Tests for modules.db_manager."""
import sqlite3
import json
from contextlib import closing
from unittest.mock import Mock
import pytest
from unittest.mock import Mock
from modules.db_manager import DBManager
-1
View File
@@ -1,6 +1,5 @@
"""Tests for modules/enums.py — enum values and flag combinations."""
import pytest
from modules.enums import AdvertFlags, DeviceRole, PayloadType, PayloadVersion, RouteType
+3 -3
View File
@@ -1,11 +1,11 @@
"""Tests for FeedManager pure formatting and filtering logic."""
import json
from datetime import datetime, timezone, timedelta
from configparser import ConfigParser
from datetime import datetime, timedelta, timezone
from unittest.mock import Mock
import pytest
from configparser import ConfigParser
from unittest.mock import Mock
from modules.feed_manager import FeedManager
+495 -10
View File
@@ -1,9 +1,10 @@
"""Tests for MessageHandler pure logic (no network, no meshcore device)."""
import time
import pytest
import configparser
from unittest.mock import Mock, MagicMock
import time
from unittest.mock import AsyncMock, Mock, patch
import pytest
from modules.message_handler import MessageHandler
from modules.models import MeshMessage
@@ -285,16 +286,19 @@ class TestCleanupStaleCacheEntries:
def test_removes_old_timestamp_cache_entries(self, handler):
now = time.time()
# Old entry: older than rf_data_timeout
handler.rf_data_by_timestamp[now - 100] = {"timestamp": now - 100, "data": "old"}
# Recent entry
handler.rf_data_by_timestamp[now] = {"timestamp": now, "data": "new"}
current_time = now + handler._cache_cleanup_interval + 1
# Old entry: well outside rf_data_timeout relative to current_time
old_ts = current_time - handler.rf_data_timeout - 10
# Recent entry: within rf_data_timeout of current_time
recent_ts = current_time - 1
handler.rf_data_by_timestamp[old_ts] = {"timestamp": old_ts, "data": "old"}
handler.rf_data_by_timestamp[recent_ts] = {"timestamp": recent_ts, "data": "new"}
# Force full cleanup
handler._last_cache_cleanup = 0
handler._cleanup_stale_cache_entries(current_time=now + handler._cache_cleanup_interval + 1)
handler._cleanup_stale_cache_entries(current_time=current_time)
# Old entry should be gone, recent kept
assert (now - 100) not in handler.rf_data_by_timestamp
assert now in handler.rf_data_by_timestamp
assert old_ts not in handler.rf_data_by_timestamp
assert recent_ts in handler.rf_data_by_timestamp
def test_removes_stale_pubkey_cache_entries(self, handler):
now = time.time()
@@ -327,3 +331,484 @@ class TestCleanupStaleCacheEntries:
handler._cleanup_stale_cache_entries(current_time=now + 1)
# Still cleaned (timeout-only cleanup still runs)
assert stale_ts not in handler.rf_data_by_timestamp
# ---------------------------------------------------------------------------
# find_recent_rf_data
# ---------------------------------------------------------------------------
class TestFindRecentRfData:
"""Tests for MessageHandler.find_recent_rf_data()."""
def _rf_entry(self, age=0, packet_prefix="aabbccdd", pubkey_prefix="1122"):
return {
"timestamp": time.time() - age,
"snr": 5,
"rssi": -80,
"packet_prefix": packet_prefix,
"pubkey_prefix": pubkey_prefix,
}
def test_returns_none_when_empty(self, handler):
handler.recent_rf_data = []
assert handler.find_recent_rf_data() is None
def test_returns_none_when_all_too_old(self, handler):
handler.rf_data_timeout = 5
handler.recent_rf_data = [self._rf_entry(age=100)]
assert handler.find_recent_rf_data() is None
def test_returns_most_recent_fallback(self, handler):
handler.rf_data_timeout = 30
entry = self._rf_entry(age=1)
handler.recent_rf_data = [entry]
result = handler.find_recent_rf_data()
assert result is entry
def test_exact_packet_prefix_match(self, handler):
handler.rf_data_timeout = 30
target = self._rf_entry(age=1, packet_prefix="deadbeefdeadbeef1234567890abcdef")
other = self._rf_entry(age=2, packet_prefix="00000000000000000000000000000000")
handler.recent_rf_data = [target, other]
result = handler.find_recent_rf_data("deadbeefdeadbeef1234567890abcdef")
assert result is target
def test_exact_pubkey_prefix_match(self, handler):
handler.rf_data_timeout = 30
target = self._rf_entry(age=1, pubkey_prefix="abcd", packet_prefix="")
other = self._rf_entry(age=2, pubkey_prefix="1111", packet_prefix="")
handler.recent_rf_data = [target, other]
result = handler.find_recent_rf_data("abcd")
assert result is target
def test_partial_packet_prefix_match(self, handler):
handler.rf_data_timeout = 30
long_prefix = "aabbccddeeff0011aabbccddeeff0011"
partial_key = "aabbccddeeff0011" + "xxxxxxxxxxxxxxxx"
target = self._rf_entry(age=1, packet_prefix=long_prefix, pubkey_prefix="")
handler.recent_rf_data = [target]
result = handler.find_recent_rf_data(partial_key)
assert result is target
def test_no_key_returns_most_recent(self, handler):
handler.rf_data_timeout = 30
old = self._rf_entry(age=10)
new = self._rf_entry(age=1)
handler.recent_rf_data = [old, new]
result = handler.find_recent_rf_data()
assert result["timestamp"] == new["timestamp"]
def test_custom_max_age(self, handler):
handler.rf_data_timeout = 30
entry = self._rf_entry(age=20)
handler.recent_rf_data = [entry]
# With max_age=5, entry is too old
assert handler.find_recent_rf_data(max_age_seconds=5) is None
# With max_age=30, entry is visible
assert handler.find_recent_rf_data(max_age_seconds=30) is entry
# ---------------------------------------------------------------------------
# handle_raw_data
# ---------------------------------------------------------------------------
class TestHandleRawData:
"""Tests for MessageHandler.handle_raw_data()."""
def _make_event(self, payload):
event = Mock()
event.payload = payload
return event
async def test_no_payload_logs_warning(self, handler):
event = Mock(spec=[])
handler.logger = Mock()
await handler.handle_raw_data(event)
handler.logger.warning.assert_called()
async def test_payload_none_logs_warning(self, handler):
event = Mock()
event.payload = None
handler.logger = Mock()
await handler.handle_raw_data(event)
handler.logger.warning.assert_called()
async def test_payload_without_data_field_logs_warning(self, handler):
event = self._make_event({"other": "stuff"})
handler.logger = Mock()
with patch.object(handler, "decode_meshcore_packet", return_value=None):
await handler.handle_raw_data(event)
handler.logger.warning.assert_called()
async def test_payload_with_hex_data_calls_decode(self, handler):
event = self._make_event({"data": "aabbccdd"})
handler.logger = Mock()
with patch.object(handler, "decode_meshcore_packet", return_value=None) as mock_decode:
await handler.handle_raw_data(event)
mock_decode.assert_called_once_with("aabbccdd")
async def test_payload_strips_0x_prefix(self, handler):
event = self._make_event({"data": "0xaabbccdd"})
handler.logger = Mock()
with patch.object(handler, "decode_meshcore_packet", return_value=None) as mock_decode:
await handler.handle_raw_data(event)
mock_decode.assert_called_once_with("aabbccdd")
async def test_decoded_packet_calls_process_advertisement(self, handler):
event = self._make_event({"data": "aabbccdd"})
handler.logger = Mock()
packet_info = {"type": "adv", "node_id": "ab"}
with patch.object(handler, "decode_meshcore_packet", return_value=packet_info):
with patch.object(handler, "_process_advertisement_packet", new_callable=AsyncMock) as mock_adv:
await handler.handle_raw_data(event)
mock_adv.assert_called_once_with(packet_info, None)
async def test_non_string_data_logs_warning(self, handler):
event = self._make_event({"data": 12345})
handler.logger = Mock()
await handler.handle_raw_data(event)
handler.logger.warning.assert_called()
async def test_exception_does_not_raise(self, handler):
event = self._make_event({"data": "aabb"})
handler.logger = Mock()
with patch.object(handler, "decode_meshcore_packet", side_effect=RuntimeError("oops")):
# Should not raise
await handler.handle_raw_data(event)
handler.logger.error.assert_called()
# ---------------------------------------------------------------------------
# handle_contact_message
# ---------------------------------------------------------------------------
class TestHandleContactMessage:
"""Tests for MessageHandler.handle_contact_message()."""
def _make_event(self, payload):
event = Mock()
event.payload = payload
event.metadata = {}
return event
def _setup_handler(self, handler):
handler.logger = Mock()
handler.bot.meshcore = Mock()
handler.bot.meshcore.contacts = {}
handler.bot.translator = None
async def test_no_payload_returns_early(self, handler):
self._setup_handler(handler)
event = Mock(spec=[])
await handler.handle_contact_message(event)
handler.logger.warning.assert_called()
async def test_payload_none_returns_early(self, handler):
self._setup_handler(handler)
event = Mock()
event.payload = None
await handler.handle_contact_message(event)
handler.logger.warning.assert_called()
async def test_old_cached_message_not_processed(self, handler):
self._setup_handler(handler)
# Set connection_time in the future relative to an old timestamp
handler.bot.connection_time = time.time()
old_ts = int(time.time()) - 3600 # 1 hour old
event = self._make_event({
"pubkey_prefix": "ab12",
"text": "hello",
"path_len": 255,
"sender_timestamp": old_ts,
})
with patch.object(handler, "process_message", new_callable=AsyncMock) as mock_pm:
with patch.object(handler, "_debug_decode_message_path", new_callable=AsyncMock):
with patch.object(handler, "_debug_decode_packet_for_message", new_callable=AsyncMock):
await handler.handle_contact_message(event)
mock_pm.assert_not_called()
async def test_new_message_calls_process_message(self, handler):
self._setup_handler(handler)
handler.bot.connection_time = None # No connection time = don't filter
event = self._make_event({
"pubkey_prefix": "ab12",
"text": "hello",
"path_len": 255,
"sender_timestamp": int(time.time()),
})
with patch.object(handler, "process_message", new_callable=AsyncMock) as mock_pm:
with patch.object(handler, "_debug_decode_message_path", new_callable=AsyncMock):
with patch.object(handler, "_debug_decode_packet_for_message", new_callable=AsyncMock):
await handler.handle_contact_message(event)
mock_pm.assert_called_once()
async def test_snr_from_payload(self, handler):
self._setup_handler(handler)
handler.bot.connection_time = None
captured = {}
async def capture_message(msg):
captured["msg"] = msg
event = self._make_event({
"pubkey_prefix": "ab12",
"text": "hello",
"path_len": 255,
"sender_timestamp": int(time.time()),
"SNR": 7,
"RSSI": -70,
})
with patch.object(handler, "process_message", side_effect=capture_message):
with patch.object(handler, "_debug_decode_message_path", new_callable=AsyncMock):
with patch.object(handler, "_debug_decode_packet_for_message", new_callable=AsyncMock):
await handler.handle_contact_message(event)
assert captured["msg"].snr == 7
assert captured["msg"].rssi == -70
async def test_direct_path_len_255(self, handler):
self._setup_handler(handler)
handler.bot.connection_time = None
captured = {}
async def capture_message(msg):
captured["msg"] = msg
event = self._make_event({
"pubkey_prefix": "ab12",
"text": "hi",
"path_len": 255,
"sender_timestamp": int(time.time()),
})
with patch.object(handler, "process_message", side_effect=capture_message):
with patch.object(handler, "_debug_decode_message_path", new_callable=AsyncMock):
with patch.object(handler, "_debug_decode_packet_for_message", new_callable=AsyncMock):
await handler.handle_contact_message(event)
assert captured["msg"].is_dm is True
async def test_message_is_dm(self, handler):
self._setup_handler(handler)
handler.bot.connection_time = None
captured = {}
async def capture_message(msg):
captured["msg"] = msg
event = self._make_event({
"pubkey_prefix": "ab12",
"text": "dm text",
"path_len": 0,
"sender_timestamp": int(time.time()),
})
with patch.object(handler, "process_message", side_effect=capture_message):
with patch.object(handler, "_debug_decode_message_path", new_callable=AsyncMock):
with patch.object(handler, "_debug_decode_packet_for_message", new_callable=AsyncMock):
await handler.handle_contact_message(event)
assert captured["msg"].is_dm is True
assert captured["msg"].content == "dm text"
async def test_contact_name_lookup(self, handler):
self._setup_handler(handler)
handler.bot.connection_time = None
handler.bot.meshcore.contacts = {
"key1": {
"public_key": "ab12deadbeef",
"name": "Alice",
"out_path": "",
"out_path_len": 0,
}
}
captured = {}
async def capture_message(msg):
captured["msg"] = msg
event = self._make_event({
"pubkey_prefix": "ab12",
"text": "hi",
"path_len": 255,
"sender_timestamp": int(time.time()),
})
with patch.object(handler, "process_message", side_effect=capture_message):
with patch.object(handler, "_debug_decode_message_path", new_callable=AsyncMock):
with patch.object(handler, "_debug_decode_packet_for_message", new_callable=AsyncMock):
await handler.handle_contact_message(event)
assert captured["msg"].sender_id == "Alice"
async def test_exception_does_not_propagate(self, handler):
self._setup_handler(handler)
event = self._make_event({
"pubkey_prefix": "ab12",
"text": "hello",
"path_len": 255,
"sender_timestamp": int(time.time()),
})
with patch.object(handler, "_debug_decode_message_path", side_effect=RuntimeError("boom")):
# Should not raise
await handler.handle_contact_message(event)
handler.logger.error.assert_called()
# ---------------------------------------------------------------------------
# handle_channel_message
# ---------------------------------------------------------------------------
class TestHandleChannelMessage:
"""Tests for MessageHandler.handle_channel_message()."""
def _setup_handler(self, handler):
handler.logger = Mock()
handler.bot.meshcore = Mock()
handler.bot.meshcore.contacts = {}
handler.bot.channel_manager = Mock()
handler.bot.channel_manager.get_channel_name = Mock(return_value="general")
handler.bot.translator = None
handler.bot.mesh_graph = None
handler.recent_rf_data = []
handler.enhanced_correlation = False
def _make_event(self, payload):
event = Mock()
event.payload = payload
return event
async def test_no_payload_returns_early(self, handler):
self._setup_handler(handler)
event = Mock(spec=[])
await handler.handle_channel_message(event)
handler.logger.warning.assert_called()
async def test_payload_none_returns_early(self, handler):
self._setup_handler(handler)
event = Mock()
event.payload = None
await handler.handle_channel_message(event)
handler.logger.warning.assert_called()
async def test_basic_channel_message_calls_process_message(self, handler):
self._setup_handler(handler)
handler.bot.connection_time = None
event = self._make_event({
"channel_idx": 0,
"text": "ALICE: hello world",
"path_len": 255,
"sender_timestamp": int(time.time()),
})
with patch.object(handler, "process_message", new_callable=AsyncMock) as mock_pm:
with patch.object(handler, "_debug_decode_message_path", new_callable=AsyncMock):
with patch.object(handler, "_debug_decode_packet_for_message", new_callable=AsyncMock):
await handler.handle_channel_message(event)
mock_pm.assert_called_once()
async def test_sender_extracted_from_text(self, handler):
self._setup_handler(handler)
handler.bot.connection_time = None
captured = {}
async def capture(msg):
captured["msg"] = msg
event = self._make_event({
"channel_idx": 0,
"text": "BOB: hi there",
"path_len": 0,
"sender_timestamp": int(time.time()),
})
with patch.object(handler, "process_message", side_effect=capture):
with patch.object(handler, "_debug_decode_message_path", new_callable=AsyncMock):
with patch.object(handler, "_debug_decode_packet_for_message", new_callable=AsyncMock):
await handler.handle_channel_message(event)
assert captured["msg"].sender_id == "BOB"
assert captured["msg"].content == "hi there"
async def test_text_without_colon_uses_full_text(self, handler):
self._setup_handler(handler)
handler.bot.connection_time = None
captured = {}
async def capture(msg):
captured["msg"] = msg
event = self._make_event({
"channel_idx": 0,
"text": "no colon here",
"path_len": 0,
"sender_timestamp": int(time.time()),
})
with patch.object(handler, "process_message", side_effect=capture):
with patch.object(handler, "_debug_decode_message_path", new_callable=AsyncMock):
with patch.object(handler, "_debug_decode_packet_for_message", new_callable=AsyncMock):
await handler.handle_channel_message(event)
assert captured["msg"].content == "no colon here"
async def test_old_cached_message_not_processed(self, handler):
self._setup_handler(handler)
handler.bot.connection_time = time.time()
old_ts = int(time.time()) - 3600
event = self._make_event({
"channel_idx": 0,
"text": "CAROL: old msg",
"path_len": 0,
"sender_timestamp": old_ts,
})
with patch.object(handler, "process_message", new_callable=AsyncMock) as mock_pm:
with patch.object(handler, "_debug_decode_message_path", new_callable=AsyncMock):
with patch.object(handler, "_debug_decode_packet_for_message", new_callable=AsyncMock):
await handler.handle_channel_message(event)
mock_pm.assert_not_called()
async def test_snr_from_payload(self, handler):
self._setup_handler(handler)
handler.bot.connection_time = None
captured = {}
async def capture(msg):
captured["msg"] = msg
event = self._make_event({
"channel_idx": 0,
"text": "DAN: test",
"path_len": 0,
"sender_timestamp": int(time.time()),
"SNR": 9,
"RSSI": -85,
})
with patch.object(handler, "process_message", side_effect=capture):
with patch.object(handler, "_debug_decode_message_path", new_callable=AsyncMock):
with patch.object(handler, "_debug_decode_packet_for_message", new_callable=AsyncMock):
await handler.handle_channel_message(event)
assert captured["msg"].snr == 9
assert captured["msg"].rssi == -85
async def test_channel_name_set_on_message(self, handler):
self._setup_handler(handler)
handler.bot.connection_time = None
handler.bot.channel_manager.get_channel_name = Mock(return_value="emergency")
captured = {}
async def capture(msg):
captured["msg"] = msg
event = self._make_event({
"channel_idx": 2,
"text": "EVE: help",
"path_len": 0,
"sender_timestamp": int(time.time()),
})
with patch.object(handler, "process_message", side_effect=capture):
with patch.object(handler, "_debug_decode_message_path", new_callable=AsyncMock):
with patch.object(handler, "_debug_decode_packet_for_message", new_callable=AsyncMock):
await handler.handle_channel_message(event)
assert captured["msg"].channel == "emergency"
assert captured["msg"].is_dm is False
async def test_exception_does_not_propagate(self, handler):
self._setup_handler(handler)
event = self._make_event({
"channel_idx": 0,
"text": "FRANK: crash",
"path_len": 0,
"sender_timestamp": int(time.time()),
})
with patch.object(handler, "_debug_decode_message_path", side_effect=RuntimeError("boom")):
await handler.handle_channel_message(event)
handler.logger.error.assert_called()
-1
View File
@@ -1,6 +1,5 @@
"""Tests for modules/models.py — MeshMessage dataclass."""
import pytest
from modules.models import MeshMessage
+4 -3
View File
@@ -1,11 +1,12 @@
"""Tests for modules.plugin_loader."""
import pytest
from pathlib import Path
from unittest.mock import Mock, MagicMock, AsyncMock
from unittest.mock import AsyncMock, MagicMock, Mock
import pytest
from modules.plugin_loader import PluginLoader
from modules.commands.base_command import BaseCommand
from modules.plugin_loader import PluginLoader
@pytest.fixture
+7 -9
View File
@@ -60,10 +60,9 @@ class TestProfanityFilterWithLibrary:
def _reset_module_state(self):
"""Ensure the profanity module is initialized for these tests."""
import modules.profanity_filter as pf
if pf._profanity_available:
if not pf._profanity_initialized:
pf.profanity.load_censor_words()
pf._profanity_initialized = True
if pf._profanity_available and not pf._profanity_initialized:
pf.profanity.load_censor_words()
pf._profanity_initialized = True
yield
def test_censor_replaces_profanity_when_library_available(self):
@@ -136,11 +135,10 @@ class TestProfanityFilterFallbackWhenLibraryUnavailable:
def test_censor_logs_warning_once_when_library_unavailable(self):
import modules.profanity_filter as pf
logger = Mock()
with patch.object(pf, "_profanity_available", False):
with patch.object(pf, "_warned_unavailable", False):
censor("hello", logger=logger)
logger.warning.assert_called_once()
assert "better-profanity" in logger.warning.call_args[0][0]
with patch.object(pf, "_profanity_available", False), patch.object(pf, "_warned_unavailable", False):
censor("hello", logger=logger)
logger.warning.assert_called_once()
assert "better-profanity" in logger.warning.call_args[0][0]
def test_hate_symbol_still_detected_and_censored_when_library_unavailable(self):
"""Hate symbols (e.g. swastika) are detected and replaced even when better_profanity is not installed."""
+273 -2
View File
@@ -1,8 +1,9 @@
"""Tests for RepeaterManager pure logic (no network, no geocoding)."""
import pytest
import configparser
from unittest.mock import Mock, MagicMock, patch
from unittest.mock import AsyncMock, Mock, patch
import pytest
from modules.repeater_manager import RepeaterManager
@@ -215,3 +216,273 @@ class TestIsInAcl:
bot.config.set("Bot", "auto_manage_contacts", "device")
rm2 = RepeaterManager(bot)
assert rm2.auto_purge_enabled is True
# ---------------------------------------------------------------------------
# _should_geocode_location
# ---------------------------------------------------------------------------
class TestShouldGeocodeLocation:
"""Tests for RepeaterManager._should_geocode_location()."""
def _loc(self, lat=47.6, lon=-122.3, state=None, country=None, city=None):
return {"latitude": lat, "longitude": lon, "state": state, "country": country, "city": city}
def test_no_existing_data_with_coords_returns_true(self, rm):
loc = self._loc(lat=47.6, lon=-122.3)
should, _ = rm._should_geocode_location(loc, existing_data=None)
assert should is True
def test_no_existing_data_zero_coords_returns_false(self, rm):
loc = self._loc(lat=0.0, lon=0.0)
should, _ = rm._should_geocode_location(loc, existing_data=None)
assert should is False
def test_no_existing_data_no_coords_returns_false(self, rm):
loc = self._loc(lat=None, lon=None)
should, _ = rm._should_geocode_location(loc, existing_data=None)
assert should is False
def test_no_existing_data_all_fields_present_returns_false(self, rm):
loc = self._loc(lat=47.6, lon=-122.3, state="WA", country="US", city="Seattle")
should, _ = rm._should_geocode_location(loc, existing_data=None)
assert should is False
def test_existing_data_same_coords_sufficient_loc_no_geocode(self, rm):
loc = self._loc(lat=47.6, lon=-122.3)
existing = {"latitude": 47.6, "longitude": -122.3, "state": "WA", "country": "US", "city": "Seattle"}
should, updated = rm._should_geocode_location(loc, existing_data=existing)
assert should is False
assert updated["state"] == "WA"
assert updated["city"] == "Seattle"
def test_existing_data_moved_triggers_geocode(self, rm):
loc = self._loc(lat=48.0, lon=-122.0) # moved > 0.001 degrees
existing = {"latitude": 47.6, "longitude": -122.3, "state": "WA", "country": "US", "city": "Seattle"}
should, _ = rm._should_geocode_location(loc, existing_data=existing)
assert should is True
def test_existing_data_missing_city_triggers_geocode(self, rm):
loc = self._loc(lat=47.6, lon=-122.3)
existing = {"latitude": 47.6, "longitude": -122.3, "state": "WA", "country": "US", "city": None}
should, _ = rm._should_geocode_location(loc, existing_data=existing)
assert should is True
def test_existing_data_no_coords_in_new_data_keeps_existing(self, rm):
loc = self._loc(lat=None, lon=None)
existing = {"latitude": 47.6, "longitude": -122.3, "state": "WA", "country": "US", "city": "Seattle"}
should, updated = rm._should_geocode_location(loc, existing_data=existing)
assert should is False
assert updated["state"] == "WA"
def test_packet_hash_cache_hit_skips_geocode(self, rm):
import time
loc = self._loc(lat=47.6, lon=-122.3)
packet_hash = "abcdef1234567890"
# Pre-seed the cache
rm.geocoding_cache[packet_hash] = time.time()
should, _ = rm._should_geocode_location(loc, existing_data=None, packet_hash=packet_hash)
assert should is False
def test_default_packet_hash_not_cached(self, rm):
loc = self._loc(lat=47.6, lon=-122.3)
# Default/invalid hash should never match cache
should, _ = rm._should_geocode_location(loc, existing_data=None, packet_hash="0000000000000000")
assert should is True # No cache hit, coords valid → should geocode
def test_expired_cache_entry_removed(self, rm):
import time
loc = self._loc(lat=47.6, lon=-122.3)
old_hash = "oldpackethash1234"
# Pre-seed with expired entry
rm.geocoding_cache[old_hash] = time.time() - rm.geocoding_cache_window - 10
rm._should_geocode_location(loc, existing_data=None)
assert old_hash not in rm.geocoding_cache
# ---------------------------------------------------------------------------
# cleanup_repeater_retention
# ---------------------------------------------------------------------------
class TestCleanupRepeaterRetention:
def test_runs_without_error_on_empty_db(self, rm):
# Tables may not exist yet; should not raise
try:
rm.cleanup_repeater_retention(daily_stats_days=30, observed_paths_days=30)
except Exception:
pass # Some tables may not exist in test DB; that's OK
def test_does_not_raise_when_db_raises(self, rm):
from unittest.mock import patch as _patch
with _patch.object(rm.db_manager, "execute_update", side_effect=Exception("db error")):
rm.cleanup_repeater_retention() # Should not raise
rm.logger.error.assert_called()
# ---------------------------------------------------------------------------
# geocoding cache delegation
# ---------------------------------------------------------------------------
class TestGeocodingCacheDelegation:
def test_get_cached_geocoding_delegates(self, rm):
rm.db_manager.get_cached_geocoding = Mock(return_value=(47.6, -122.3))
result = rm.get_cached_geocoding("Seattle, WA")
assert result == (47.6, -122.3)
rm.db_manager.get_cached_geocoding.assert_called_once_with("Seattle, WA")
def test_cache_geocoding_delegates(self, rm):
rm.db_manager.cache_geocoding = Mock()
rm.cache_geocoding("Seattle, WA", 47.6, -122.3)
rm.db_manager.cache_geocoding.assert_called_once_with("Seattle, WA", 47.6, -122.3, 720)
def test_cleanup_geocoding_cache_delegates(self, rm):
rm.db_manager.cleanup_geocoding_cache = Mock()
rm.cleanup_geocoding_cache()
rm.db_manager.cleanup_geocoding_cache.assert_called_once()
# ---------------------------------------------------------------------------
# get_complete_contact_database (async)
# ---------------------------------------------------------------------------
class TestGetCompleteContactDatabase:
async def test_returns_empty_list_on_db_error(self, rm):
rm.db_manager.execute_query = Mock(side_effect=Exception("db fail"))
result = await rm.get_complete_contact_database()
assert result == []
async def test_returns_all_results_without_filter(self, rm):
rm.db_manager.execute_query = Mock(return_value=[
{"public_key": "aabb", "name": "Node1", "role": "repeater"},
])
result = await rm.get_complete_contact_database()
assert len(result) == 1
assert result[0]["name"] == "Node1"
async def test_with_role_filter(self, rm):
rm.db_manager.execute_query = Mock(return_value=[])
await rm.get_complete_contact_database(role_filter="repeater")
call_args = rm.db_manager.execute_query.call_args
assert "repeater" in str(call_args)
async def test_not_include_historical(self, rm):
rm.db_manager.execute_query = Mock(return_value=[])
await rm.get_complete_contact_database(include_historical=False)
call_args = rm.db_manager.execute_query.call_args
assert "is_currently_tracked" in str(call_args)
async def test_not_include_historical_with_role(self, rm):
rm.db_manager.execute_query = Mock(return_value=[])
await rm.get_complete_contact_database(role_filter="companion", include_historical=False)
call_args = rm.db_manager.execute_query.call_args
assert "is_currently_tracked" in str(call_args)
# ---------------------------------------------------------------------------
# get_contact_statistics (async)
# ---------------------------------------------------------------------------
class TestGetContactStatistics:
async def test_returns_empty_dict_on_error(self, rm):
rm.db_manager.execute_query = Mock(side_effect=Exception("fail"))
result = await rm.get_contact_statistics()
assert result == {}
async def test_returns_stats_structure(self, rm):
rm.db_manager.execute_query = Mock(side_effect=[
[{"count": 42}], # total_heard
[{"count": 10}], # currently_tracked
[{"count": 5}], # recent_activity
[{"role": "repeater", "count": 3}, {"role": "companion", "count": 39}], # by_role
[{"device_type": "Repeater", "count": 3}], # by_type
])
result = await rm.get_contact_statistics()
assert result["total_heard"] == 42
assert result["currently_tracked"] == 10
assert result["recent_activity"] == 5
assert result["by_role"]["repeater"] == 3
async def test_returns_zeros_on_empty_db(self, rm):
rm.db_manager.execute_query = Mock(return_value=[])
result = await rm.get_contact_statistics()
assert result.get("total_heard", 0) == 0
# ---------------------------------------------------------------------------
# get_contacts_by_role convenience wrappers (async)
# ---------------------------------------------------------------------------
class TestGetContactsByRole:
async def test_get_repeater_devices_combines_roles(self, rm):
async def fake_db(role_filter=None, include_historical=True):
if role_filter == "repeater":
return [{"name": "RPT1"}]
elif role_filter == "roomserver":
return [{"name": "RS1"}]
return []
with patch.object(rm, "get_complete_contact_database", side_effect=fake_db):
result = await rm.get_repeater_devices()
assert len(result) == 2
async def test_get_companion_contacts(self, rm):
with patch.object(rm, "get_complete_contact_database", return_value=[{"name": "Alice"}]) as mock_db:
result = await rm.get_companion_contacts()
mock_db.assert_called_once_with(role_filter="companion", include_historical=True)
assert result[0]["name"] == "Alice"
async def test_get_sensor_devices(self, rm):
with patch.object(rm, "get_complete_contact_database", return_value=[]) as mock_db:
await rm.get_sensor_devices()
mock_db.assert_called_once_with(role_filter="sensor", include_historical=True)
async def test_get_gateway_devices(self, rm):
with patch.object(rm, "get_complete_contact_database", return_value=[]) as mock_db:
await rm.get_gateway_devices()
mock_db.assert_called_once_with(role_filter="gateway", include_historical=True)
async def test_get_bot_devices(self, rm):
with patch.object(rm, "get_complete_contact_database", return_value=[]) as mock_db:
await rm.get_bot_devices()
mock_db.assert_called_once_with(role_filter="bot", include_historical=True)
# ---------------------------------------------------------------------------
# check_and_auto_purge (async)
# ---------------------------------------------------------------------------
class TestCheckAndAutoPurge:
async def test_returns_false_when_disabled(self, rm):
rm.auto_purge_enabled = False
result = await rm.check_and_auto_purge()
assert result is False
async def test_returns_false_when_below_threshold(self, rm):
rm.auto_purge_enabled = True
rm.auto_purge_threshold = 280
rm.bot.meshcore = Mock()
rm.bot.meshcore.contacts = {str(i): {} for i in range(100)} # 100 contacts
result = await rm.check_and_auto_purge()
assert result is False
async def test_triggers_purge_when_above_threshold(self, rm):
rm.auto_purge_enabled = True
rm.auto_purge_threshold = 10
rm.bot.meshcore = Mock()
rm.bot.meshcore.contacts = {str(i): {} for i in range(15)} # 15 > threshold
with patch.object(rm, "_auto_purge_repeaters", new_callable=AsyncMock, return_value=True) as mock_purge:
result = await rm.check_and_auto_purge()
mock_purge.assert_called_once()
assert result is True
async def test_returns_false_on_exception(self, rm):
rm.auto_purge_enabled = True
rm.bot.meshcore = Mock(side_effect=Exception("fail"))
result = await rm.check_and_auto_purge()
assert result is False
+5 -5
View File
@@ -1,16 +1,16 @@
"""Tests for modules.security_utils."""
import pytest
from pathlib import Path
from unittest.mock import patch
import pytest
from modules.security_utils import (
validate_pubkey_format,
validate_safe_path,
validate_external_url,
sanitize_input,
validate_api_key_format,
validate_external_url,
validate_port_number,
validate_pubkey_format,
validate_safe_path,
)
+3 -4
View File
@@ -1,14 +1,13 @@
"""Tests for modules.service_plugin_loader."""
import pytest
import configparser
from pathlib import Path
from unittest.mock import Mock, MagicMock
from unittest.mock import MagicMock, Mock
import pytest
from modules.service_plugin_loader import ServicePluginLoader
from modules.service_plugins.base_service import BaseServicePlugin
# Minimal local service source (valid BaseServicePlugin subclass)
_LOCAL_SERVICE_SOURCE = '''
from modules.service_plugins.base_service import BaseServicePlugin
+3 -2
View File
@@ -1,8 +1,9 @@
"""Tests for modules/transmission_tracker.py."""
import time
from unittest.mock import Mock
import pytest
from unittest.mock import Mock, MagicMock
from modules.transmission_tracker import TransmissionRecord, TransmissionTracker
@@ -71,7 +72,7 @@ class TestRecordTransmission:
def test_multiple_records_same_second(self, tracker):
rec1 = tracker.record_transmission("a", "ch", "channel")
rec2 = tracker.record_transmission("b", "ch", "channel")
key = int(rec1.timestamp)
int(rec1.timestamp)
# Both records should be in the same (or nearby) bucket
assert rec1 in tracker.pending_transmissions.get(int(rec1.timestamp), [])
assert rec2 in tracker.pending_transmissions.get(int(rec2.timestamp), [])
+336 -8
View File
@@ -1,18 +1,27 @@
"""Tests for modules.utils."""
import pytest
from unittest.mock import MagicMock
import configparser
from pathlib import Path
from unittest.mock import MagicMock, Mock, patch
from modules.utils import (
abbreviate_location,
truncate_string,
decode_escape_sequences,
parse_location_string,
calculate_distance,
format_elapsed_display,
parse_path_string,
decode_path_len_byte,
calculate_packet_hash,
calculate_path_distances,
check_internet_connectivity,
decode_escape_sequences,
decode_path_len_byte,
format_elapsed_display,
format_keyword_response_with_placeholders,
format_location_for_display,
get_config_timezone,
get_major_city_queries,
is_valid_timezone,
parse_location_string,
parse_path_string,
resolve_path,
truncate_string,
)
@@ -262,6 +271,325 @@ class TestParsePathString:
assert parse_path_string("015fab", prefix_hex_chars=2) == ["01", "5F", "AB"]
# ---------------------------------------------------------------------------
# is_valid_timezone
# ---------------------------------------------------------------------------
class TestIsValidTimezone:
"""Tests for is_valid_timezone()."""
def test_valid_utc(self):
assert is_valid_timezone("UTC") is True
def test_valid_us_eastern(self):
assert is_valid_timezone("America/New_York") is True
def test_valid_europe_london(self):
assert is_valid_timezone("Europe/London") is True
def test_invalid_returns_false(self):
assert is_valid_timezone("Not/A/Timezone") is False
def test_empty_string_returns_false(self):
assert is_valid_timezone("") is False
def test_whitespace_only_returns_false(self):
assert is_valid_timezone(" ") is False
def test_strips_whitespace_before_checking(self):
assert is_valid_timezone(" UTC ") is True
# ---------------------------------------------------------------------------
# get_config_timezone
# ---------------------------------------------------------------------------
class TestGetConfigTimezone:
"""Tests for get_config_timezone()."""
def _config(self, tz_value=""):
cfg = configparser.ConfigParser()
cfg.add_section("Bot")
if tz_value:
cfg.set("Bot", "timezone", tz_value)
return cfg
def test_valid_timezone_returned(self):
cfg = self._config("UTC")
tz, iana = get_config_timezone(cfg)
assert iana == "UTC"
assert tz is not None
def test_invalid_timezone_falls_back_to_utc_iana(self):
cfg = self._config("Not/Valid")
_, iana = get_config_timezone(cfg)
assert iana == "UTC"
def test_empty_timezone_falls_back(self):
cfg = self._config("")
_, iana = get_config_timezone(cfg)
assert iana == "UTC"
def test_invalid_timezone_logs_warning_when_logger_provided(self):
cfg = self._config("Bad/Zone")
logger = Mock()
get_config_timezone(cfg, logger)
logger.warning.assert_called_once()
def test_no_logger_no_crash_on_invalid(self):
cfg = self._config("Bad/Zone")
_, iana = get_config_timezone(cfg, None)
assert iana == "UTC"
# ---------------------------------------------------------------------------
# format_location_for_display
# ---------------------------------------------------------------------------
class TestFormatLocationForDisplay:
"""Tests for format_location_for_display()."""
def test_none_city_returns_none(self):
assert format_location_for_display(None) is None
def test_empty_city_returns_none(self):
assert format_location_for_display("") is None
def test_city_only(self):
result = format_location_for_display("Seattle", max_length=50)
assert "Seattle" in result
def test_city_and_state(self):
result = format_location_for_display("Seattle", "WA", max_length=50)
assert "Seattle" in result
assert "WA" in result
def test_state_not_duplicated_when_same_as_city(self):
result = format_location_for_display("Seattle", "Seattle", max_length=50)
# "Seattle" should not appear twice as comma-joined
assert result.count("Seattle") == 1
def test_respects_max_length(self):
result = format_location_for_display("Very Long City Name That Goes On", "State", max_length=15)
assert len(result) <= 15
# ---------------------------------------------------------------------------
# get_major_city_queries
# ---------------------------------------------------------------------------
class TestGetMajorCityQueries:
"""Tests for get_major_city_queries()."""
def test_known_city_returns_list(self):
result = get_major_city_queries("seattle")
assert isinstance(result, list)
assert len(result) > 0
assert any("Seattle" in q for q in result)
def test_unknown_city_returns_empty(self):
result = get_major_city_queries("tinyunknownvillage")
assert result == []
def test_case_insensitive(self):
assert get_major_city_queries("Seattle") == get_major_city_queries("seattle")
def test_new_york_returns_multiple_queries(self):
result = get_major_city_queries("new york")
assert len(result) >= 1
def test_portland_returns_multiple_queries(self):
# Portland has OR and ME variants
result = get_major_city_queries("portland")
assert len(result) >= 2
# ---------------------------------------------------------------------------
# resolve_path
# ---------------------------------------------------------------------------
class TestResolvePath:
"""Tests for resolve_path()."""
def test_absolute_path_unchanged(self):
result = resolve_path("/var/lib/bot/data.db", "/opt/bot")
assert result == "/var/lib/bot/data.db"
def test_relative_path_resolved_to_base_dir(self):
result = resolve_path("data.db", "/opt/bot")
assert result == "/opt/bot/data.db"
def test_path_object_input(self):
result = resolve_path(Path("data.db"), Path("/opt/bot"))
assert result == "/opt/bot/data.db"
def test_returns_string(self):
result = resolve_path("data.db", "/opt/bot")
assert isinstance(result, str)
def test_dot_base_dir_resolves_to_cwd(self):
import os
result = resolve_path("data.db", ".")
assert result == os.path.join(os.getcwd(), "data.db")
# ---------------------------------------------------------------------------
# check_internet_connectivity
# ---------------------------------------------------------------------------
class TestCheckInternetConnectivity:
"""Tests for check_internet_connectivity()."""
def test_returns_true_when_socket_connects(self):
mock_sock = Mock()
with patch("modules.utils.socket.socket") as mock_socket_cls:
mock_socket_cls.return_value = mock_sock
result = check_internet_connectivity(host="8.8.8.8", port=53, timeout=1.0)
assert result is True
mock_sock.connect.assert_called_once_with(("8.8.8.8", 53))
def test_returns_false_when_all_fail(self):
with patch("modules.utils.socket.socket") as mock_socket_cls, \
patch("modules.utils.urllib.request.urlopen") as mock_urlopen:
mock_socket_cls.return_value.connect.side_effect = OSError("refused")
mock_urlopen.side_effect = OSError("no net")
result = check_internet_connectivity(host="8.8.8.8", port=53, timeout=1.0)
assert result is False
def test_falls_back_to_http_when_socket_fails(self):
mock_response = Mock()
mock_response.close = Mock()
with patch("modules.utils.socket.socket") as mock_socket_cls, \
patch("modules.utils.urllib.request.urlopen") as mock_urlopen:
mock_socket_cls.return_value.connect.side_effect = OSError("refused")
mock_urlopen.return_value = mock_response
result = check_internet_connectivity(host="8.8.8.8", port=53, timeout=1.0)
assert result is True
# ---------------------------------------------------------------------------
# calculate_path_distances
# ---------------------------------------------------------------------------
class TestCalculatePathDistances:
"""Tests for calculate_path_distances()."""
def _bot(self):
bot = Mock()
bot.db_manager = Mock()
bot.prefix_hex_chars = 2
bot.logger = Mock()
return bot
def test_empty_path_returns_direct(self):
path_dist, fl_dist = calculate_path_distances(self._bot(), "")
assert "direct" in path_dist.lower()
def test_direct_path_returns_direct(self):
path_dist, fl_dist = calculate_path_distances(self._bot(), "Direct")
assert "direct" in path_dist.lower()
def test_no_db_manager_returns_unknown(self):
bot = Mock(spec=[]) # No db_manager attribute
path_dist, fl_dist = calculate_path_distances(bot, "01,5f")
assert "unknown" in path_dist.lower()
def test_single_node_returns_locally(self):
bot = self._bot()
with patch("modules.utils._get_node_location_from_db", return_value=None):
path_dist, fl_dist = calculate_path_distances(bot, "01")
assert "local" in path_dist.lower() or "1 hop" in path_dist.lower()
def test_two_nodes_with_locations_returns_distance(self):
bot = self._bot()
# Seattle and Portland coords
locations = [((47.6062, -122.3321), None), ((45.5152, -122.6784), None)]
with patch("modules.utils._get_node_location_from_db", side_effect=locations):
path_dist, fl_dist = calculate_path_distances(bot, "01,5f")
assert "km" in path_dist
assert "km" in fl_dist
def test_two_nodes_no_locations_returns_unknown(self):
bot = self._bot()
with patch("modules.utils._get_node_location_from_db", return_value=None):
path_dist, fl_dist = calculate_path_distances(bot, "01,5f")
assert "unknown" in path_dist.lower()
# ---------------------------------------------------------------------------
# format_keyword_response_with_placeholders
# ---------------------------------------------------------------------------
class TestFormatKeywordResponseWithPlaceholders:
"""Tests for format_keyword_response_with_placeholders()."""
def _bot(self):
bot = Mock()
bot.config = configparser.ConfigParser()
bot.config.add_section("Bot")
bot.config.set("Bot", "timezone", "UTC")
bot.db_manager = Mock()
bot.prefix_hex_chars = 2
bot.logger = Mock()
bot.translator = None
return bot
def _msg(self, **kwargs):
msg = Mock()
msg.sender_id = kwargs.get("sender_id", "Alice")
msg.path = kwargs.get("path", "01,5f")
msg.snr = kwargs.get("snr", 10)
msg.rssi = kwargs.get("rssi", -80)
msg.timestamp = kwargs.get("timestamp")
msg.hops = kwargs.get("hops")
return msg
def test_sender_placeholder(self):
bot = self._bot()
msg = self._msg(sender_id="Bob")
with patch("modules.utils.calculate_path_distances", return_value=("", "")):
result = format_keyword_response_with_placeholders("{sender}", msg, bot)
assert result == "Bob"
def test_no_message_uses_unknown_defaults(self):
bot = self._bot()
result = format_keyword_response_with_placeholders("{sender}", None, bot)
assert result == "Unknown"
def test_mesh_info_total_contacts(self):
bot = self._bot()
mesh_info = {"total_contacts": 42}
result = format_keyword_response_with_placeholders("{total_contacts}", None, bot, mesh_info)
assert result == "42"
def test_missing_placeholder_returns_format_string(self):
bot = self._bot()
# {nonexistent} is not in replacements -> KeyError -> returns raw string
result = format_keyword_response_with_placeholders("{nonexistent}", None, bot)
assert result == "{nonexistent}"
def test_hops_label_singular(self):
bot = self._bot()
msg = self._msg(hops=1)
with patch("modules.utils.calculate_path_distances", return_value=("", "")):
result = format_keyword_response_with_placeholders("{hops_label}", msg, bot)
assert result == "1 hop"
def test_hops_label_plural(self):
bot = self._bot()
msg = self._msg(hops=3)
with patch("modules.utils.calculate_path_distances", return_value=("", "")):
result = format_keyword_response_with_placeholders("{hops_label}", msg, bot)
assert result == "3 hops"
def test_connection_info_contains_snr_rssi(self):
bot = self._bot()
msg = self._msg(snr=12, rssi=-75)
with patch("modules.utils.calculate_path_distances", return_value=("", "")):
result = format_keyword_response_with_placeholders("{connection_info}", msg, bot)
assert "SNR" in result
assert "RSSI" in result
class TestCalculatePacketHashPathLength:
"""Tests that calculate_packet_hash uses decode_path_len_byte so multi-byte paths skip correctly."""
+120 -119
View File
@@ -3,23 +3,24 @@
Unit tests for MeshGraph class
"""
import pytest
from datetime import datetime, timedelta
import pytest
from modules.mesh_graph import MeshGraph
from tests.helpers import create_test_edge, populate_test_graph
@pytest.mark.unit
class TestMeshGraphEdgeManagement:
"""Test edge creation, updates, and retrieval."""
def test_add_edge_new(self, mesh_graph):
"""Test creating a new edge."""
from_prefix = "01"
to_prefix = "7e"
mesh_graph.add_edge(from_prefix, to_prefix)
edge = mesh_graph.get_edge(from_prefix, to_prefix)
assert edge is not None
assert edge['from_prefix'] == from_prefix.lower()
@@ -28,61 +29,61 @@ class TestMeshGraphEdgeManagement:
assert isinstance(edge['first_seen'], datetime)
assert isinstance(edge['last_seen'], datetime)
assert edge['first_seen'] == edge['last_seen']
def test_add_edge_update_existing(self, mesh_graph):
"""Test updating an existing edge."""
from_prefix = "01"
to_prefix = "7e"
# Create initial edge
mesh_graph.add_edge(from_prefix, to_prefix, hop_position=0)
first_seen = mesh_graph.edges[(from_prefix.lower(), to_prefix.lower())]['first_seen']
# Wait a tiny bit to ensure timestamps differ
import time
time.sleep(0.01)
# Update edge
mesh_graph.add_edge(from_prefix, to_prefix, hop_position=1)
edge = mesh_graph.get_edge(from_prefix, to_prefix)
assert edge['observation_count'] == 2
assert edge['first_seen'] == first_seen # First seen should not change
assert edge['last_seen'] > first_seen # Last seen should update
assert edge['avg_hop_position'] == 0.5 # (0 + 1) / 2
def test_add_edge_public_keys(self, mesh_graph):
"""Test edge creation and updates with public keys."""
from_prefix = "01"
to_prefix = "7e"
from_key = "0101010101010101010101010101010101010101010101010101010101010101"
to_key = "7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e"
# Create edge with public keys
mesh_graph.add_edge(from_prefix, to_prefix, from_public_key=from_key, to_public_key=to_key)
edge = mesh_graph.get_edge(from_prefix, to_prefix)
assert edge['from_public_key'] == from_key
assert edge['to_public_key'] == to_key
# Update with new public key
new_from_key = "0202020202020202020202020202020202020202020202020202020202020202"
mesh_graph.add_edge(from_prefix, to_prefix, from_public_key=new_from_key)
edge = mesh_graph.get_edge(from_prefix, to_prefix)
assert edge['from_public_key'] == new_from_key
assert edge['to_public_key'] == to_key # Should remain unchanged
def test_add_edge_hop_position(self, mesh_graph):
"""Test hop position tracking and weighted average calculation."""
from_prefix = "01"
to_prefix = "7e"
# Add edge multiple times with different hop positions
mesh_graph.add_edge(from_prefix, to_prefix, hop_position=0)
mesh_graph.add_edge(from_prefix, to_prefix, hop_position=1)
mesh_graph.add_edge(from_prefix, to_prefix, hop_position=2)
edge = mesh_graph.get_edge(from_prefix, to_prefix)
assert edge['observation_count'] == 3
# Weighted average: (0*0 + 1*1 + 2*2) / 3 = 5/3 ≈ 1.667
@@ -92,45 +93,45 @@ class TestMeshGraphEdgeManagement:
# After 2nd: ((0 * 1) + 1) / 2 = 0.5
# After 3rd: ((0.5 * 2) + 2) / 3 = (1 + 2) / 3 = 1.0
assert abs(edge['avg_hop_position'] - 1.0) < 0.01
def test_add_edge_geographic_distance(self, mesh_graph):
"""Test geographic distance storage and updates."""
from_prefix = "01"
to_prefix = "7e"
# Create edge with distance
mesh_graph.add_edge(from_prefix, to_prefix, geographic_distance=10.5)
edge = mesh_graph.get_edge(from_prefix, to_prefix)
assert edge['geographic_distance'] == 10.5
# Update distance
mesh_graph.add_edge(from_prefix, to_prefix, geographic_distance=12.3)
edge = mesh_graph.get_edge(from_prefix, to_prefix)
assert edge['geographic_distance'] == 12.3
def test_get_edge_existing(self, mesh_graph):
"""Test retrieving an existing edge."""
from_prefix = "01"
to_prefix = "7e"
mesh_graph.add_edge(from_prefix, to_prefix)
edge = mesh_graph.get_edge(from_prefix, to_prefix)
assert edge is not None
assert edge['from_prefix'] == from_prefix.lower()
assert edge['to_prefix'] == to_prefix.lower()
def test_get_edge_nonexistent(self, mesh_graph):
"""Test retrieving a non-existent edge returns None."""
edge = mesh_graph.get_edge("99", "aa")
assert edge is None
def test_get_edge_case_insensitive(self, mesh_graph):
"""Test that get_edge is case-insensitive."""
mesh_graph.add_edge("01", "7E")
# Try various case combinations
assert mesh_graph.get_edge("01", "7e") is not None
assert mesh_graph.get_edge("01", "7E") is not None
@@ -212,36 +213,36 @@ class TestMeshGraphPathValidation:
"""Test validation with sufficient observations."""
from_prefix = "01"
to_prefix = "7e"
# Create edge with recent timestamp and sufficient observations
mesh_graph.add_edge(from_prefix, to_prefix)
# Add more observations
for _ in range(4):
mesh_graph.add_edge(from_prefix, to_prefix)
is_valid, confidence = mesh_graph.validate_path_segment(from_prefix, to_prefix, min_observations=3)
assert is_valid is True
assert 0.0 <= confidence <= 1.0
assert confidence > 0.5 # Should have high confidence with recent data and 5 observations
def test_validate_path_segment_insufficient_observations(self, mesh_graph):
"""Test validation fails when observations < min_observations."""
from_prefix = "01"
to_prefix = "7e"
mesh_graph.add_edge(from_prefix, to_prefix) # Only 1 observation
is_valid, confidence = mesh_graph.validate_path_segment(from_prefix, to_prefix, min_observations=3)
assert is_valid is False
assert confidence == 0.0
def test_validate_path_segment_stale(self, mesh_graph):
"""Test confidence decay with old last_seen."""
from_prefix = "01"
to_prefix = "7e"
# Create edge with old timestamp
old_time = datetime.now() - timedelta(days=10) # 10 days ago
edge_key = (from_prefix.lower(), to_prefix.lower())
@@ -254,47 +255,47 @@ class TestMeshGraphPathValidation:
'avg_hop_position': None,
'geographic_distance': None
}
is_valid, confidence = mesh_graph.validate_path_segment(from_prefix, to_prefix, min_observations=1)
assert is_valid is True # Still valid (has observations)
assert confidence < 0.3 # But low confidence due to staleness (10 days = ~240 hours, well past 48h half-life)
def test_validate_path_segment_bidirectional(self, mesh_graph):
"""Test bidirectional edge bonus."""
from_prefix = "01"
to_prefix = "7e"
# Create forward edge
mesh_graph.add_edge(from_prefix, to_prefix)
for _ in range(4):
mesh_graph.add_edge(from_prefix, to_prefix)
# Validate without bidirectional check
is_valid, confidence_forward = mesh_graph.validate_path_segment(
from_prefix, to_prefix, min_observations=1, check_bidirectional=False
)
# Create reverse edge
mesh_graph.add_edge(to_prefix, from_prefix)
for _ in range(4):
mesh_graph.add_edge(to_prefix, from_prefix)
# Validate with bidirectional check
is_valid_bidir, confidence_bidir = mesh_graph.validate_path_segment(
from_prefix, to_prefix, min_observations=1, check_bidirectional=True
)
assert is_valid is True
assert is_valid_bidir is True
assert confidence_bidir > confidence_forward # Bidirectional should have higher confidence
assert confidence_bidir <= min(1.0, confidence_forward + 0.15) # Bonus is +0.15 max
def test_validate_path_segment_confidence_calculation(self, mesh_graph):
"""Test confidence calculation components."""
from_prefix = "01"
to_prefix = "7e"
# Create edge with known values
now = datetime.now()
edge_key = (from_prefix.lower(), to_prefix.lower())
@@ -307,48 +308,48 @@ class TestMeshGraphPathValidation:
'avg_hop_position': None,
'geographic_distance': None
}
is_valid, confidence = mesh_graph.validate_path_segment(from_prefix, to_prefix, min_observations=1)
assert is_valid is True
# With 20 observations and very recent timestamp, confidence should be very high
assert confidence > 0.8
def test_validate_path_complete(self, mesh_graph):
"""Test full path validation."""
# Create a path: 01 -> 7e -> 86 -> e0
path_nodes = ['01', '7e', '86', 'e0']
# Add edges for the path
for i in range(len(path_nodes) - 1):
mesh_graph.add_edge(path_nodes[i], path_nodes[i + 1])
# Add extra observations for some edges
if i < 2:
mesh_graph.add_edge(path_nodes[i], path_nodes[i + 1])
is_valid, avg_confidence = mesh_graph.validate_path(path_nodes, min_observations=1)
assert is_valid is True
assert 0.0 <= avg_confidence <= 1.0
def test_validate_path_invalid_segment(self, mesh_graph):
"""Test path validation fails when one segment is invalid."""
path_nodes = ['01', '7e', '86', 'e0']
# Add edges for most of the path
mesh_graph.add_edge('01', '7e')
mesh_graph.add_edge('7e', '86')
# Missing edge: 86 -> e0
is_valid, confidence = mesh_graph.validate_path(path_nodes, min_observations=1)
assert is_valid is False
assert confidence == 0.0
def test_validate_path_single_node(self, mesh_graph):
"""Test edge case: single node path is always valid."""
is_valid, confidence = mesh_graph.validate_path(['01'], min_observations=1)
assert is_valid is True
assert confidence == 1.0
@@ -356,103 +357,103 @@ class TestMeshGraphPathValidation:
@pytest.mark.unit
class TestMeshGraphCandidateScoring:
"""Test candidate scoring with various edge combinations."""
def test_get_candidate_score_no_edges(self, mesh_graph):
"""Test candidate with no graph edges returns 0.0."""
score = mesh_graph.get_candidate_score('99', '01', '7e', min_observations=1)
assert score == 0.0
def test_get_candidate_score_prev_edge_only(self, mesh_graph):
"""Test scoring with only previous edge."""
# Create edge from previous node to candidate
mesh_graph.add_edge('01', '7e')
for _ in range(4):
mesh_graph.add_edge('01', '7e') # 5 total observations
score = mesh_graph.get_candidate_score('7e', '01', None, min_observations=1)
assert score > 0.0
assert score <= 1.0
def test_get_candidate_score_next_edge_only(self, mesh_graph):
"""Test scoring with only next edge."""
# Create edge from candidate to next node
mesh_graph.add_edge('7e', '86')
for _ in range(4):
mesh_graph.add_edge('7e', '86') # 5 total observations
score = mesh_graph.get_candidate_score('7e', None, '86', min_observations=1)
assert score > 0.0
assert score <= 1.0
def test_get_candidate_score_both_edges(self, mesh_graph):
"""Test scoring with both prev and next edges."""
# Create both edges
mesh_graph.add_edge('01', '7e')
mesh_graph.add_edge('7e', '86')
score = mesh_graph.get_candidate_score('7e', '01', '86', min_observations=1)
assert score > 0.0
assert score <= 1.0
# Should be average of both edge confidences
def test_get_candidate_score_bidirectional(self, mesh_graph):
"""Test bidirectional edge checking increases score."""
# Create forward edge only
mesh_graph.add_edge('01', '7e')
for _ in range(4):
mesh_graph.add_edge('01', '7e')
score_forward = mesh_graph.get_candidate_score('7e', '01', None, min_observations=1, use_bidirectional=False)
# Create reverse edge (bidirectional)
mesh_graph.add_edge('7e', '01')
for _ in range(4):
mesh_graph.add_edge('7e', '01')
score_bidir = mesh_graph.get_candidate_score('7e', '01', None, min_observations=1, use_bidirectional=True)
assert score_bidir > score_forward # Bidirectional should score higher
def test_get_candidate_score_hop_position_match(self, mesh_graph):
"""Test hop position validation bonus."""
# Create edge with avg_hop_position = 1.0
mesh_graph.add_edge('01', '7e', hop_position=1)
mesh_graph.add_edge('01', '7e', hop_position=1) # Average stays at 1.0
# Score with matching hop position
score_match = mesh_graph.get_candidate_score('7e', '01', None, min_observations=1, hop_position=1, use_hop_position=True)
# Score with non-matching hop position
score_no_match = mesh_graph.get_candidate_score('7e', '01', None, min_observations=1, hop_position=5, use_hop_position=True)
assert score_match > score_no_match # Matching position should have bonus
def test_get_candidate_score_geographic_bonus(self, mesh_graph):
"""Test geographic distance data bonus."""
# Create edge without geographic distance
mesh_graph.add_edge('01', '7e')
score_no_geo = mesh_graph.get_candidate_score('7e', '01', None, min_observations=1)
# Update edge with geographic distance
mesh_graph.add_edge('01', '7e', geographic_distance=10.5)
score_with_geo = mesh_graph.get_candidate_score('7e', '01', None, min_observations=1)
assert score_with_geo > score_no_geo # Geographic data adds +0.05 bonus
def test_get_candidate_score_combined_features(self, mesh_graph):
"""Test scoring with all features enabled."""
# Create bidirectional edge with hop position and geographic distance
mesh_graph.add_edge('01', '7e', hop_position=1, geographic_distance=10.5)
mesh_graph.add_edge('7e', '01', hop_position=0, geographic_distance=10.5)
score = mesh_graph.get_candidate_score(
'7e', '01', '86', min_observations=1,
use_bidirectional=True, use_hop_position=True, hop_position=1
)
assert 0.0 <= score <= 1.0
# Should be high with all bonuses
@@ -460,68 +461,68 @@ class TestMeshGraphCandidateScoring:
@pytest.mark.unit
class TestMeshGraphMultiHop:
"""Test multi-hop path inference."""
def test_find_intermediate_nodes_2hop(self, mesh_graph):
"""Test finding intermediate node in 2-hop path."""
# Create 2-hop path: 01 -> 7e -> 86 (no direct 01 -> 86)
mesh_graph.add_edge('01', '7e')
for _ in range(4):
mesh_graph.add_edge('01', '7e')
mesh_graph.add_edge('7e', '86')
for _ in range(4):
mesh_graph.add_edge('7e', '86')
candidates = mesh_graph.find_intermediate_nodes('01', '86', min_observations=1, max_hops=2)
assert len(candidates) > 0
# Should find '7e' as intermediate
intermediate_prefixes = [c[0] for c in candidates]
assert '7e' in intermediate_prefixes
def test_find_intermediate_nodes_direct_edge(self, mesh_graph):
"""Test when direct edge exists, intermediate search still works."""
# Create direct edge
mesh_graph.add_edge('01', '86')
for _ in range(4):
mesh_graph.add_edge('01', '86')
# Also create 2-hop path
mesh_graph.add_edge('01', '7e')
mesh_graph.add_edge('7e', '86')
candidates = mesh_graph.find_intermediate_nodes('01', '86', min_observations=1, max_hops=2)
# Should still find intermediate nodes even though direct edge exists
# (the function skips direct edges but continues searching)
intermediate_prefixes = [c[0] for c in candidates]
[c[0] for c in candidates]
# May or may not include '7e' depending on implementation, but should not error
assert isinstance(candidates, list)
def test_find_intermediate_nodes_3hop(self, mesh_graph):
"""Test 3-hop path inference."""
# Create 3-hop path: 01 -> 7e -> 86 -> e0
mesh_graph.add_edge('01', '7e')
mesh_graph.add_edge('7e', '86')
mesh_graph.add_edge('86', 'e0')
candidates = mesh_graph.find_intermediate_nodes('01', 'e0', min_observations=1, max_hops=3)
# Should find intermediate nodes in 3-hop path
intermediate_prefixes = [c[0] for c in candidates]
[c[0] for c in candidates]
# Should find either '7e' or '86' or both
assert len(candidates) > 0
def test_find_intermediate_nodes_no_path(self, mesh_graph):
"""Test when no path exists returns empty list."""
# Create isolated edges that don't connect
mesh_graph.add_edge('01', '7e')
mesh_graph.add_edge('99', 'aa')
candidates = mesh_graph.find_intermediate_nodes('01', 'aa', min_observations=1, max_hops=2)
assert candidates == []
def test_find_intermediate_nodes_min_observations(self, mesh_graph):
"""Test filtering by min_observations."""
# Create path with one edge below threshold
@@ -529,76 +530,76 @@ class TestMeshGraphMultiHop:
mesh_graph.add_edge('7e', '86')
for _ in range(4):
mesh_graph.add_edge('7e', '86') # 5 observations
# With min_observations=3, should not find path through 7e
candidates = mesh_graph.find_intermediate_nodes('01', '86', min_observations=3, max_hops=2)
assert len(candidates) == 0 # 01->7e has only 1 observation, below threshold
@pytest.mark.unit
class TestMeshGraphPersistence:
"""Test edge persistence and write strategies."""
def test_write_strategy_immediate(self, mock_bot, test_db):
"""Test immediate write strategy."""
mock_bot.config.set('Path_Command', 'graph_write_strategy', 'immediate')
graph = MeshGraph(mock_bot)
graph.add_edge('01', '7e')
# Check database directly
results = test_db.execute_query('SELECT * FROM mesh_connections WHERE from_prefix = ? AND to_prefix = ?', ('01', '7e'))
assert len(results) == 1
assert results[0]['observation_count'] == 1
def test_write_strategy_batched(self, mock_bot, test_db):
"""Test batched write strategy."""
mock_bot.config.set('Path_Command', 'graph_write_strategy', 'batched')
mock_bot.config.set('Path_Command', 'graph_batch_max_pending', '5')
graph = MeshGraph(mock_bot)
# Add edges (should be batched)
for i in range(3):
graph.add_edge(f'0{i}', '7e')
# Edges should be in pending_updates, not yet in DB (unless max_pending reached)
# With 3 edges and max_pending=5, they should still be pending
assert len(graph.pending_updates) >= 0 # May have been flushed if max_pending reached
# Force flush
graph._flush_pending_updates_sync()
# Now should be in database
results = test_db.execute_query('SELECT * FROM mesh_connections')
assert len(results) == 3
def test_write_strategy_hybrid(self, mock_bot, test_db):
"""Test hybrid strategy (immediate for new, batched for updates)."""
mock_bot.config.set('Path_Command', 'graph_write_strategy', 'hybrid')
graph = MeshGraph(mock_bot)
# New edge should be written immediately
graph.add_edge('01', '7e')
results = test_db.execute_query('SELECT * FROM mesh_connections WHERE from_prefix = ? AND to_prefix = ?', ('01', '7e'))
assert len(results) == 1
# Update should be batched
graph.add_edge('01', '7e')
assert ('01', '7e') in graph.pending_updates
def test_load_from_database(self, mock_bot, test_db):
"""Test loading edges on initialization."""
# Pre-populate database with edges
test_db.execute_update('''
INSERT INTO mesh_connections
INSERT INTO mesh_connections
(from_prefix, to_prefix, observation_count, first_seen, last_seen)
VALUES (?, ?, ?, ?, ?)
''', ('01', '7e', 5, datetime.now().isoformat(), datetime.now().isoformat()))
# Create new graph instance (should load from DB)
graph = MeshGraph(mock_bot)
# Check that edge was loaded
edge = graph.get_edge('01', '7e')
assert edge is not None
+37 -37
View File
@@ -3,19 +3,19 @@
Unit tests for MeshGraph edge management
"""
import pytest
from datetime import datetime, timedelta
from tests.helpers import create_test_edge
import pytest
@pytest.mark.unit
class TestMeshGraphEdges:
"""Test MeshGraph edge management functionality."""
def test_add_new_edge(self, mesh_graph):
"""Test adding a new edge to the graph."""
mesh_graph.add_edge('01', '7e')
edge = mesh_graph.get_edge('01', '7e')
assert edge is not None
assert edge['from_prefix'] == '01'
@@ -23,55 +23,55 @@ class TestMeshGraphEdges:
assert edge['observation_count'] == 1
assert edge['first_seen'] is not None
assert edge['last_seen'] is not None
def test_add_edge_with_public_keys(self, mesh_graph):
"""Test adding edge with public keys."""
from_key = '0101010101010101010101010101010101010101010101010101010101010101'
to_key = '7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e'
mesh_graph.add_edge('01', '7e', from_public_key=from_key, to_public_key=to_key)
edge = mesh_graph.get_edge('01', '7e')
assert edge['from_public_key'] == from_key
assert edge['to_public_key'] == to_key
def test_add_edge_with_hop_position(self, mesh_graph):
"""Test adding edge with hop position."""
mesh_graph.add_edge('01', '7e', hop_position=2)
edge = mesh_graph.get_edge('01', '7e')
assert edge['avg_hop_position'] == 2.0
def test_add_edge_with_geographic_distance(self, mesh_graph):
"""Test adding edge with geographic distance."""
mesh_graph.add_edge('01', '7e', geographic_distance=15.5)
edge = mesh_graph.get_edge('01', '7e')
assert edge['geographic_distance'] == 15.5
def test_update_existing_edge(self, mesh_graph):
"""Test updating an existing edge increments observation count."""
mesh_graph.add_edge('01', '7e')
mesh_graph.add_edge('01', '7e')
edge = mesh_graph.get_edge('01', '7e')
assert edge['observation_count'] == 2
def test_update_edge_hop_position_average(self, mesh_graph):
"""Test that updating edge recalculates average hop position."""
mesh_graph.add_edge('01', '7e', hop_position=1)
mesh_graph.add_edge('01', '7e', hop_position=3)
edge = mesh_graph.get_edge('01', '7e')
# Average should be (1 + 3) / 2 = 2.0
assert edge['avg_hop_position'] == 2.0
# Add another observation
mesh_graph.add_edge('01', '7e', hop_position=2)
edge = mesh_graph.get_edge('01', '7e')
# Average should be (2.0 * 2 + 2) / 3 = 2.0
assert edge['avg_hop_position'] == 2.0
def test_update_edge_public_keys(self, mesh_graph):
"""Test that updating edge can add missing public keys."""
# Add edge without keys
@@ -79,95 +79,95 @@ class TestMeshGraphEdges:
edge = mesh_graph.get_edge('01', '7e')
assert edge['from_public_key'] is None
assert edge['to_public_key'] is None
# Update with keys
from_key = '0101010101010101010101010101010101010101010101010101010101010101'
to_key = '7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e7e'
mesh_graph.add_edge('01', '7e', from_public_key=from_key, to_public_key=to_key)
edge = mesh_graph.get_edge('01', '7e')
assert edge['from_public_key'] == from_key
assert edge['to_public_key'] == to_key
def test_update_edge_geographic_distance(self, mesh_graph):
"""Test that updating edge can update geographic distance."""
mesh_graph.add_edge('01', '7e', geographic_distance=10.0)
mesh_graph.add_edge('01', '7e', geographic_distance=15.5)
edge = mesh_graph.get_edge('01', '7e')
assert edge['geographic_distance'] == 15.5
def test_get_edge_nonexistent(self, mesh_graph):
"""Test getting a non-existent edge returns None."""
edge = mesh_graph.get_edge('01', '99')
assert edge is None
def test_has_edge(self, mesh_graph):
"""Test has_edge method."""
assert mesh_graph.has_edge('01', '7e') is False
mesh_graph.add_edge('01', '7e')
assert mesh_graph.has_edge('01', '7e') is True
assert mesh_graph.has_edge('7e', '01') is False # Direction matters
def test_prefix_normalization(self, mesh_graph):
"""Test that prefixes are normalized to lowercase and truncated."""
mesh_graph.add_edge('01AB', '7EFF')
# Should normalize to '01' and '7e'
assert mesh_graph.has_edge('01', '7e') is True
assert mesh_graph.has_edge('01ab', '7eff') is True
assert mesh_graph.has_edge('01AB', '7EFF') is True
def test_get_outgoing_edges(self, mesh_graph):
"""Test getting all outgoing edges from a node."""
mesh_graph.add_edge('01', '7e')
mesh_graph.add_edge('01', '86')
mesh_graph.add_edge('7e', '01') # Reverse direction
outgoing = mesh_graph.get_outgoing_edges('01')
assert len(outgoing) == 2
prefixes = {edge['to_prefix'] for edge in outgoing}
assert '7e' in prefixes
assert '86' in prefixes
def test_get_incoming_edges(self, mesh_graph):
"""Test getting all incoming edges to a node."""
mesh_graph.add_edge('01', '7e')
mesh_graph.add_edge('86', '7e')
mesh_graph.add_edge('7e', '01') # Reverse direction
incoming = mesh_graph.get_incoming_edges('7e')
assert len(incoming) == 2
prefixes = {edge['from_prefix'] for edge in incoming}
assert '01' in prefixes
assert '86' in prefixes
def test_empty_prefix_handling(self, mesh_graph):
"""Test that empty prefixes are ignored."""
initial_count = len(mesh_graph.edges)
mesh_graph.add_edge('', '7e')
mesh_graph.add_edge('01', '')
mesh_graph.add_edge('', '')
assert len(mesh_graph.edges) == initial_count
def test_edge_last_seen_updates(self, mesh_graph):
"""Test that last_seen timestamp updates on edge updates."""
first_time = datetime.now() - timedelta(seconds=5)
# Manually set first_seen to past time
mesh_graph.add_edge('01', '7e')
edge = mesh_graph.get_edge('01', '7e')
edge['first_seen'] = first_time
edge['last_seen'] = first_time
# Wait a moment and update
import time
time.sleep(0.1)
mesh_graph.add_edge('01', '7e')
edge = mesh_graph.get_edge('01', '7e')
assert edge['last_seen'] > first_time
assert edge['first_seen'] == first_time # First seen shouldn't change
+30 -31
View File
@@ -4,133 +4,132 @@ Unit tests for MeshGraph multi-hop inference
"""
import pytest
from tests.helpers import create_test_edge
@pytest.mark.unit
class TestMeshGraphMultiHop:
"""Test MeshGraph multi-hop inference functionality."""
def test_find_intermediate_nodes_direct_edge(self, mesh_graph):
"""Test that direct edges are not returned as intermediate nodes."""
mesh_graph.add_edge('01', '7e')
candidates = mesh_graph.find_intermediate_nodes('01', '7e')
# Direct edge exists, so no intermediate nodes should be returned
assert len(candidates) == 0
def test_find_intermediate_nodes_2hop_path(self, mesh_graph):
"""Test finding intermediate nodes in a 2-hop path."""
mesh_graph.add_edge('01', '7e')
mesh_graph.add_edge('7e', '86')
candidates = mesh_graph.find_intermediate_nodes('01', '86')
assert len(candidates) > 0
assert candidates[0][0] == '7e' # Intermediate node
assert candidates[0][1] > 0.0 # Has score
def test_find_intermediate_nodes_3hop_path(self, mesh_graph):
"""Test finding intermediate nodes in a 3-hop path."""
mesh_graph.add_edge('01', '7e')
mesh_graph.add_edge('7e', '86')
mesh_graph.add_edge('86', 'e0')
candidates = mesh_graph.find_intermediate_nodes('01', 'e0', max_hops=3)
assert len(candidates) > 0
# Should return '86' (the node before destination in 3-hop path)
assert candidates[0][0] == '86'
assert candidates[0][1] > 0.0
def test_find_intermediate_nodes_no_path(self, mesh_graph):
"""Test that no candidates are returned when no path exists."""
mesh_graph.add_edge('01', '7e')
mesh_graph.add_edge('86', 'e0') # Disconnected components
candidates = mesh_graph.find_intermediate_nodes('01', 'e0')
assert len(candidates) == 0
def test_find_intermediate_nodes_min_observations(self, mesh_graph):
"""Test that min_observations filter is applied."""
mesh_graph.add_edge('01', '7e') # Only 1 observation
mesh_graph.add_edge('7e', '86')
mesh_graph.add_edge('7e', '86') # Ensure 7e->86 has 2 observations
# With min_observations=3, should find no candidates (both edges need 3+)
candidates = mesh_graph.find_intermediate_nodes('01', '86', min_observations=3)
assert len(candidates) == 0
# Add more observations to both edges
mesh_graph.add_edge('01', '7e')
mesh_graph.add_edge('01', '7e') # Now 01->7e has 3 observations
mesh_graph.add_edge('7e', '86') # Now 7e->86 has 3 observations
candidates = mesh_graph.find_intermediate_nodes('01', '86', min_observations=3)
assert len(candidates) > 0
def test_find_intermediate_nodes_bidirectional_bonus(self, mesh_graph):
"""Test that bidirectional paths get higher scores."""
# Unidirectional path
mesh_graph.add_edge('01', '7e')
mesh_graph.add_edge('7e', '86')
candidates_uni = mesh_graph.find_intermediate_nodes('01', '86')
# Add reverse edges (bidirectional)
mesh_graph.add_edge('7e', '01')
mesh_graph.add_edge('86', '7e')
candidates_bi = mesh_graph.find_intermediate_nodes('01', '86')
assert len(candidates_bi) > 0
assert candidates_bi[0][1] > candidates_uni[0][1]
def test_find_intermediate_nodes_multiple_candidates(self, mesh_graph):
"""Test finding multiple intermediate node candidates."""
# Path 1: 01 -> 7e -> 86
mesh_graph.add_edge('01', '7e')
mesh_graph.add_edge('7e', '86')
# Path 2: 01 -> 7a -> 86
mesh_graph.add_edge('01', '7a')
mesh_graph.add_edge('7a', '86')
candidates = mesh_graph.find_intermediate_nodes('01', '86')
assert len(candidates) >= 2
# Should be sorted by score (highest first)
scores = [c[1] for c in candidates]
assert scores == sorted(scores, reverse=True)
def test_find_intermediate_nodes_3hop_score_reduction(self, mesh_graph):
"""Test that 3-hop paths have reduced scores."""
# 2-hop path
mesh_graph.add_edge('01', '7e')
mesh_graph.add_edge('7e', '86')
candidates_2hop = mesh_graph.find_intermediate_nodes('01', '86', max_hops=2)
# 3-hop path
mesh_graph.add_edge('01', '7e')
mesh_graph.add_edge('7e', '86')
mesh_graph.add_edge('86', 'e0')
candidates_3hop = mesh_graph.find_intermediate_nodes('01', 'e0', max_hops=3)
# 3-hop should have lower score due to 0.8 multiplier
if len(candidates_2hop) > 0 and len(candidates_3hop) > 0:
# Both paths exist, 3-hop should be lower
assert candidates_3hop[0][1] < candidates_2hop[0][1]
def test_find_intermediate_nodes_max_hops_limit(self, mesh_graph):
"""Test that max_hops parameter limits search depth."""
mesh_graph.add_edge('01', '7e')
mesh_graph.add_edge('7e', '86')
mesh_graph.add_edge('86', 'e0')
# With max_hops=2, should not find 3-hop path
candidates = mesh_graph.find_intermediate_nodes('01', 'e0', max_hops=2)
assert len(candidates) == 0
# With max_hops=3, should find it
candidates = mesh_graph.find_intermediate_nodes('01', 'e0', max_hops=3)
assert len(candidates) > 0
def test_find_intermediate_nodes_weakest_link_scoring(self, mesh_graph):
"""Test that path score uses weakest link (minimum confidence)."""
# Create path where one edge has low confidence
@@ -138,20 +137,20 @@ class TestMeshGraphMultiHop:
mesh_graph.add_edge('7e', '86')
mesh_graph.add_edge('7e', '86')
mesh_graph.add_edge('7e', '86') # 3 observations
candidates = mesh_graph.find_intermediate_nodes('01', '86')
assert len(candidates) > 0
# Score should be limited by the weaker edge (01->7e)
score = candidates[0][1]
assert score < 1.0 # Should be less than perfect due to weak link
def test_find_intermediate_nodes_self_loop_prevention(self, mesh_graph):
"""Test that paths don't loop back to source."""
mesh_graph.add_edge('01', '7e')
mesh_graph.add_edge('7e', '01') # Loop back
mesh_graph.add_edge('7e', '86')
candidates = mesh_graph.find_intermediate_nodes('01', '86')
# Should still find valid path through 7e
assert len(candidates) > 0
+3 -3
View File
@@ -10,15 +10,15 @@ Covers the optimizations added for low-memory devices (Raspberry Pi Zero 2 W):
- capture_enabled flag (graph_capture_enabled config setting)
"""
import time
import sqlite3
import pytest
import time
from contextlib import closing
from datetime import datetime, timedelta
from unittest.mock import MagicMock
from modules.mesh_graph import MeshGraph
import pytest
from modules.mesh_graph import MeshGraph
# ---------------------------------------------------------------------------
# Helper
+27 -28
View File
@@ -4,111 +4,110 @@ Unit tests for MeshGraph candidate scoring
"""
import pytest
from tests.helpers import create_test_edge
@pytest.mark.unit
class TestMeshGraphScoring:
"""Test MeshGraph candidate scoring functionality."""
def test_get_candidate_score_no_edges(self, mesh_graph):
"""Test scoring candidate with no graph edges."""
score = mesh_graph.get_candidate_score('01', None, None)
assert score == 0.0
def test_get_candidate_score_prev_edge_only(self, mesh_graph):
"""Test scoring candidate with only previous edge."""
mesh_graph.add_edge('7e', '01')
score = mesh_graph.get_candidate_score('01', '7e', None)
assert score > 0.0
assert score <= 1.0
def test_get_candidate_score_next_edge_only(self, mesh_graph):
"""Test scoring candidate with only next edge."""
mesh_graph.add_edge('01', '86')
score = mesh_graph.get_candidate_score('01', None, '86')
assert score > 0.0
assert score <= 1.0
def test_get_candidate_score_both_edges(self, mesh_graph):
"""Test scoring candidate with both previous and next edges."""
mesh_graph.add_edge('7e', '01')
mesh_graph.add_edge('01', '86')
score = mesh_graph.get_candidate_score('01', '7e', '86')
assert score > 0.0
assert score <= 1.0
def test_get_candidate_score_bidirectional_bonus(self, mesh_graph):
"""Test that bidirectional edges increase score."""
# Unidirectional
mesh_graph.add_edge('7e', '01')
score_uni = mesh_graph.get_candidate_score('01', '7e', None, use_bidirectional=True)
# Add reverse edge
mesh_graph.add_edge('01', '7e')
score_bi = mesh_graph.get_candidate_score('01', '7e', None, use_bidirectional=True)
assert score_bi > score_uni
def test_get_candidate_score_hop_position_match(self, mesh_graph):
"""Test hop position validation bonus."""
# Add edge with avg_hop_position = 2.0
mesh_graph.add_edge('7e', '01', hop_position=2)
mesh_graph.add_edge('7e', '01', hop_position=2) # Average stays at 2.0
# Score at position 2 (should match)
score_match = mesh_graph.get_candidate_score('01', '7e', None, hop_position=2, use_hop_position=True)
# Score at position 5 (shouldn't match)
score_mismatch = mesh_graph.get_candidate_score('01', '7e', None, hop_position=5, use_hop_position=True)
assert score_match > score_mismatch
def test_get_candidate_score_hop_position_tolerance(self, mesh_graph):
"""Test that hop position allows small tolerance."""
mesh_graph.add_edge('7e', '01', hop_position=2)
mesh_graph.add_edge('7e', '01', hop_position=2)
# Position 2.3 should still match (within 0.5 tolerance)
score = mesh_graph.get_candidate_score('01', '7e', None, hop_position=2, use_hop_position=True)
assert score > 0.0
def test_get_candidate_score_min_observations(self, mesh_graph):
"""Test that scoring respects min_observations."""
mesh_graph.add_edge('7e', '01') # Only 1 observation
score = mesh_graph.get_candidate_score('01', '7e', None, min_observations=3)
assert score == 0.0
# Add more observations
mesh_graph.add_edge('7e', '01')
mesh_graph.add_edge('7e', '01')
score = mesh_graph.get_candidate_score('01', '7e', None, min_observations=3)
assert score > 0.0
def test_get_candidate_score_hop_position_disabled(self, mesh_graph):
"""Test that hop position validation can be disabled."""
mesh_graph.add_edge('7e', '01', hop_position=2)
mesh_graph.add_edge('7e', '01', hop_position=2)
# With hop position disabled, should still score
score = mesh_graph.get_candidate_score('01', '7e', None, hop_position=5, use_hop_position=False)
assert score > 0.0
def test_get_candidate_score_bidirectional_disabled(self, mesh_graph):
"""Test that bidirectional check can be disabled."""
mesh_graph.add_edge('7e', '01')
mesh_graph.add_edge('01', '7e') # Reverse edge
score_bi_enabled = mesh_graph.get_candidate_score('01', '7e', None, use_bidirectional=True)
score_bi_disabled = mesh_graph.get_candidate_score('01', '7e', None, use_bidirectional=False)
assert score_bi_enabled > score_bi_disabled
def test_get_candidate_score_average_of_edges(self, mesh_graph):
"""Test that score averages confidence from multiple edges."""
# Create edges with different observation counts
@@ -116,7 +115,7 @@ class TestMeshGraphScoring:
mesh_graph.add_edge('01', '86')
mesh_graph.add_edge('01', '86')
mesh_graph.add_edge('01', '86') # 3 observations
# Score should be average of both edges
score = mesh_graph.get_candidate_score('01', '7e', '86')
assert 0.0 < score < 1.0
+32 -30
View File
@@ -3,61 +3,63 @@
Unit tests for MeshGraph path validation
"""
import pytest
from datetime import datetime, timedelta
from tests.helpers import create_test_edge, create_test_path
import pytest
from tests.helpers import create_test_path
@pytest.mark.unit
class TestMeshGraphValidation:
"""Test MeshGraph path validation functionality."""
def test_validate_path_segment_exists(self, mesh_graph):
"""Test validating an existing path segment."""
mesh_graph.add_edge('01', '7e', hop_position=1)
is_valid, confidence = mesh_graph.validate_path_segment('01', '7e')
assert is_valid is True
assert 0.0 <= confidence <= 1.0
def test_validate_path_segment_nonexistent(self, mesh_graph):
"""Test validating a non-existent path segment."""
is_valid, confidence = mesh_graph.validate_path_segment('01', '99')
assert is_valid is False
assert confidence == 0.0
def test_validate_path_segment_min_observations(self, mesh_graph):
"""Test that validation respects min_observations threshold."""
mesh_graph.add_edge('01', '7e')
# Edge has observation_count=1
is_valid, confidence = mesh_graph.validate_path_segment('01', '7e', min_observations=3)
assert is_valid is False
assert confidence == 0.0
# Add more observations
mesh_graph.add_edge('01', '7e')
mesh_graph.add_edge('01', '7e')
mesh_graph.add_edge('01', '7e')
is_valid, confidence = mesh_graph.validate_path_segment('01', '7e', min_observations=3)
assert is_valid is True
def test_validate_path_segment_recency(self, mesh_graph):
"""Test that validation considers recency of edge."""
# Add edge with recent timestamp
mesh_graph.add_edge('01', '7e')
recent_is_valid, recent_confidence = mesh_graph.validate_path_segment('01', '7e')
# Manually set old timestamp
edge = mesh_graph.get_edge('01', '7e')
edge['last_seen'] = datetime.now() - timedelta(days=30)
stale_is_valid, stale_confidence = mesh_graph.validate_path_segment('01', '7e')
# Recent edge should have higher confidence
assert recent_confidence > stale_confidence
def test_validate_path_segment_bidirectional(self, mesh_graph):
"""Test bidirectional edge bonus."""
# Add unidirectional edge
@@ -65,76 +67,76 @@ class TestMeshGraphValidation:
is_valid, unidirectional_confidence = mesh_graph.validate_path_segment(
'01', '7e', check_bidirectional=True
)
# Add reverse edge
mesh_graph.add_edge('7e', '01')
is_valid, bidirectional_confidence = mesh_graph.validate_path_segment(
'01', '7e', check_bidirectional=True
)
# Bidirectional should have higher confidence
assert bidirectional_confidence > unidirectional_confidence
def test_validate_path_segment_bidirectional_min_observations(self, mesh_graph):
"""Test that bidirectional check respects min_observations."""
# Add forward edge with enough observations
mesh_graph.add_edge('01', '7e', hop_position=1)
mesh_graph.add_edge('01', '7e', hop_position=1)
mesh_graph.add_edge('01', '7e', hop_position=1) # 3 observations
mesh_graph.add_edge('7e', '01') # Reverse edge with only 1 observation
# With min_observations=3, forward edge should be valid
is_valid, confidence = mesh_graph.validate_path_segment(
'01', '7e', min_observations=3, check_bidirectional=True
)
# Should still be valid from forward edge, but no bidirectional bonus
assert is_valid is True
# Add more observations to reverse edge
mesh_graph.add_edge('7e', '01')
mesh_graph.add_edge('7e', '01')
is_valid, confidence_with_bonus = mesh_graph.validate_path_segment(
'01', '7e', min_observations=3, check_bidirectional=True
)
assert confidence_with_bonus > confidence
def test_validate_path_single_node(self, mesh_graph):
"""Test that single-node path is always valid."""
is_valid, confidence = mesh_graph.validate_path(['01'])
assert is_valid is True
assert confidence == 1.0
def test_validate_path_empty(self, mesh_graph):
"""Test that empty path is always valid."""
is_valid, confidence = mesh_graph.validate_path([])
assert is_valid is True
assert confidence == 1.0
def test_validate_path_valid(self, populated_mesh_graph):
"""Test validating a valid path."""
path = create_test_path(['01', '7e', '86'])
is_valid, confidence = populated_mesh_graph.validate_path(path)
assert is_valid is True
assert confidence > 0.0
def test_validate_path_invalid_missing_edge(self, populated_mesh_graph):
"""Test validating a path with missing edge."""
path = create_test_path(['01', '99', '86']) # 01->99 doesn't exist
is_valid, confidence = populated_mesh_graph.validate_path(path)
assert is_valid is False
assert confidence == 0.0
def test_validate_path_invalid_insufficient_observations(self, mesh_graph):
"""Test validating a path with edge below min_observations."""
mesh_graph.add_edge('01', '7e') # Only 1 observation
path = create_test_path(['01', '7e'])
is_valid, confidence = mesh_graph.validate_path(path, min_observations=3)
assert is_valid is False
assert confidence == 0.0
def test_validate_path_average_confidence(self, mesh_graph):
"""Test that path validation returns average confidence."""
# Create path with edges of different observation counts
@@ -145,10 +147,10 @@ class TestMeshGraphValidation:
mesh_graph.add_edge('7e', '86')
mesh_graph.add_edge('7e', '86')
mesh_graph.add_edge('7e', '86') # 5 observations
path = create_test_path(['01', '7e', '86'])
is_valid, confidence = mesh_graph.validate_path(path)
assert is_valid is True
# Confidence should be average of both segments
assert 0.0 < confidence < 1.0
+72 -72
View File
@@ -3,256 +3,256 @@
Unit tests for PathCommand graph-based selection logic
"""
import pytest
from unittest.mock import Mock, patch
from datetime import datetime
from modules.commands.path_command import PathCommand
from tests.helpers import create_test_repeater, create_test_edge, populate_test_graph
from tests.helpers import create_test_repeater
@pytest.mark.unit
class TestPathCommandGraphSelection:
"""Test PathCommand._select_repeater_by_graph method."""
def test_select_repeater_by_graph_no_graph(self, mock_bot):
"""Test when graph_based_validation is False."""
# Disable graph validation
mock_bot.config.set('Path_Command', 'graph_based_validation', 'false')
path_cmd = PathCommand(mock_bot)
repeaters = [create_test_repeater('01', 'Test Repeater')]
result = path_cmd._select_repeater_by_graph(repeaters, '01', ['01'])
assert result == (None, 0.0, None)
def test_select_repeater_by_graph_no_mesh_graph(self, mock_bot):
"""Test when mesh_graph is None."""
mock_bot.mesh_graph = None
path_cmd = PathCommand(mock_bot)
repeaters = [create_test_repeater('01', 'Test Repeater')]
result = path_cmd._select_repeater_by_graph(repeaters, '01', ['01'])
assert result == (None, 0.0, None)
def test_select_repeater_by_graph_no_context(self, mock_bot, mesh_graph):
"""Test when node_id not in path_context."""
mock_bot.mesh_graph = mesh_graph
path_cmd = PathCommand(mock_bot)
repeaters = [create_test_repeater('99', 'Test Repeater')]
result = path_cmd._select_repeater_by_graph(repeaters, '99', ['01', '7e', '86'])
assert result == (None, 0.0, None)
def test_select_repeater_by_graph_direct_edge(self, mock_bot, mesh_graph):
"""Test selection with strong direct edge."""
mock_bot.mesh_graph = mesh_graph
# Create strong edge: 01 -> 7e
mesh_graph.add_edge('01', '7e')
for _ in range(10):
mesh_graph.add_edge('01', '7e') # 11 total observations
path_cmd = PathCommand(mock_bot)
# Create repeaters with matching prefix
repeaters = [
create_test_repeater('7e', 'Test Repeater 7e', public_key='7e' * 32)
]
result = path_cmd._select_repeater_by_graph(repeaters, '7e', ['01', '7e', '86'])
assert result[0] is not None # Should select a repeater
assert result[1] > 0.7 # High confidence
assert result[2] == 'graph' # Direct edge method
def test_select_repeater_by_graph_stored_public_key_bonus(self, mock_bot, mesh_graph):
"""Test stored public key bonus."""
mock_bot.mesh_graph = mesh_graph
# Create edge with stored public key
public_key = '7e' * 32 # 64 hex chars
mesh_graph.add_edge('01', '7e', from_public_key='01' * 32, to_public_key=public_key)
for _ in range(5):
mesh_graph.add_edge('01', '7e')
path_cmd = PathCommand(mock_bot)
# Create repeater with matching public key
repeaters = [
create_test_repeater('7e', 'Test Repeater', public_key=public_key),
create_test_repeater('7e', 'Other Repeater', public_key='aa' * 32) # Different key
]
result = path_cmd._select_repeater_by_graph(repeaters, '7e', ['01', '7e', '86'])
assert result[0] is not None
# Should select the repeater with matching public key
assert result[0]['public_key'] == public_key
assert result[1] > 0.5 # Should have good confidence with stored key bonus
def test_select_repeater_by_graph_star_bias(self, mock_bot, mesh_graph):
"""Test star bias multiplier application."""
mock_bot.mesh_graph = mesh_graph
# Create edges for both repeaters
mesh_graph.add_edge('01', '7e')
mesh_graph.add_edge('01', '7a')
path_cmd = PathCommand(mock_bot)
# Create one starred and one non-starred repeater
repeaters = [
create_test_repeater('7e', 'Starred Repeater', is_starred=True),
create_test_repeater('7a', 'Regular Repeater', is_starred=False)
]
result = path_cmd._select_repeater_by_graph(repeaters, '7e', ['01', '7e'])
assert result[0] is not None
# Starred repeater should be selected even if both have similar graph scores
assert result[0]['is_starred'] is True
def test_select_repeater_by_graph_multi_hop(self, mock_bot, mesh_graph):
"""Test multi-hop inference when direct edge has low confidence."""
mock_bot.mesh_graph = mesh_graph
# Create 2-hop path: 01 -> 7e -> 86 (no direct 01 -> 86)
# Need at least 3 observations per edge for min_edge_observations
for _ in range(3):
mesh_graph.add_edge('01', '7e')
mesh_graph.add_edge('7e', '86')
path_cmd = PathCommand(mock_bot)
# Create repeater that's the intermediate node
repeaters = [
create_test_repeater('7e', 'Intermediate Repeater', public_key='7e' * 32)
]
# Try to select 7e when path is 01 -> 7e -> 86
result = path_cmd._select_repeater_by_graph(repeaters, '7e', ['01', '7e', '86'])
assert result[0] is not None
# Should use graph method (direct edge exists)
assert result[2] in ('graph', 'graph_multihop')
def test_select_repeater_by_graph_hop_position(self, mock_bot, mesh_graph):
"""Test hop position validation."""
mock_bot.mesh_graph = mesh_graph
# Create edge with avg_hop_position = 1.0
# Need at least 3 observations for min_edge_observations
for _ in range(3):
mesh_graph.add_edge('01', '7e', hop_position=1)
path_cmd = PathCommand(mock_bot)
path_cmd.graph_use_hop_position = True
repeaters = [create_test_repeater('7e', 'Test Repeater')]
# Path where 7e is at position 1
result = path_cmd._select_repeater_by_graph(repeaters, '7e', ['01', '7e', '86'])
assert result[0] is not None
assert result[1] > 0.0
def test_select_repeater_by_graph_multiple_candidates(self, mock_bot, mesh_graph):
"""Test selection from multiple candidates."""
mock_bot.mesh_graph = mesh_graph
# Create edges with different strengths
mesh_graph.add_edge('01', '7e')
for _ in range(10):
mesh_graph.add_edge('01', '7e') # Strong edge
mesh_graph.add_edge('01', '7a')
for _ in range(2):
mesh_graph.add_edge('01', '7a') # Weaker edge
path_cmd = PathCommand(mock_bot)
repeaters = [
create_test_repeater('7e', 'Strong Edge Repeater'),
create_test_repeater('7a', 'Weak Edge Repeater')
]
result = path_cmd._select_repeater_by_graph(repeaters, '7e', ['01', '7e'])
assert result[0] is not None
# Should select the one with stronger edge (7e)
assert result[0]['name'] == 'Strong Edge Repeater'
assert result[1] > 0.5
def test_select_repeater_by_graph_confidence_conversion(self, mock_bot, mesh_graph):
"""Test graph score to confidence conversion."""
mock_bot.mesh_graph = mesh_graph
# Create very strong edge
mesh_graph.add_edge('01', '7e')
for _ in range(20):
mesh_graph.add_edge('01', '7e')
path_cmd = PathCommand(mock_bot)
repeaters = [create_test_repeater('7e', 'Test Repeater')]
result = path_cmd._select_repeater_by_graph(repeaters, '7e', ['01', '7e'])
assert result[0] is not None
# Confidence should be capped at 1.0
assert 0.0 <= result[1] <= 1.0
def test_select_repeater_by_graph_star_bias_exceeds_one(self, mock_bot, mesh_graph):
"""Test star bias can exceed 1.0 but confidence is normalized."""
mock_bot.mesh_graph = mesh_graph
mesh_graph.add_edge('01', '7e')
for _ in range(5):
mesh_graph.add_edge('01', '7e')
path_cmd = PathCommand(mock_bot)
path_cmd.star_bias_multiplier = 2.5 # High multiplier
# Create starred repeater
repeaters = [create_test_repeater('7e', 'Starred Repeater', is_starred=True)]
result = path_cmd._select_repeater_by_graph(repeaters, '7e', ['01', '7e'])
assert result[0] is not None
# Confidence should still be capped appropriately
assert 0.0 <= result[1] <= 1.0
def test_select_repeater_by_graph_prefix_extraction(self, mock_bot, mesh_graph):
"""Test prefix extraction from public_key."""
mock_bot.mesh_graph = mesh_graph
# Create edge
mesh_graph.add_edge('01', '7e')
path_cmd = PathCommand(mock_bot)
# Repeater with public key starting with '7e'
public_key = '7e' + '00' * 31 # 7e prefix
repeaters = [create_test_repeater('7e', 'Test Repeater', public_key=public_key)]
result = path_cmd._select_repeater_by_graph(repeaters, '7e', ['01', '7e'])
assert result[0] is not None
def test_select_repeater_by_graph_missing_public_key(self, mock_bot, mesh_graph):
"""Test handling when public_key is missing."""
mock_bot.mesh_graph = mesh_graph
mesh_graph.add_edge('01', '7e')
path_cmd = PathCommand(mock_bot)
# Repeater without public_key
repeater = create_test_repeater('7e', 'Test Repeater')
del repeater['public_key'] # Remove public key
result = path_cmd._select_repeater_by_graph([repeater], '7e', ['01', '7e'])
# Should skip this repeater (no prefix to match)
assert result == (None, 0.0, None) or result[0] is None
+32 -31
View File
@@ -3,8 +3,9 @@
Unit tests for PathCommand graph-based selection
"""
import pytest
from unittest.mock import Mock
from modules.commands.path_command import PathCommand
from tests.helpers import create_test_repeater
@@ -12,7 +13,7 @@ from tests.helpers import create_test_repeater
@pytest.mark.unit
class TestPathCommandGraphSelection:
"""Test PathCommand._select_repeater_by_graph functionality."""
@pytest.fixture
def path_command(self, mock_bot, populated_mesh_graph):
"""Create a PathCommand instance with graph enabled."""
@@ -27,41 +28,41 @@ class TestPathCommandGraphSelection:
command.graph_prefer_stored_keys = True
command.star_bias_multiplier = 2.5
return command
def test_select_repeater_direct_edge(self, path_command, populated_mesh_graph):
"""Test selecting repeater with direct graph edge."""
repeaters = [
create_test_repeater('7e', 'Repeater 7e', public_key='7e' * 32)
]
path_context = ['01', '7e', '86']
repeater, confidence, method = path_command._select_repeater_by_graph(
repeaters, '7e', path_context
)
assert repeater is not None
assert confidence > 0.0
assert method == 'graph'
def test_select_repeater_stored_key_bonus(self, path_command, populated_mesh_graph):
"""Test that stored public keys provide bonus."""
# Add edge with stored public key
stored_key = '7e' * 32
populated_mesh_graph.add_edge('01', '7e', to_public_key=stored_key)
repeaters = [
create_test_repeater('7e', 'Repeater 7e', public_key=stored_key),
create_test_repeater('7e', 'Other 7e', public_key='7f' * 32)
]
path_context = ['01', '7e', '86']
repeater, confidence, method = path_command._select_repeater_by_graph(
repeaters, '7e', path_context
)
assert repeater is not None
assert repeater['name'] == 'Repeater 7e' # Should prefer stored key match
def test_select_repeater_star_bias(self, path_command, populated_mesh_graph):
"""Test that starred repeaters get score boost."""
repeaters = [
@@ -69,87 +70,87 @@ class TestPathCommandGraphSelection:
create_test_repeater('7e', 'Regular Repeater', is_starred=False)
]
path_context = ['01', '7e', '86']
repeater, confidence, method = path_command._select_repeater_by_graph(
repeaters, '7e', path_context
)
assert repeater is not None
assert repeater['is_starred'] is True
def test_select_repeater_multihop(self, path_command, populated_mesh_graph):
"""Test multi-hop inference when direct edge has low confidence."""
# Create 2-hop path: 01 -> 7e -> 86 -> e0
populated_mesh_graph.add_edge('01', '7e')
populated_mesh_graph.add_edge('7e', '86')
populated_mesh_graph.add_edge('86', 'e0')
repeaters = [
create_test_repeater('86', 'Intermediate Repeater')
]
path_context = ['01', '86', 'e0']
repeater, confidence, method = path_command._select_repeater_by_graph(
repeaters, '86', path_context
)
assert repeater is not None
assert method == 'graph_multihop' or method == 'graph'
def test_select_repeater_no_graph(self, path_command):
"""Test that selection returns None when graph is disabled."""
path_command.graph_based_validation = False
repeaters = [create_test_repeater('7e', 'Repeater')]
path_context = ['01', '7e', '86']
repeater, confidence, method = path_command._select_repeater_by_graph(
repeaters, '7e', path_context
)
assert repeater is None
assert confidence == 0.0
assert method is None
def test_select_repeater_no_candidates(self, path_command, populated_mesh_graph):
"""Test that selection returns None when no valid candidates."""
repeaters = [
create_test_repeater('99', 'Unknown Repeater') # No graph edges
]
path_context = ['01', '99', '86']
repeater, confidence, method = path_command._select_repeater_by_graph(
repeaters, '99', path_context
)
assert repeater is None or confidence == 0.0
def test_select_repeater_confidence_conversion(self, path_command, populated_mesh_graph):
"""Test that graph scores are converted to confidence properly."""
repeaters = [
create_test_repeater('7e', 'Repeater 7e')
]
path_context = ['01', '7e', '86']
repeater, confidence, method = path_command._select_repeater_by_graph(
repeaters, '7e', path_context
)
assert 0.0 <= confidence <= 1.0
def test_select_repeater_hop_position(self, path_command, mesh_graph):
"""Test that hop position is used when enabled."""
path_command.graph_use_hop_position = True
# Add edge with specific hop position (add multiple times to establish avg)
mesh_graph.add_edge('01', '7e', hop_position=1)
mesh_graph.add_edge('01', '7e', hop_position=1) # Keep avg at 1.0
repeaters = [create_test_repeater('7e', 'Repeater')]
path_context = ['01', '7e', '86'] # 7e is at position 1
repeater, confidence, method = path_command._select_repeater_by_graph(
repeaters, '7e', path_context
)
assert repeater is not None
+2 -1
View File
@@ -3,8 +3,9 @@
Unit tests for PathCommand multi-byte path support: routing_info usage and comma/prefix parsing.
"""
import pytest
from unittest.mock import Mock
from modules.commands.path_command import PathCommand
from modules.models import MeshMessage