mirror of
https://github.com/agessaman/meshcore-bot.git
synced 2026-11-01 02:22:47 +00:00
agessaman
a42ba576db
Codex review of the bot_prefix fix; these helper defects only mattered once traces reached the graph. - Our own trace also confirms the bot -> first hop link, which was missing. Edges follow the chain bot -> h1 ... hn -> bot with 1-based hop positions in path order, as the message path writer numbers them (they were reversed). - A public key is stored only when its prefix belongs to one recent repeater. The location lookup still picks a position for the distance, but its pick among colliding prefixes no longer becomes the stored key (also in the one-hop case). - bot_prefix reads the radio's current key on each access, so reconnecting or rebooting into another radio is picked up. - extract_repeater_prefixes_from_path no longer drops a last hop equal to our prefix: a companion never forwards, so it is a repeater sharing our short prefix. On dev bot_prefix was always None, so this keeps the attribution dev already had.
704 lines
29 KiB
Python
704 lines
29 KiB
Python
"""Tests for modules/transmission_tracker.py."""
|
|
|
|
import json
|
|
import sqlite3
|
|
import time
|
|
from contextlib import closing
|
|
from unittest.mock import Mock
|
|
|
|
import pytest
|
|
|
|
from modules.transmission_tracker import TransmissionRecord, TransmissionTracker
|
|
|
|
|
|
def _chan(text, idx=0, sender=None):
|
|
"""A received GRP_TXT that decrypted on channel ``idx`` to ``text`` (from ``sender`` when given)."""
|
|
message = f"{sender}: {text}" if sender else text
|
|
return {"payload_type": 5, "payload_hex": "", "channel_idx": idx, "channel_text": text, "channel_message": message}
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_bot(mock_logger):
|
|
"""Minimal bot mock for TransmissionTracker."""
|
|
bot = Mock()
|
|
bot.logger = mock_logger
|
|
bot.meshcore = None # No device connected
|
|
bot.prefix_hex_chars = 2
|
|
return bot
|
|
|
|
|
|
@pytest.fixture
|
|
def tracker(mock_bot):
|
|
"""TransmissionTracker instance with a mock bot."""
|
|
return TransmissionTracker(mock_bot)
|
|
|
|
|
|
class TestTransmissionRecord:
|
|
"""Tests for TransmissionRecord dataclass."""
|
|
|
|
def test_default_fields(self):
|
|
rec = TransmissionRecord(
|
|
timestamp=1234.0,
|
|
content="hello",
|
|
target="general",
|
|
message_type="channel",
|
|
)
|
|
assert rec.repeat_count == 0
|
|
assert rec.packet_hash is None
|
|
assert rec.command_id is None
|
|
assert rec.repeater_prefixes == set()
|
|
assert rec.repeater_counts == {}
|
|
|
|
def test_custom_fields(self):
|
|
rec = TransmissionRecord(
|
|
timestamp=5678.0,
|
|
content="dm text",
|
|
target="Alice",
|
|
message_type="dm",
|
|
packet_hash="abcd1234",
|
|
command_id="cmd-001",
|
|
)
|
|
assert rec.packet_hash == "abcd1234"
|
|
assert rec.command_id == "cmd-001"
|
|
|
|
|
|
class TestRecordTransmission:
|
|
"""Tests for TransmissionTracker.record_transmission()."""
|
|
|
|
def test_returns_transmission_record(self, tracker):
|
|
rec = tracker.record_transmission("hello", "general", "channel")
|
|
assert isinstance(rec, TransmissionRecord)
|
|
assert rec.content == "hello"
|
|
assert rec.target == "general"
|
|
assert rec.message_type == "channel"
|
|
|
|
def test_stores_in_pending(self, tracker):
|
|
rec = tracker.record_transmission("msg", "ch", "channel", channel_idx=0)
|
|
key = int(rec.timestamp)
|
|
assert key in tracker.pending_transmissions
|
|
assert rec in tracker.pending_transmissions[key]
|
|
|
|
def test_multiple_records_same_second(self, tracker):
|
|
rec1 = tracker.record_transmission("a", "ch", "channel")
|
|
rec2 = tracker.record_transmission("b", "ch", "channel")
|
|
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), [])
|
|
|
|
def test_with_command_id(self, tracker):
|
|
rec = tracker.record_transmission("msg", "ch", "channel", command_id="cmd-42")
|
|
assert rec.command_id == "cmd-42"
|
|
|
|
|
|
class TestMatchPacketHash:
|
|
"""Tests for TransmissionTracker.match_packet_hash()."""
|
|
|
|
def test_null_hash_returns_none(self, tracker):
|
|
assert tracker.match_packet_hash("", time.time()) is None
|
|
assert tracker.match_packet_hash("0000000000000000", time.time()) is None
|
|
|
|
def test_matches_pending_transmission(self, tracker):
|
|
rec = tracker.record_transmission("msg", "general", "channel", channel_idx=0)
|
|
result = tracker.match_packet_hash("deadbeef", rec.timestamp + 1, _chan("msg"))
|
|
assert result is not None
|
|
assert result.packet_hash == "deadbeef"
|
|
|
|
def test_already_confirmed_returned_immediately(self, tracker):
|
|
rec = tracker.record_transmission("msg", "ch", "channel", channel_idx=0)
|
|
# First match confirms it
|
|
tracker.match_packet_hash("abc123", rec.timestamp, _chan("msg"))
|
|
# Second call returns same confirmed record
|
|
result2 = tracker.match_packet_hash("abc123", time.time())
|
|
assert result2 is not None
|
|
assert result2.packet_hash == "abc123"
|
|
|
|
def test_no_match_outside_window(self, tracker):
|
|
rec = tracker.record_transmission("msg", "ch", "channel", channel_idx=0)
|
|
# RF timestamp far in the future
|
|
result = tracker.match_packet_hash("deadbeef", rec.timestamp + 9999, _chan("msg"))
|
|
assert result is None
|
|
|
|
|
|
class TestRecordRepeat:
|
|
"""Tests for TransmissionTracker.record_repeat()."""
|
|
|
|
def test_null_hash_returns_false(self, tracker):
|
|
assert tracker.record_repeat("") is False
|
|
assert tracker.record_repeat("0000000000000000") is False
|
|
|
|
def test_repeat_increments_count(self, tracker):
|
|
rec = tracker.record_transmission("msg", "ch", "channel", channel_idx=0)
|
|
# First confirm the hash
|
|
tracker.match_packet_hash("hash01", rec.timestamp, _chan("msg"))
|
|
# Now record a repeat
|
|
result = tracker.record_repeat("hash01", repeater_prefix="7e")
|
|
assert result is True
|
|
assert rec.repeat_count == 1
|
|
assert "7e" in rec.repeater_prefixes
|
|
|
|
def test_repeat_without_prefix(self, tracker):
|
|
rec = tracker.record_transmission("msg", "ch", "channel", channel_idx=0)
|
|
tracker.match_packet_hash("hash02", rec.timestamp, _chan("msg"))
|
|
result = tracker.record_repeat("hash02")
|
|
assert result is True
|
|
assert rec.repeat_count == 1
|
|
assert rec.repeater_counts.get("_unknown") == 1
|
|
|
|
def test_multiple_repeats_same_repeater(self, tracker):
|
|
rec = tracker.record_transmission("msg", "ch", "channel", channel_idx=0)
|
|
tracker.match_packet_hash("hash03", rec.timestamp, _chan("msg"))
|
|
tracker.record_repeat("hash03", repeater_prefix="01")
|
|
tracker.record_repeat("hash03", repeater_prefix="01")
|
|
assert rec.repeat_count == 2
|
|
assert rec.repeater_counts["01"] == 2
|
|
|
|
def test_unmatched_hash_returns_false(self, tracker):
|
|
result = tracker.record_repeat("nonexistent_hash")
|
|
assert result is False
|
|
|
|
|
|
class TestGetRepeatInfo:
|
|
"""Tests for TransmissionTracker.get_repeat_info()."""
|
|
|
|
def test_unknown_hash_returns_zeros(self, tracker):
|
|
info = tracker.get_repeat_info(packet_hash="unknown")
|
|
assert info["repeat_count"] == 0
|
|
assert info["repeater_prefixes"] == []
|
|
assert info["repeater_counts"] == {}
|
|
|
|
def test_lookup_by_packet_hash(self, tracker):
|
|
rec = tracker.record_transmission("msg", "ch", "channel", channel_idx=0)
|
|
tracker.match_packet_hash("hashXX", rec.timestamp, _chan("msg"))
|
|
tracker.record_repeat("hashXX", repeater_prefix="7e")
|
|
info = tracker.get_repeat_info(packet_hash="hashXX")
|
|
assert info["repeat_count"] == 1
|
|
assert "7e" in info["repeater_prefixes"]
|
|
|
|
def test_lookup_by_command_id(self, tracker):
|
|
rec = tracker.record_transmission("msg", "ch", "channel", command_id="cmd-99", channel_idx=0)
|
|
tracker.match_packet_hash("hashYY", rec.timestamp, _chan("msg"))
|
|
tracker.record_repeat("hashYY", repeater_prefix="ab")
|
|
info = tracker.get_repeat_info(command_id="cmd-99")
|
|
assert info["repeat_count"] == 1
|
|
assert "ab" in info["repeater_prefixes"]
|
|
|
|
|
|
class TestExtractRepeaterPrefixes:
|
|
"""Tests for TransmissionTracker.extract_repeater_prefixes_from_path()."""
|
|
|
|
def test_extracts_last_hop_from_path_string(self, tracker):
|
|
result = tracker.extract_repeater_prefixes_from_path("01,7e,86")
|
|
assert result == ["86"]
|
|
|
|
def test_extracts_from_path_nodes(self, tracker):
|
|
result = tracker.extract_repeater_prefixes_from_path(None, path_nodes=["01", "7e", "86"])
|
|
assert result == ["86"]
|
|
|
|
def test_a_repeater_sharing_our_prefix_keeps_the_repeat(self, tracker):
|
|
"""A companion never forwards, so a last hop matching our prefix is a repeater."""
|
|
tracker.bot_prefix = "86"
|
|
assert tracker.extract_repeater_prefixes_from_path("01,86") == ["86"]
|
|
assert tracker.extract_repeater_prefixes_from_path(None, ["01", "86"]) == ["86"]
|
|
|
|
def test_empty_path_returns_empty(self, tracker):
|
|
result = tracker.extract_repeater_prefixes_from_path(None)
|
|
assert result == []
|
|
|
|
def test_path_with_route_type_annotation(self, tracker):
|
|
result = tracker.extract_repeater_prefixes_from_path("01,7e,55 via ROUTE_TYPE_FLOOD")
|
|
assert result == ["55"]
|
|
|
|
def test_single_node_path(self, tracker):
|
|
result = tracker.extract_repeater_prefixes_from_path("7e")
|
|
assert result == ["7e"]
|
|
|
|
|
|
class TestCleanupOldRecords:
|
|
"""Tests for TransmissionTracker.cleanup_old_records()."""
|
|
|
|
def test_removes_old_pending(self, tracker):
|
|
# Inject a record with an old timestamp
|
|
old_rec = TransmissionRecord(
|
|
timestamp=time.time() - 600, # 10 minutes ago (beyond cleanup_after=300)
|
|
content="old msg",
|
|
target="ch",
|
|
message_type="channel",
|
|
)
|
|
old_key = int(old_rec.timestamp)
|
|
tracker.pending_transmissions[old_key] = [old_rec]
|
|
tracker.cleanup_old_records()
|
|
assert old_key not in tracker.pending_transmissions
|
|
|
|
def test_keeps_recent_pending(self, tracker):
|
|
rec = tracker.record_transmission("recent", "ch", "channel")
|
|
key = int(rec.timestamp)
|
|
tracker.cleanup_old_records()
|
|
assert key in tracker.pending_transmissions
|
|
|
|
def test_removes_old_confirmed_without_repeats(self, tracker):
|
|
old_rec = TransmissionRecord(
|
|
timestamp=time.time() - 600,
|
|
content="old",
|
|
target="ch",
|
|
message_type="channel",
|
|
packet_hash="stale_hash",
|
|
)
|
|
tracker.confirmed_transmissions["stale_hash"] = old_rec
|
|
tracker.cleanup_old_records()
|
|
assert "stale_hash" not in tracker.confirmed_transmissions
|
|
|
|
def test_keeps_old_confirmed_with_repeats(self, tracker):
|
|
old_rec = TransmissionRecord(
|
|
timestamp=time.time() - 600,
|
|
content="old",
|
|
target="ch",
|
|
message_type="channel",
|
|
packet_hash="repeat_hash",
|
|
repeat_count=3,
|
|
)
|
|
tracker.confirmed_transmissions["repeat_hash"] = old_rec
|
|
tracker.cleanup_old_records()
|
|
assert "repeat_hash" in tracker.confirmed_transmissions
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _make_bot_with_device(mock_logger, pubkey, prefix_hex_chars=2):
|
|
"""Build a minimal bot mock whose meshcore.device.public_key == pubkey."""
|
|
device = Mock()
|
|
device.public_key = pubkey
|
|
meshcore = Mock()
|
|
meshcore.device = device
|
|
bot = Mock()
|
|
bot.logger = mock_logger
|
|
bot.meshcore = meshcore
|
|
bot.prefix_hex_chars = prefix_hex_chars
|
|
return bot
|
|
|
|
|
|
def _make_db_with_packet_stream(db_path: str) -> None:
|
|
"""Create a minimal packet_stream table in a SQLite file."""
|
|
with closing(sqlite3.connect(db_path)) as conn:
|
|
conn.execute("""
|
|
CREATE TABLE IF NOT EXISTS packet_stream (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
type TEXT,
|
|
timestamp REAL,
|
|
data TEXT
|
|
)
|
|
""")
|
|
conn.commit()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests for _update_bot_prefix (lines 57-67)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestBotPrefixFromSelfInfo:
|
|
"""meshcore_py reports the radio's key in self_info, after the tracker is created."""
|
|
|
|
def test_prefix_is_read_once_the_radio_reports_its_key(self, mock_logger):
|
|
bot = Mock()
|
|
bot.logger = mock_logger
|
|
bot.prefix_hex_chars = 2
|
|
bot.meshcore = Mock(self_info={})
|
|
tracker = TransmissionTracker(bot)
|
|
assert tracker.bot_prefix is None
|
|
bot.meshcore.self_info = {"public_key": "C3D4" + "00" * 30}
|
|
assert tracker.bot_prefix == "c3"
|
|
|
|
|
|
class TestBotPrefixFollowsReconnect:
|
|
def test_a_reconnect_to_another_radio_changes_the_prefix(self, mock_logger):
|
|
bot = Mock()
|
|
bot.logger = mock_logger
|
|
bot.prefix_hex_chars = 2
|
|
bot.meshcore = Mock(self_info={"public_key": "c3" + "00" * 31})
|
|
tracker = TransmissionTracker(bot)
|
|
assert tracker.bot_prefix == "c3"
|
|
bot.meshcore = Mock(self_info={"public_key": "e5" + "00" * 31})
|
|
assert tracker.bot_prefix == "e5"
|
|
|
|
|
|
class TestUpdateBotPrefix:
|
|
"""Cover lines 57-67: _update_bot_prefix with str and bytes public_key."""
|
|
|
|
def test_str_pubkey_sets_bot_prefix(self, mock_logger):
|
|
"""When public_key is a str, bot_prefix is set to its first prefix_hex_chars."""
|
|
bot = _make_bot_with_device(mock_logger, pubkey="abcdef1234")
|
|
tracker = TransmissionTracker(bot)
|
|
assert tracker.bot_prefix == "ab"
|
|
|
|
def test_str_pubkey_prefix_hex_chars_4(self, mock_logger):
|
|
"""prefix_hex_chars=4 slices the first 4 characters."""
|
|
bot = _make_bot_with_device(mock_logger, pubkey="deadbeef99", prefix_hex_chars=4)
|
|
tracker = TransmissionTracker(bot)
|
|
assert tracker.bot_prefix == "dead"
|
|
|
|
def test_bytes_pubkey_sets_bot_prefix(self, mock_logger):
|
|
"""When public_key is bytes, bot_prefix is the hex of the first byte."""
|
|
bot = _make_bot_with_device(mock_logger, pubkey=b"\xab\xcd\xef")
|
|
tracker = TransmissionTracker(bot)
|
|
assert tracker.bot_prefix == "ab"
|
|
|
|
def test_bytes_pubkey_zero_byte(self, mock_logger):
|
|
"""Bytes public key starting with 0x00 produces '00'."""
|
|
bot = _make_bot_with_device(mock_logger, pubkey=b"\x00\xff")
|
|
tracker = TransmissionTracker(bot)
|
|
assert tracker.bot_prefix == "00"
|
|
|
|
def test_str_pubkey_too_short_stays_none(self, mock_logger):
|
|
"""A one-character str pubkey does not satisfy len >= 2; prefix stays None."""
|
|
bot = _make_bot_with_device(mock_logger, pubkey="a")
|
|
tracker = TransmissionTracker(bot)
|
|
assert tracker.bot_prefix is None
|
|
|
|
def test_no_meshcore_leaves_prefix_none(self, mock_logger):
|
|
"""When bot.meshcore is None the prefix is never set."""
|
|
bot = Mock()
|
|
bot.logger = mock_logger
|
|
bot.meshcore = None
|
|
bot.prefix_hex_chars = 2
|
|
tracker = TransmissionTracker(bot)
|
|
assert tracker.bot_prefix is None
|
|
|
|
def test_exception_during_prefix_update_leaves_prefix_none(self, mock_logger):
|
|
"""If accessing device.public_key raises an exception bot_prefix remains None."""
|
|
device = Mock()
|
|
# Make accessing public_key raise an exception.
|
|
type(device).public_key = property(lambda s: (_ for _ in ()).throw(RuntimeError("boom")))
|
|
meshcore = Mock()
|
|
meshcore.device = device
|
|
bot = Mock()
|
|
bot.logger = mock_logger
|
|
bot.meshcore = meshcore
|
|
bot.prefix_hex_chars = 2
|
|
tracker = TransmissionTracker(bot)
|
|
assert tracker.bot_prefix is None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests for _update_command_in_database (lines 190, 198-246)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _build_tracker_with_db(mock_logger, tmp_path):
|
|
"""Build a tracker whose bot has a real SQLite DB at tmp_path/test.db."""
|
|
db_path = str(tmp_path / "test.db")
|
|
_make_db_with_packet_stream(db_path)
|
|
|
|
config = Mock()
|
|
config.has_section = Mock(return_value=False)
|
|
config.has_option = Mock(return_value=False)
|
|
config.get = Mock(return_value="")
|
|
|
|
db_manager = Mock()
|
|
db_manager.db_path = db_path
|
|
|
|
bot = Mock()
|
|
bot.logger = mock_logger
|
|
bot.meshcore = None
|
|
bot.prefix_hex_chars = 2
|
|
bot.config = config
|
|
bot.bot_root = str(tmp_path)
|
|
bot.db_manager = db_manager
|
|
bot.web_viewer_integration = Mock() # truthy so the DB path is reached
|
|
|
|
return TransmissionTracker(bot), db_path
|
|
|
|
|
|
class TestUpdateCommandInDatabase:
|
|
"""Cover lines 190 and 198-246: _update_command_in_database."""
|
|
|
|
def test_early_return_when_command_id_is_none(self, mock_logger, tmp_path):
|
|
"""Line 190: method returns immediately when record.command_id is None."""
|
|
tracker, db_path = _build_tracker_with_db(mock_logger, tmp_path)
|
|
rec = TransmissionRecord(
|
|
timestamp=time.time(),
|
|
content="msg",
|
|
target="ch",
|
|
message_type="channel",
|
|
command_id=None,
|
|
)
|
|
# No exception and the DB is untouched (table stays empty).
|
|
tracker._update_command_in_database(rec)
|
|
with closing(sqlite3.connect(db_path)) as conn:
|
|
count = conn.execute("SELECT COUNT(*) FROM packet_stream").fetchone()[0]
|
|
assert count == 0
|
|
|
|
def test_updates_matching_row_in_database(self, mock_logger, tmp_path):
|
|
"""Lines 198-246: a matching command row is found and updated."""
|
|
tracker, db_path = _build_tracker_with_db(mock_logger, tmp_path)
|
|
|
|
command_id = "cmd-update-test"
|
|
initial_data = {
|
|
"command_id": command_id,
|
|
"repeat_count": 0,
|
|
"repeater_prefixes": [],
|
|
"repeater_counts": {},
|
|
}
|
|
|
|
# Insert a row into packet_stream.
|
|
with closing(sqlite3.connect(db_path)) as conn:
|
|
conn.execute(
|
|
"INSERT INTO packet_stream (type, timestamp, data) VALUES (?, ?, ?)",
|
|
("command", time.time(), json.dumps(initial_data)),
|
|
)
|
|
conn.commit()
|
|
|
|
# Build a record whose command_id matches.
|
|
rec = TransmissionRecord(
|
|
timestamp=time.time(),
|
|
content="hello",
|
|
target="general",
|
|
message_type="channel",
|
|
command_id=command_id,
|
|
repeat_count=3,
|
|
)
|
|
rec.repeater_prefixes = {"7e", "ab"}
|
|
rec.repeater_counts = {"7e": 2, "ab": 1}
|
|
|
|
tracker._update_command_in_database(rec)
|
|
|
|
# Verify the row was updated.
|
|
with closing(sqlite3.connect(db_path)) as conn:
|
|
row = conn.execute(
|
|
"SELECT data FROM packet_stream WHERE type = 'command'"
|
|
).fetchone()
|
|
|
|
assert row is not None
|
|
updated = json.loads(row[0])
|
|
assert updated["repeat_count"] == 3
|
|
assert sorted(updated["repeater_prefixes"]) == ["7e", "ab"] or set(updated["repeater_prefixes"]) == {"7e", "ab"}
|
|
assert updated["repeater_counts"]["7e"] == 2
|
|
assert updated["repeater_counts"]["ab"] == 1
|
|
|
|
def test_no_matching_row_leaves_db_unchanged(self, mock_logger, tmp_path):
|
|
"""If no row matches command_id, the DB is not modified."""
|
|
tracker, db_path = _build_tracker_with_db(mock_logger, tmp_path)
|
|
|
|
other_data = {"command_id": "other-cmd", "repeat_count": 0}
|
|
with closing(sqlite3.connect(db_path)) as conn:
|
|
conn.execute(
|
|
"INSERT INTO packet_stream (type, timestamp, data) VALUES (?, ?, ?)",
|
|
("command", time.time(), json.dumps(other_data)),
|
|
)
|
|
conn.commit()
|
|
|
|
rec = TransmissionRecord(
|
|
timestamp=time.time(),
|
|
content="msg",
|
|
target="ch",
|
|
message_type="channel",
|
|
command_id="no-match-cmd",
|
|
repeat_count=1,
|
|
)
|
|
tracker._update_command_in_database(rec)
|
|
|
|
with closing(sqlite3.connect(db_path)) as conn:
|
|
row = conn.execute(
|
|
"SELECT data FROM packet_stream WHERE type = 'command'"
|
|
).fetchone()
|
|
assert json.loads(row[0])["command_id"] == "other-cmd"
|
|
|
|
def test_malformed_json_row_is_skipped(self, mock_logger, tmp_path):
|
|
"""A row with invalid JSON is silently skipped (json.JSONDecodeError path)."""
|
|
tracker, db_path = _build_tracker_with_db(mock_logger, tmp_path)
|
|
|
|
with closing(sqlite3.connect(db_path)) as conn:
|
|
conn.execute(
|
|
"INSERT INTO packet_stream (type, timestamp, data) VALUES (?, ?, ?)",
|
|
("command", time.time(), "NOT VALID JSON"),
|
|
)
|
|
conn.commit()
|
|
|
|
rec = TransmissionRecord(
|
|
timestamp=time.time(),
|
|
content="msg",
|
|
target="ch",
|
|
message_type="channel",
|
|
command_id="cmd-x",
|
|
repeat_count=1,
|
|
)
|
|
# Should not raise even though JSON is malformed.
|
|
tracker._update_command_in_database(rec)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests for line 314: path with parenthesis hop-count annotation
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestExtractRepeaterPrefixesParenPath:
|
|
"""Cover line 314: path containing '(' (hop-count annotation) is stripped."""
|
|
|
|
def test_path_with_paren_stripped_before_split(self, mock_logger):
|
|
"""'01,7e,86(3)' should extract '86' after stripping the parenthesised part."""
|
|
bot = Mock()
|
|
bot.logger = mock_logger
|
|
bot.meshcore = None
|
|
bot.prefix_hex_chars = 2
|
|
tracker = TransmissionTracker(bot)
|
|
tracker.bot_prefix = None
|
|
|
|
result = tracker.extract_repeater_prefixes_from_path("01,7e,86(3)")
|
|
assert result == ["86"]
|
|
|
|
def test_path_with_paren_and_via(self, mock_logger):
|
|
"""Combined annotation: ' via ROUTE_TYPE_*' and '(' in the path part."""
|
|
bot = Mock()
|
|
bot.logger = mock_logger
|
|
bot.meshcore = None
|
|
bot.prefix_hex_chars = 2
|
|
tracker = TransmissionTracker(bot)
|
|
tracker.bot_prefix = None
|
|
|
|
result = tracker.extract_repeater_prefixes_from_path("01,7e,ab(2) via ROUTE_TYPE_FLOOD")
|
|
assert result == ["ab"]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Automatic cleanup via _maybe_cleanup
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestMaybeCleanup:
|
|
"""Tests for automatic periodic cleanup in TransmissionTracker."""
|
|
|
|
def test_maybe_cleanup_runs_after_interval(self, tracker):
|
|
"""_maybe_cleanup runs cleanup_old_records when interval has elapsed."""
|
|
# Record an old transmission manually
|
|
old_record = TransmissionRecord(
|
|
timestamp=time.time() - 600, # 10 minutes ago (past cleanup_after=300s)
|
|
content="old", target="chan", message_type="channel",
|
|
)
|
|
tracker.pending_transmissions[int(old_record.timestamp)] = [old_record]
|
|
# Force the interval to have elapsed
|
|
tracker._last_cleanup_time = 0.0
|
|
tracker._maybe_cleanup()
|
|
# Old record should be cleaned up
|
|
assert int(old_record.timestamp) not in tracker.pending_transmissions
|
|
|
|
def test_maybe_cleanup_skips_within_interval(self, tracker):
|
|
"""_maybe_cleanup does NOT run cleanup if interval hasn't elapsed."""
|
|
old_record = TransmissionRecord(
|
|
timestamp=time.time() - 600,
|
|
content="old", target="chan", message_type="channel",
|
|
)
|
|
tracker.pending_transmissions[int(old_record.timestamp)] = [old_record]
|
|
# Set last cleanup to now — interval hasn't elapsed
|
|
tracker._last_cleanup_time = time.time()
|
|
tracker._maybe_cleanup()
|
|
# Old record should still be present (cleanup didn't run)
|
|
assert int(old_record.timestamp) in tracker.pending_transmissions
|
|
|
|
def test_record_transmission_triggers_cleanup(self, tracker):
|
|
"""record_transmission calls _maybe_cleanup, cleaning stale records."""
|
|
old_record = TransmissionRecord(
|
|
timestamp=time.time() - 600,
|
|
content="old", target="chan", message_type="channel",
|
|
)
|
|
old_key = int(old_record.timestamp)
|
|
tracker.pending_transmissions[old_key] = [old_record]
|
|
tracker._last_cleanup_time = 0.0 # Force cleanup to run
|
|
# Recording a new transmission should trigger cleanup
|
|
tracker.record_transmission("new msg", "general", "channel")
|
|
assert old_key not in tracker.pending_transmissions
|
|
|
|
|
|
class TestOnlyOurOwnPacketsMatch:
|
|
"""A pending transmission is claimed only by a packet carrying its identity."""
|
|
|
|
def test_an_unrelated_packet_does_not_claim_a_channel_send(self, tracker):
|
|
rec = tracker.record_transmission("hello", "general", "channel", channel_idx=0)
|
|
assert tracker.match_packet_hash("aaaa", rec.timestamp + 1, {"payload_type": 4, "payload_hex": "c1" * 40}) is None
|
|
assert tracker.match_packet_hash("bbbb", rec.timestamp + 1, _chan("someone else's text")) is None
|
|
assert tracker.match_packet_hash("cccc", rec.timestamp + 1, _chan("hello", idx=1)) is None
|
|
assert rec.packet_hash is None
|
|
assert tracker.match_packet_hash("dddd", rec.timestamp + 2, _chan("hello")) is rec
|
|
assert tracker.record_repeat("dddd", "7e") is True
|
|
|
|
def test_a_hash_without_packet_evidence_is_only_looked_up(self, tracker):
|
|
rec = tracker.record_transmission("hello", "general", "channel", channel_idx=0)
|
|
assert tracker.match_packet_hash("eeee", rec.timestamp + 1) is None
|
|
assert tracker.record_repeat("eeee", "7e") is False
|
|
assert rec.repeat_count == 0
|
|
|
|
def test_a_dm_is_matched_by_recipient_and_sender_hash(self, mock_bot):
|
|
mock_bot.meshcore = Mock(self_info={"public_key": "ab" + "00" * 31})
|
|
tracker = TransmissionTracker(mock_bot)
|
|
rec = tracker.record_transmission("hi", "Alice", "dm", recipient_key="c4" + "11" * 31)
|
|
other_recipient = {"payload_type": 2, "payload_hex": "d5ab" + "00" * 20}
|
|
other_sender = {"payload_type": 2, "payload_hex": "c4ee" + "00" * 20}
|
|
ours = {"payload_type": 2, "payload_hex": "c4ab" + "00" * 20}
|
|
assert tracker.match_packet_hash("1111", rec.timestamp, other_recipient) is None
|
|
assert tracker.match_packet_hash("2222", rec.timestamp, other_sender) is None
|
|
assert tracker.match_packet_hash("3333", rec.timestamp, ours) is rec
|
|
|
|
def test_a_trace_is_matched_by_its_tag(self, tracker):
|
|
tag = 0x12345678
|
|
rec = tracker.record_transmission("trace", "", "trace", command_id=str(tag), trace_tag=tag)
|
|
other = {"payload_type": 9, "payload_hex": (0x0BADF00D).to_bytes(4, "little").hex() + "00" * 5}
|
|
ours = {"payload_type": 9, "payload_hex": tag.to_bytes(4, "little").hex() + "00" * 5}
|
|
assert tracker.match_packet_hash("4444", rec.timestamp, other) is None
|
|
assert tracker.match_packet_hash("5555", rec.timestamp, ours) is rec
|
|
|
|
def test_a_record_without_identity_is_never_claimed(self, tracker):
|
|
rec = tracker.record_transmission("msg", "ch", "channel")
|
|
assert tracker.match_packet_hash("6666", rec.timestamp, _chan("msg")) is None
|
|
|
|
|
|
class TestRepeatCountWrites:
|
|
async def test_repeat_count_is_written_off_the_event_loop(self, tracker, monkeypatch):
|
|
import threading
|
|
|
|
threads = []
|
|
monkeypatch.setattr(tracker, "_update_command_in_database", lambda rec: threads.append(threading.current_thread().name))
|
|
tracker.bot.web_viewer_integration = Mock()
|
|
rec = tracker.record_transmission("msg", "ch", "channel", command_id="cmd-1", channel_idx=0)
|
|
tracker.match_packet_hash("7777", rec.timestamp, _chan("msg"))
|
|
assert tracker.record_repeat("7777", "7e") is True
|
|
tracker._db_executor.shutdown(wait=True)
|
|
assert threads and threads[0] != threading.current_thread().name
|
|
|
|
|
|
class TestChannelSenderIsChecked:
|
|
@pytest.fixture
|
|
def named_tracker(self, mock_bot):
|
|
mock_bot.meshcore = Mock(self_info={"name": "Bot [RF]: west", "public_key": "ab" * 32})
|
|
return TransmissionTracker(mock_bot)
|
|
|
|
def test_another_node_sending_the_same_text_does_not_claim_it(self, named_tracker):
|
|
rec = named_tracker.record_transmission("hello", "general", "channel", channel_idx=0)
|
|
assert named_tracker.match_packet_hash("8888", rec.timestamp, _chan("hello", sender="OtherNode")) is None
|
|
assert rec.packet_hash is None
|
|
|
|
def test_a_name_the_prefix_heuristic_cannot_split_still_matches(self, named_tracker):
|
|
rec = named_tracker.record_transmission("hello", "general", "channel", channel_idx=0)
|
|
packet = {"payload_type": 5, "payload_hex": "", "channel_idx": 0,
|
|
"channel_text": "Bot [RF]: west: hello", "channel_message": "Bot [RF]: west: hello"}
|
|
assert named_tracker.match_packet_hash("9999", rec.timestamp, packet) is rec
|
|
|
|
|
|
def test_close_waits_for_queued_writes_and_later_writes_run_inline(tracker, monkeypatch):
|
|
import threading
|
|
import time as _time
|
|
|
|
written = []
|
|
|
|
def slow_write(rec):
|
|
_time.sleep(0.05)
|
|
written.append(threading.current_thread().name)
|
|
|
|
monkeypatch.setattr(tracker, "_update_command_in_database", slow_write)
|
|
|
|
async def _queue():
|
|
tracker._schedule_command_update(TransmissionRecord(timestamp=0.0, content="x", target="c", message_type="channel"))
|
|
|
|
import asyncio
|
|
asyncio.run(_queue())
|
|
tracker.close()
|
|
assert len(written) == 1
|
|
tracker._schedule_command_update(TransmissionRecord(timestamp=0.0, content="y", target="c", message_type="channel"))
|
|
assert written[-1] == threading.current_thread().name
|