mirror of
https://github.com/agessaman/meshcore-bot.git
synced 2026-11-04 03:52:34 +00:00
agessaman
0ad9f4e69a
bot_prefix and graph_trace_helper's bot key read meshcore.device, which meshcore_py doesn't have, so both were always None and update_mesh_graph_from_trace_data skipped every trace. Both now come from meshcore.self_info's public_key (device stays a fallback), and bot_prefix is read lazily because the tracker is built before the radio connects. The helper names the bot at the trace's hash width, so the last hop's edge matches the hops' own width, and marks 2-byte edges with prefix_bytes=2. trace_command read update_graph_two_byte but never used it; 2-byte traces now follow it instead of update_graph_one_byte.
342 lines
15 KiB
Python
342 lines
15 KiB
Python
"""Tests for modules.graph_trace_helper — update_mesh_graph_from_trace_data."""
|
|
|
|
import configparser
|
|
from unittest.mock import MagicMock, Mock
|
|
|
|
import pytest
|
|
|
|
from modules.graph_trace_helper import update_mesh_graph_from_trace_data
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _make_bot(bot_prefix="aa", has_mesh_graph=True, has_transmission_tracker=True):
|
|
"""Create a minimal mock bot for graph_trace_helper tests."""
|
|
bot = MagicMock()
|
|
bot.logger = Mock()
|
|
|
|
config = configparser.ConfigParser()
|
|
config.add_section("Path_Command")
|
|
config.set("Path_Command", "graph_edge_expiration_days", "7")
|
|
bot.config = config
|
|
|
|
if has_mesh_graph:
|
|
bot.mesh_graph = MagicMock()
|
|
else:
|
|
bot.mesh_graph = None
|
|
|
|
if has_transmission_tracker:
|
|
bot.transmission_tracker = MagicMock()
|
|
bot.transmission_tracker.bot_prefix = bot_prefix
|
|
bot.transmission_tracker.match_packet_hash = Mock(return_value=None)
|
|
else:
|
|
bot.transmission_tracker = None
|
|
|
|
# DB manager returns empty results by default
|
|
bot.db_manager = MagicMock()
|
|
bot.db_manager.execute_query = Mock(return_value=[])
|
|
|
|
# The radio's key, consistent with bot_prefix (none when bot_prefix is empty)
|
|
bot.meshcore = MagicMock()
|
|
bot.meshcore.self_info = {"public_key": (bot_prefix.lower() * 64)[:64]} if bot_prefix else {}
|
|
bot.meshcore.device = None
|
|
|
|
return bot
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Early exit / guard cases
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestEarlyExits:
|
|
def test_empty_path_hashes_returns_immediately(self):
|
|
bot = _make_bot()
|
|
update_mesh_graph_from_trace_data(bot, [], {})
|
|
bot.mesh_graph.add_edge.assert_not_called()
|
|
|
|
def test_none_path_hashes_treated_as_empty(self):
|
|
bot = _make_bot()
|
|
# empty list is falsy, so this is covered by the guard
|
|
update_mesh_graph_from_trace_data(bot, [], {})
|
|
bot.mesh_graph.add_edge.assert_not_called()
|
|
|
|
def test_no_mesh_graph_returns_immediately(self):
|
|
bot = _make_bot(has_mesh_graph=False)
|
|
update_mesh_graph_from_trace_data(bot, ["ab"], {})
|
|
# Should log and return without crash
|
|
bot.logger.debug.assert_called()
|
|
|
|
def test_no_transmission_tracker_returns_immediately(self):
|
|
bot = _make_bot(has_transmission_tracker=False)
|
|
update_mesh_graph_from_trace_data(bot, ["ab"], {})
|
|
bot.logger.debug.assert_called()
|
|
|
|
def test_missing_mesh_graph_attribute(self):
|
|
bot = _make_bot()
|
|
del bot.mesh_graph
|
|
update_mesh_graph_from_trace_data(bot, ["ab"], {})
|
|
# No crash expected
|
|
|
|
def test_missing_transmission_tracker_attribute(self):
|
|
bot = _make_bot()
|
|
del bot.transmission_tracker
|
|
update_mesh_graph_from_trace_data(bot, ["ab"], {})
|
|
# No crash expected
|
|
|
|
def test_empty_bot_prefix_returns_immediately(self):
|
|
bot = _make_bot(bot_prefix=None)
|
|
update_mesh_graph_from_trace_data(bot, ["ab"], {})
|
|
bot.mesh_graph.add_edge.assert_not_called()
|
|
|
|
def test_empty_string_bot_prefix_returns_immediately(self):
|
|
bot = _make_bot(bot_prefix="")
|
|
update_mesh_graph_from_trace_data(bot, ["ab"], {})
|
|
bot.mesh_graph.add_edge.assert_not_called()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# is_our_trace resolution
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestIsOurTraceResolution:
|
|
def test_is_our_trace_none_resolves_false_when_no_match(self):
|
|
"""When packet_hash doesn't match, is_our_trace stays False."""
|
|
bot = _make_bot(bot_prefix="aa")
|
|
bot.transmission_tracker.match_packet_hash = Mock(return_value=None)
|
|
update_mesh_graph_from_trace_data(bot, ["bb"], {"packet_hash": "abc123"})
|
|
# Not our trace → should add one edge (last_node → bot)
|
|
bot.mesh_graph.add_edge.assert_called_once()
|
|
|
|
def test_is_our_trace_none_resolves_true_when_match(self):
|
|
"""When packet_hash matches a transmission record, is_our_trace becomes True."""
|
|
bot = _make_bot(bot_prefix="aa")
|
|
bot.transmission_tracker.match_packet_hash = Mock(return_value={"matched": True})
|
|
# Single hop — should trigger immediate-neighbor path (bidirectional)
|
|
update_mesh_graph_from_trace_data(bot, ["bb"], {"packet_hash": "abc123"})
|
|
# Bidirectional: 2 add_edge calls
|
|
assert bot.mesh_graph.add_edge.call_count == 2
|
|
|
|
def test_is_our_trace_explicit_true(self):
|
|
"""Explicit is_our_trace=True skips transmission tracker lookup."""
|
|
bot = _make_bot(bot_prefix="aa")
|
|
bot.transmission_tracker.match_packet_hash = Mock(return_value=None)
|
|
update_mesh_graph_from_trace_data(bot, ["bb"], {}, is_our_trace=True)
|
|
# Single hop + explicit True → immediate neighbor → bidirectional
|
|
assert bot.mesh_graph.add_edge.call_count == 2
|
|
|
|
def test_is_our_trace_explicit_false(self):
|
|
"""Explicit is_our_trace=False skips transmission tracker lookup."""
|
|
bot = _make_bot(bot_prefix="aa")
|
|
bot.transmission_tracker.match_packet_hash = Mock(return_value={"matched": True})
|
|
update_mesh_graph_from_trace_data(bot, ["bb"], {}, is_our_trace=False)
|
|
# Not our trace even though match_packet_hash would return a record
|
|
bot.mesh_graph.add_edge.assert_called_once()
|
|
|
|
def test_is_our_trace_true_no_packet_hash(self):
|
|
"""is_our_trace None with no packet_hash defaults to False."""
|
|
bot = _make_bot(bot_prefix="aa")
|
|
update_mesh_graph_from_trace_data(bot, ["bb"], {})
|
|
# No packet_hash → is_our_trace stays False → one edge
|
|
bot.mesh_graph.add_edge.assert_called_once()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Immediate neighbor (single hop, is_our_trace=True)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestImmediateNeighbor:
|
|
def test_single_hop_creates_bidirectional_edges(self):
|
|
bot = _make_bot(bot_prefix="aa")
|
|
update_mesh_graph_from_trace_data(bot, ["bb"], {}, is_our_trace=True)
|
|
assert bot.mesh_graph.add_edge.call_count == 2
|
|
|
|
def test_single_hop_edge_directions(self):
|
|
bot = _make_bot(bot_prefix="aa")
|
|
update_mesh_graph_from_trace_data(bot, ["bb"], {}, is_our_trace=True)
|
|
calls = bot.mesh_graph.add_edge.call_args_list
|
|
from_prefixes = {c.kwargs.get("from_prefix") or c[1].get("from_prefix") for c in calls}
|
|
to_prefixes = {c.kwargs.get("to_prefix") or c[1].get("to_prefix") for c in calls}
|
|
# Both directions: aa↔bb
|
|
assert "aa" in from_prefixes or "aa" in to_prefixes
|
|
assert "bb" in from_prefixes or "bb" in to_prefixes
|
|
|
|
def test_single_hop_lowercase_normalization(self):
|
|
"""Path hashes are lowercased before use."""
|
|
bot = _make_bot(bot_prefix="AA")
|
|
update_mesh_graph_from_trace_data(bot, ["BB"], {}, is_our_trace=True)
|
|
assert bot.mesh_graph.add_edge.call_count == 2
|
|
calls = bot.mesh_graph.add_edge.call_args_list
|
|
# Check lowercase normalization
|
|
all_prefixes = set()
|
|
for c in calls:
|
|
all_prefixes.add(c.kwargs.get("from_prefix") or (c[0][0] if c[0] else None))
|
|
all_prefixes.add(c.kwargs.get("to_prefix") or (c[0][1] if len(c[0]) > 1 else None))
|
|
# Should contain lowercase versions
|
|
assert "aa" in all_prefixes or "bb" in all_prefixes
|
|
|
|
def test_single_hop_with_unique_db_key(self):
|
|
"""When DB returns exactly one public_key for the neighbor, it's used."""
|
|
bot = _make_bot(bot_prefix="aa")
|
|
# First query returns count=1, second returns the key
|
|
bot.db_manager.execute_query = Mock(side_effect=[
|
|
[{"count": 1}],
|
|
[{"public_key": "bb" * 32}],
|
|
[], # bot location query
|
|
])
|
|
update_mesh_graph_from_trace_data(bot, ["bb"], {}, is_our_trace=True)
|
|
assert bot.mesh_graph.add_edge.call_count == 2
|
|
|
|
def test_single_hop_with_ambiguous_db_results(self):
|
|
"""When DB returns count != 1, no key is resolved but edges still added."""
|
|
bot = _make_bot(bot_prefix="aa")
|
|
bot.db_manager.execute_query = Mock(return_value=[{"count": 2}])
|
|
update_mesh_graph_from_trace_data(bot, ["bb"], {}, is_our_trace=True)
|
|
assert bot.mesh_graph.add_edge.call_count == 2
|
|
|
|
def test_single_hop_db_exception_handled(self):
|
|
"""DB exceptions during key lookup are swallowed; edges still added."""
|
|
bot = _make_bot(bot_prefix="aa")
|
|
bot.db_manager.execute_query = Mock(side_effect=Exception("DB error"))
|
|
update_mesh_graph_from_trace_data(bot, ["bb"], {}, is_our_trace=True)
|
|
assert bot.mesh_graph.add_edge.call_count == 2
|
|
|
|
def test_single_hop_no_meshcore_device(self):
|
|
"""No meshcore device → bot_key is None, edges still added."""
|
|
bot = _make_bot(bot_prefix="aa")
|
|
bot.meshcore = None
|
|
update_mesh_graph_from_trace_data(bot, ["bb"], {}, is_our_trace=True)
|
|
assert bot.mesh_graph.add_edge.call_count == 2
|
|
|
|
def test_single_hop_device_pubkey_bytes(self):
|
|
"""Device public key as bytes is hex-encoded."""
|
|
bot = _make_bot(bot_prefix="aa")
|
|
bot.meshcore.self_info = {}
|
|
bot.meshcore.device = MagicMock(public_key=b"\xaa\xbb")
|
|
update_mesh_graph_from_trace_data(bot, ["bb"], {}, is_our_trace=True)
|
|
assert bot.mesh_graph.add_edge.call_count == 2
|
|
assert bot.mesh_graph.add_edge.call_args_list[0].kwargs["from_public_key"] == "aabb"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Regular trace (bot is destination, not immediate neighbor)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestRegularTrace:
|
|
def test_two_hop_creates_two_edges(self):
|
|
"""[a, b] path (bot receives from b via a): two edges expected."""
|
|
bot = _make_bot(bot_prefix="cc")
|
|
update_mesh_graph_from_trace_data(bot, ["aa", "bb"], {})
|
|
# 1 edge: last_node(bb) → bot(cc)
|
|
# 1 edge: aa → bb
|
|
assert bot.mesh_graph.add_edge.call_count == 2
|
|
|
|
def test_single_node_path_creates_one_edge(self):
|
|
"""Single-node path: last_node → bot."""
|
|
bot = _make_bot(bot_prefix="cc")
|
|
update_mesh_graph_from_trace_data(bot, ["aa"], {})
|
|
assert bot.mesh_graph.add_edge.call_count == 1
|
|
|
|
def test_three_hop_creates_three_edges(self):
|
|
"""[a, b, c] path creates 3 edges."""
|
|
bot = _make_bot(bot_prefix="dd")
|
|
update_mesh_graph_from_trace_data(bot, ["aa", "bb", "cc"], {})
|
|
assert bot.mesh_graph.add_edge.call_count == 3
|
|
|
|
def test_path_hashes_lowercased(self):
|
|
"""Uppercase path hashes are normalized to lowercase."""
|
|
bot = _make_bot(bot_prefix="cc")
|
|
update_mesh_graph_from_trace_data(bot, ["AA", "BB"], {})
|
|
assert bot.mesh_graph.add_edge.call_count == 2
|
|
calls = bot.mesh_graph.add_edge.call_args_list
|
|
# All prefixes should be lowercase
|
|
for c in calls:
|
|
kw = c.kwargs
|
|
if "from_prefix" in kw:
|
|
assert kw["from_prefix"] == kw["from_prefix"].lower()
|
|
if "to_prefix" in kw:
|
|
assert kw["to_prefix"] == kw["to_prefix"].lower()
|
|
|
|
def test_last_edge_points_to_bot(self):
|
|
"""The last_node→bot edge has to_prefix equal to bot_prefix."""
|
|
bot = _make_bot(bot_prefix="cc")
|
|
update_mesh_graph_from_trace_data(bot, ["aa"], {})
|
|
call_kwargs = bot.mesh_graph.add_edge.call_args.kwargs
|
|
assert call_kwargs.get("to_prefix") == "cc"
|
|
assert call_kwargs.get("from_prefix") == "aa"
|
|
|
|
def test_db_exception_in_regular_path_handled(self):
|
|
"""DB exceptions don't prevent edges from being added."""
|
|
bot = _make_bot(bot_prefix="cc")
|
|
bot.db_manager.execute_query = Mock(side_effect=Exception("DB error"))
|
|
update_mesh_graph_from_trace_data(bot, ["aa", "bb"], {})
|
|
assert bot.mesh_graph.add_edge.call_count == 2
|
|
|
|
def test_unique_key_resolved_for_last_node(self):
|
|
"""When exactly one key matches the last_node prefix, it's used."""
|
|
bot = _make_bot(bot_prefix="cc")
|
|
bot.db_manager.execute_query = Mock(side_effect=[
|
|
[{"count": 1}],
|
|
[{"public_key": "aa" * 32}],
|
|
])
|
|
update_mesh_graph_from_trace_data(bot, ["aa"], {})
|
|
assert bot.mesh_graph.add_edge.call_count == 1
|
|
|
|
def test_mesh_graph_add_edge_exception_propagates(self):
|
|
"""Exception from add_edge is not caught (it's a programming error)."""
|
|
bot = _make_bot(bot_prefix="cc")
|
|
bot.mesh_graph.add_edge = Mock(side_effect=RuntimeError("mesh error"))
|
|
with pytest.raises(RuntimeError):
|
|
update_mesh_graph_from_trace_data(bot, ["aa"], {})
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Multi-hop intermediate edges
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestMultiHopEdges:
|
|
def test_hop_positions_decrease_towards_bot(self):
|
|
"""Hop positions are assigned based on distance from bot."""
|
|
bot = _make_bot(bot_prefix="dd")
|
|
update_mesh_graph_from_trace_data(bot, ["aa", "bb", "cc"], {})
|
|
calls = bot.mesh_graph.add_edge.call_args_list
|
|
# Collect hop_position values from kwargs
|
|
hop_positions = [c.kwargs.get("hop_position") for c in calls]
|
|
assert all(h is not None for h in hop_positions)
|
|
|
|
def test_single_hop_position_is_1(self):
|
|
"""Single-hop path has hop_position=1."""
|
|
bot = _make_bot(bot_prefix="bb")
|
|
update_mesh_graph_from_trace_data(bot, ["aa"], {})
|
|
call_kwargs = bot.mesh_graph.add_edge.call_args.kwargs
|
|
assert call_kwargs.get("hop_position") == 1
|
|
|
|
def test_no_meshcore_in_regular_path(self):
|
|
"""No meshcore still creates edges."""
|
|
bot = _make_bot(bot_prefix="cc")
|
|
bot.meshcore = None
|
|
update_mesh_graph_from_trace_data(bot, ["aa", "bb"], {})
|
|
assert bot.mesh_graph.add_edge.call_count == 2
|
|
|
|
|
|
|
|
class TestBotIdentityFromRadio:
|
|
"""The bot's node comes from the radio's key (meshcore self_info), at the trace's hash width."""
|
|
|
|
def test_one_byte_trace_names_the_bot_by_its_first_byte(self):
|
|
bot = _make_bot(bot_prefix=None)
|
|
bot.meshcore.self_info = {"public_key": "c3d4" + "00" * 30}
|
|
update_mesh_graph_from_trace_data(bot, ["aa"], {})
|
|
kwargs = bot.mesh_graph.add_edge.call_args.kwargs
|
|
assert kwargs["to_prefix"] == "c3"
|
|
assert kwargs["to_public_key"] == "c3d4" + "00" * 30
|
|
assert "prefix_bytes" not in kwargs
|
|
|
|
def test_two_byte_trace_names_the_bot_by_two_bytes_and_flags_the_edge(self):
|
|
bot = _make_bot(bot_prefix=None)
|
|
bot.meshcore.self_info = {"public_key": "c3d4" + "00" * 30}
|
|
update_mesh_graph_from_trace_data(bot, ["a1b2", "e5f6"], {})
|
|
last = bot.mesh_graph.add_edge.call_args_list[0].kwargs
|
|
assert (last["from_prefix"], last["to_prefix"]) == ("e5f6", "c3d4")
|
|
assert all(c.kwargs.get("prefix_bytes") == 2 for c in bot.mesh_graph.add_edge.call_args_list)
|