mirror of
https://github.com/agessaman/meshcore-bot.git
synced 2026-11-04 11:57:16 +00:00
agessaman
b750ad43f0
core.py conflicts were S3's simplifications of is_radio_offline, _record_send_failure and _record_send_success colliding with #313's rewrite of the same methods; dev's versions are kept.
1700 lines
63 KiB
Python
1700 lines
63 KiB
Python
"""Tests for MeshCoreBot logic (config loading, radio settings, helpers)."""
|
|
|
|
import asyncio
|
|
import socket
|
|
import sqlite3
|
|
import struct
|
|
import threading
|
|
import time
|
|
from pathlib import Path
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from modules.core import MeshCoreBot
|
|
from modules.i18n import Translator
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _write_config(path: Path, db_path: Path, extra: str = "") -> None:
|
|
path.write_text(
|
|
f"""[Connection]
|
|
connection_type = serial
|
|
serial_port = /dev/ttyUSB0
|
|
timeout = 30
|
|
|
|
[Bot]
|
|
db_path = {db_path.as_posix()}
|
|
prefix_bytes = 1
|
|
|
|
[Channels]
|
|
monitor_channels = #general
|
|
{extra}
|
|
""",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def bot(tmp_path):
|
|
"""Minimal MeshCoreBot from a temporary config file."""
|
|
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))
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# bot_root property
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestBotRoot:
|
|
def test_returns_config_directory(self, tmp_path):
|
|
config_file = tmp_path / "config.ini"
|
|
db_path = tmp_path / "bot.db"
|
|
_write_config(config_file, db_path)
|
|
bot = MeshCoreBot(config_file=str(config_file))
|
|
assert bot.bot_root == tmp_path.resolve()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _get_radio_settings
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestGetRadioSettings:
|
|
def test_returns_dict_with_expected_keys(self, bot):
|
|
settings = bot._get_radio_settings()
|
|
assert "connection_type" in settings
|
|
assert "serial_port" in settings
|
|
assert "ble_device_name" in settings
|
|
assert "hostname" in settings
|
|
assert "tcp_port" in settings
|
|
assert "timeout" in settings
|
|
|
|
def test_reads_connection_type(self, bot):
|
|
assert bot._get_radio_settings()["connection_type"] == "serial"
|
|
|
|
def test_reads_serial_port(self, bot):
|
|
assert bot._get_radio_settings()["serial_port"] == "/dev/ttyUSB0"
|
|
|
|
def test_reads_timeout(self, bot):
|
|
assert bot._get_radio_settings()["timeout"] == 30
|
|
|
|
def test_defaults_when_missing(self, tmp_path):
|
|
config_file = tmp_path / "config.ini"
|
|
db_path = tmp_path / "bot.db"
|
|
# Write config without optional fields
|
|
config_file.write_text(
|
|
f"""[Connection]
|
|
connection_type = ble
|
|
|
|
[Bot]
|
|
db_path = {db_path.as_posix()}
|
|
|
|
[Channels]
|
|
monitor_channels = #general
|
|
""",
|
|
encoding="utf-8",
|
|
)
|
|
b = MeshCoreBot(config_file=str(config_file))
|
|
settings = b._get_radio_settings()
|
|
assert settings["ble_device_name"] == ""
|
|
assert settings["hostname"] == ""
|
|
assert settings["tcp_port"] == 5000
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# reload_config
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestReloadConfig:
|
|
def test_reload_succeeds_with_same_radio_settings(self, tmp_path):
|
|
config_file = tmp_path / "config.ini"
|
|
db_path = tmp_path / "bot.db"
|
|
_write_config(config_file, db_path)
|
|
b = MeshCoreBot(config_file=str(config_file))
|
|
success, msg = b.reload_config()
|
|
assert success is True
|
|
|
|
def test_reload_fails_when_radio_settings_changed(self, tmp_path):
|
|
config_file = tmp_path / "config.ini"
|
|
db_path = tmp_path / "bot.db"
|
|
_write_config(config_file, db_path)
|
|
b = MeshCoreBot(config_file=str(config_file))
|
|
# Change serial port in config file
|
|
_write_config(config_file, db_path, extra="")
|
|
config_file.write_text(
|
|
f"""[Connection]
|
|
connection_type = serial
|
|
serial_port = /dev/ttyUSB1
|
|
timeout = 30
|
|
|
|
[Bot]
|
|
db_path = {db_path.as_posix()}
|
|
|
|
[Channels]
|
|
monitor_channels = #general
|
|
""",
|
|
encoding="utf-8",
|
|
)
|
|
success, msg = b.reload_config()
|
|
assert success is False
|
|
assert "serial_port" in msg.lower() or "restart" in msg.lower()
|
|
|
|
def test_reload_config_not_found_returns_false(self, tmp_path):
|
|
config_file = tmp_path / "config.ini"
|
|
db_path = tmp_path / "bot.db"
|
|
_write_config(config_file, db_path)
|
|
b = MeshCoreBot(config_file=str(config_file))
|
|
# Delete the config file
|
|
config_file.unlink()
|
|
success, msg = b.reload_config()
|
|
assert success is False
|
|
|
|
def test_reload_drops_keys_deleted_from_file(self, tmp_path):
|
|
"""ConfigParser.read() merges; reload must not resurrect deleted keys.
|
|
|
|
The web settings UI deletes rows (e.g. announce.* triggers) from
|
|
config.ini — a reload afterwards has to reflect the deletion.
|
|
"""
|
|
config_file = tmp_path / "config.ini"
|
|
db_path = tmp_path / "bot.db"
|
|
_write_config(config_file, db_path, extra="[Channels_List]\nseattle = Seattle chat\n")
|
|
b = MeshCoreBot(config_file=str(config_file))
|
|
assert b.config.get("Channels_List", "seattle") == "Seattle chat"
|
|
# Remove the key (and its whole section) from the file.
|
|
_write_config(config_file, db_path)
|
|
success, _ = b.reload_config()
|
|
assert success is True
|
|
assert not b.config.has_section("Channels_List") or not b.config.has_option(
|
|
"Channels_List", "seattle"
|
|
)
|
|
|
|
def test_reload_reinstantiates_command_plugins(self, tmp_path):
|
|
"""Command settings are cached in __init__; reload must refresh them."""
|
|
config_file = tmp_path / "config.ini"
|
|
db_path = tmp_path / "bot.db"
|
|
_write_config(config_file, db_path, extra="[Ping_Command]\nenabled = true\n")
|
|
b = MeshCoreBot(config_file=str(config_file))
|
|
if not hasattr(b, "command_manager"):
|
|
pytest.skip("bot built without command_manager")
|
|
before = b.command_manager.commands.get("ping")
|
|
_write_config(config_file, db_path, extra="[Ping_Command]\nenabled = false\n")
|
|
success, _ = b.reload_config()
|
|
assert success is True
|
|
after = b.command_manager.commands.get("ping")
|
|
assert after is not None and after is not before
|
|
assert b.config.getboolean("Ping_Command", "enabled") is False
|
|
|
|
def test_reload_replaces_translator_cache_atomically(self, tmp_path):
|
|
config_file = tmp_path / "config.ini"
|
|
db_path = tmp_path / "bot.db"
|
|
_write_config(
|
|
config_file,
|
|
db_path,
|
|
extra="[Localization]\nlanguage = en\ntranslation_path = translations/\n",
|
|
)
|
|
b = MeshCoreBot(config_file=str(config_file))
|
|
old_cache = b._translator_cache
|
|
|
|
_write_config(
|
|
config_file,
|
|
db_path,
|
|
extra="[Localization]\nlanguage = de\ntranslation_path = translations/\n",
|
|
)
|
|
success, _ = b.reload_config()
|
|
|
|
assert success is True
|
|
assert b.translation_path == "translations/"
|
|
assert b.translator.language == "de"
|
|
assert b._translator_cache == {"de": b.translator}
|
|
assert b._translator_cache is not old_cache
|
|
|
|
def test_failed_reload_restores_translator_cache(self, tmp_path):
|
|
config_file = tmp_path / "config.ini"
|
|
db_path = tmp_path / "bot.db"
|
|
_write_config(
|
|
config_file,
|
|
db_path,
|
|
extra="[Localization]\nlanguage = en\ntranslation_path = translations/\n",
|
|
)
|
|
b = MeshCoreBot(config_file=str(config_file))
|
|
old_translator = b.translator
|
|
old_path = b.translation_path
|
|
old_cache = b._translator_cache
|
|
|
|
_write_config(
|
|
config_file,
|
|
db_path,
|
|
extra="[Localization]\nlanguage = de\ntranslation_path = translations/\n",
|
|
)
|
|
with patch("modules.core.CommandManager", side_effect=RuntimeError("boom")):
|
|
success, _ = b.reload_config()
|
|
|
|
assert success is False
|
|
assert b.translator is old_translator
|
|
assert b.translation_path == old_path
|
|
assert b._translator_cache is old_cache
|
|
|
|
|
|
class TestResponseTranslators:
|
|
def test_locale_only_catalog_is_available_by_base_language(
|
|
self, tmp_path, monkeypatch
|
|
):
|
|
translations = tmp_path / "translations"
|
|
translations.mkdir()
|
|
(translations / "en.json").write_text(
|
|
'{"commands": {"hello": {"response_format": "Hello"}}}',
|
|
encoding="utf-8",
|
|
)
|
|
(translations / "fr-CA.json").write_text(
|
|
'{"commands": {"hello": {"response_format": "Allo"}}}',
|
|
encoding="utf-8",
|
|
)
|
|
monkeypatch.chdir(tmp_path)
|
|
|
|
b = object.__new__(MeshCoreBot)
|
|
b.logger = MagicMock()
|
|
b.translation_path = str(translations)
|
|
b.local_translation_path = str(tmp_path / "local" / "translations")
|
|
b.translator = Translator("en", b.translation_path, b.local_translation_path)
|
|
b._translator_cache = {"en": b.translator}
|
|
|
|
assert b.available_languages() == {"en", "fr"}
|
|
french = b.get_translator("fr")
|
|
assert french.language == "fr-CA"
|
|
assert french.translate("commands.hello.response_format") == "Allo"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# key_prefix / is_valid_prefix
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestKeyPrefixHelpers:
|
|
def test_key_prefix_returns_first_n_chars(self, bot):
|
|
assert bot.key_prefix("deadbeef1234") == "de"
|
|
|
|
def test_key_prefix_uses_prefix_hex_chars(self, bot):
|
|
bot.prefix_hex_chars = 4
|
|
assert bot.key_prefix("deadbeef1234") == "dead"
|
|
|
|
def test_is_valid_prefix_correct_length(self, bot):
|
|
assert bot.is_valid_prefix("de") is True
|
|
|
|
def test_is_valid_prefix_wrong_length(self, bot):
|
|
assert bot.is_valid_prefix("d") is False
|
|
assert bot.is_valid_prefix("dead") is False
|
|
|
|
def test_prefix_hex_chars_from_config(self, tmp_path):
|
|
config_file = tmp_path / "config2.ini"
|
|
db_path = tmp_path / "bot2.db"
|
|
config_file.write_text(
|
|
f"""[Connection]
|
|
connection_type = ble
|
|
|
|
[Bot]
|
|
db_path = {db_path.as_posix()}
|
|
prefix_bytes = 2
|
|
|
|
[Channels]
|
|
monitor_channels = #general
|
|
""",
|
|
encoding="utf-8",
|
|
)
|
|
b = MeshCoreBot(config_file=str(config_file))
|
|
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)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _probe_radio_health (PR4 — zombie-connection detection)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestProbeRadioHealth:
|
|
"""Tests for MeshCoreBot._probe_radio_health()."""
|
|
|
|
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 test_returns_false_when_meshcore_is_none(self, tmp_path):
|
|
bot = self._make_bot(tmp_path)
|
|
bot.connected = True
|
|
bot.meshcore = None
|
|
scheduled = []
|
|
|
|
async def fake_schedule(reason):
|
|
scheduled.append(reason)
|
|
|
|
bot._schedule_transport_reconnect = fake_schedule
|
|
result = asyncio.run(bot._probe_radio_health())
|
|
assert result is False
|
|
assert scheduled == ['probe_not_connected']
|
|
|
|
def test_returns_false_when_not_connected(self, tmp_path):
|
|
bot = self._make_bot(tmp_path)
|
|
bot.connected = True
|
|
bot.meshcore = MagicMock()
|
|
bot.meshcore.is_connected = False
|
|
scheduled = []
|
|
|
|
async def fake_schedule(reason):
|
|
scheduled.append(reason)
|
|
|
|
bot._schedule_transport_reconnect = fake_schedule
|
|
result = asyncio.run(bot._probe_radio_health())
|
|
assert result is False
|
|
assert scheduled == ['probe_not_connected']
|
|
|
|
def test_error_event_increments_fail_count(self, tmp_path):
|
|
from meshcore.events import EventType
|
|
bot = self._make_bot(tmp_path)
|
|
bot._radio_fail_count = 0
|
|
|
|
error_event = MagicMock()
|
|
error_event.type = EventType.ERROR
|
|
error_event.payload = {"reason": "no_event_received"}
|
|
|
|
bot.meshcore = MagicMock()
|
|
bot.meshcore.is_connected = True
|
|
bot.meshcore.commands.get_time = MagicMock(
|
|
return_value=_make_coro(error_event)
|
|
)
|
|
|
|
result = asyncio.run(bot._probe_radio_health())
|
|
assert result is False
|
|
assert bot._radio_fail_count == 1
|
|
|
|
def test_error_event_below_threshold_does_not_reconnect(self, tmp_path):
|
|
from meshcore.events import EventType
|
|
bot = self._make_bot(tmp_path)
|
|
bot._radio_fail_count = 0
|
|
bot.config.set("Bot", "radio_probe_fail_threshold", "3")
|
|
|
|
error_event = MagicMock()
|
|
error_event.type = EventType.ERROR
|
|
error_event.payload = {"reason": "no_event_received"}
|
|
|
|
bot.meshcore = MagicMock()
|
|
bot.meshcore.is_connected = True
|
|
bot.meshcore.commands.get_time = MagicMock(
|
|
return_value=_make_coro(error_event)
|
|
)
|
|
|
|
reconnect_called = []
|
|
|
|
async def fake_reconnect():
|
|
reconnect_called.append(True)
|
|
return True
|
|
|
|
bot.reconnect_radio = fake_reconnect
|
|
|
|
asyncio.run(bot._probe_radio_health())
|
|
assert not reconnect_called
|
|
|
|
def test_error_event_at_threshold_triggers_zombie_detection(self, tmp_path):
|
|
"""At threshold, zombie state is set and CRITICAL logged — no reconnect attempt."""
|
|
from meshcore.events import EventType
|
|
bot = self._make_bot(tmp_path)
|
|
bot._radio_fail_count = 2 # this probe makes it 3
|
|
bot.config.set("Bot", "radio_probe_fail_threshold", "3")
|
|
|
|
error_event = MagicMock()
|
|
error_event.type = EventType.ERROR
|
|
error_event.payload = {"reason": "no_event_received"}
|
|
|
|
bot.meshcore = MagicMock()
|
|
bot.meshcore.is_connected = True
|
|
bot.meshcore.commands.get_time = MagicMock(
|
|
return_value=_make_coro(error_event)
|
|
)
|
|
|
|
reconnect_called = []
|
|
|
|
async def fake_reconnect():
|
|
reconnect_called.append(True)
|
|
return True
|
|
|
|
bot.reconnect_radio = fake_reconnect
|
|
|
|
result = asyncio.run(bot._probe_radio_health())
|
|
assert result is False
|
|
assert bot._radio_fail_count == 0 # reset after trigger
|
|
assert bot.is_radio_zombie is True # zombie flag set
|
|
assert not reconnect_called # reconnect NOT called — power cycle needed
|
|
|
|
def test_success_resets_fail_count(self, tmp_path):
|
|
from meshcore.events import EventType
|
|
bot = self._make_bot(tmp_path)
|
|
bot._radio_fail_count = 2
|
|
|
|
ok_event = MagicMock()
|
|
ok_event.type = EventType.OK
|
|
|
|
bot.meshcore = MagicMock()
|
|
bot.meshcore.is_connected = True
|
|
bot.meshcore.commands.get_time = MagicMock(
|
|
return_value=_make_coro(ok_event)
|
|
)
|
|
|
|
result = asyncio.run(bot._probe_radio_health())
|
|
assert result is True
|
|
assert bot._radio_fail_count == 0
|
|
|
|
def test_success_logs_recovery_when_previously_failed(self, tmp_path):
|
|
from meshcore.events import EventType
|
|
bot = self._make_bot(tmp_path)
|
|
bot._radio_fail_count = 1
|
|
|
|
ok_event = MagicMock()
|
|
ok_event.type = EventType.OK
|
|
|
|
bot.meshcore = MagicMock()
|
|
bot.meshcore.is_connected = True
|
|
bot.meshcore.commands.get_time = MagicMock(
|
|
return_value=_make_coro(ok_event)
|
|
)
|
|
|
|
with patch.object(bot.logger, "info") as mock_info:
|
|
asyncio.run(bot._probe_radio_health())
|
|
|
|
logged_messages = [str(c) for c in mock_info.call_args_list]
|
|
assert any("recovered" in m for m in logged_messages)
|
|
|
|
def test_success_no_recovery_log_when_no_prior_failure(self, tmp_path):
|
|
from meshcore.events import EventType
|
|
bot = self._make_bot(tmp_path)
|
|
bot._radio_fail_count = 0
|
|
|
|
ok_event = MagicMock()
|
|
ok_event.type = EventType.OK
|
|
|
|
bot.meshcore = MagicMock()
|
|
bot.meshcore.is_connected = True
|
|
bot.meshcore.commands.get_time = MagicMock(
|
|
return_value=_make_coro(ok_event)
|
|
)
|
|
|
|
with patch.object(bot.logger, "info") as mock_info:
|
|
asyncio.run(bot._probe_radio_health())
|
|
|
|
logged_messages = [str(c) for c in mock_info.call_args_list]
|
|
assert not any("recovered" in m for m in logged_messages)
|
|
|
|
def test_asyncio_timeout_returns_false_and_warns(self, tmp_path):
|
|
bot = self._make_bot(tmp_path)
|
|
bot.meshcore = MagicMock()
|
|
bot.meshcore.is_connected = True
|
|
|
|
async def slow_get_time():
|
|
raise asyncio.TimeoutError()
|
|
|
|
bot.meshcore.commands.get_time = MagicMock(return_value=slow_get_time())
|
|
|
|
async def run():
|
|
async def fake_wait_for(coro, timeout):
|
|
raise asyncio.TimeoutError()
|
|
|
|
with patch("asyncio.wait_for", side_effect=fake_wait_for):
|
|
return await bot._probe_radio_health()
|
|
|
|
result = asyncio.run(run())
|
|
assert result is False
|
|
|
|
def test_generic_exception_returns_false_and_warns(self, tmp_path):
|
|
bot = self._make_bot(tmp_path)
|
|
bot.meshcore = MagicMock()
|
|
bot.meshcore.is_connected = True
|
|
|
|
async def failing_get_time():
|
|
raise RuntimeError("device exploded")
|
|
|
|
bot.meshcore.commands.get_time = MagicMock(
|
|
return_value=failing_get_time()
|
|
)
|
|
|
|
with patch.object(bot.logger, "warning") as mock_warn:
|
|
result = asyncio.run(bot._probe_radio_health())
|
|
|
|
assert result is False
|
|
assert mock_warn.called
|
|
|
|
def test_tcp_probe_error_at_threshold_schedules_reconnect_not_zombie(self, tmp_path):
|
|
from meshcore.events import EventType
|
|
|
|
bot = self._make_bot(tmp_path)
|
|
bot.config.set("Connection", "connection_type", "tcp")
|
|
bot.connected = True
|
|
bot._tcp_probe_fail_count = 2
|
|
bot.config.set("Connection", "radio_probe_fail_threshold", "3")
|
|
|
|
error_event = MagicMock()
|
|
error_event.type = EventType.ERROR
|
|
error_event.payload = {"reason": "no_event_received"}
|
|
|
|
bot.meshcore = MagicMock()
|
|
bot.meshcore.is_connected = True
|
|
bot.meshcore.commands.get_time = MagicMock(return_value=_make_coro(error_event))
|
|
|
|
scheduled = []
|
|
|
|
async def fake_schedule(reason):
|
|
scheduled.append(reason)
|
|
|
|
bot._schedule_transport_reconnect = fake_schedule
|
|
|
|
result = asyncio.run(bot._probe_radio_health())
|
|
assert result is False
|
|
assert bot.is_radio_zombie is False
|
|
assert scheduled == ['tcp_probe_failed']
|
|
|
|
def test_tcp_probe_timeout_at_threshold_schedules_reconnect(self, tmp_path):
|
|
bot = self._make_bot(tmp_path)
|
|
bot.config.set("Connection", "connection_type", "tcp")
|
|
bot.connected = True
|
|
bot._tcp_probe_fail_count = 2
|
|
bot.config.set("Connection", "radio_probe_fail_threshold", "3")
|
|
|
|
bot.meshcore = MagicMock()
|
|
bot.meshcore.is_connected = True
|
|
bot.meshcore.commands.get_time = MagicMock(return_value=_make_coro(None))
|
|
|
|
scheduled = []
|
|
|
|
async def fake_schedule(reason):
|
|
scheduled.append(reason)
|
|
|
|
bot._schedule_transport_reconnect = fake_schedule
|
|
|
|
async def run():
|
|
async def fake_wait_for(coro, timeout):
|
|
raise asyncio.TimeoutError()
|
|
|
|
with patch("asyncio.wait_for", side_effect=fake_wait_for):
|
|
return await bot._probe_radio_health()
|
|
|
|
result = asyncio.run(run())
|
|
assert result is False
|
|
assert scheduled == ['tcp_probe_failed']
|
|
|
|
|
|
class TestTransportReconnect:
|
|
"""Transport reconnect scheduling (TCP/serial/BLE)."""
|
|
|
|
def _make_bot(self, tmp_path: Path, connection_type: str = "serial") -> MeshCoreBot:
|
|
config_file = tmp_path / "config.ini"
|
|
db_path = tmp_path / "bot.db"
|
|
_write_config(config_file, db_path)
|
|
bot = MeshCoreBot(config_file=str(config_file))
|
|
if connection_type == "tcp":
|
|
bot.config.set("Connection", "connection_type", "tcp")
|
|
bot.config.set("Connection", "hostname", "127.0.0.1")
|
|
bot.connected = True
|
|
return bot
|
|
|
|
def test_schedule_skips_manual_disconnect(self, tmp_path):
|
|
bot = self._make_bot(tmp_path)
|
|
asyncio.run(bot._schedule_transport_reconnect("manual_disconnect"))
|
|
assert bot._transport_reconnect_in_progress is False
|
|
|
|
def test_schedule_skips_when_already_in_progress(self, tmp_path):
|
|
bot = self._make_bot(tmp_path)
|
|
bot._transport_reconnect_in_progress = True
|
|
asyncio.run(bot._schedule_transport_reconnect("tcp_disconnect"))
|
|
assert bot._transport_reconnect_in_progress is True
|
|
|
|
def test_schedule_sets_in_progress_and_metadata(self, tmp_path):
|
|
bot = self._make_bot(tmp_path)
|
|
with patch.object(bot, "_update_radio_connected_metadata") as mock_meta:
|
|
with patch("asyncio.create_task") as mock_create_task:
|
|
|
|
def swallow_task(coro):
|
|
# Scheduling is mocked; close coroutine so it is not left unawaited.
|
|
coro.close()
|
|
return MagicMock()
|
|
|
|
mock_create_task.side_effect = swallow_task
|
|
asyncio.run(bot._schedule_transport_reconnect("tcp_disconnect"))
|
|
assert bot._transport_reconnect_in_progress is True
|
|
mock_meta.assert_called_once_with(False)
|
|
mock_create_task.assert_called_once()
|
|
|
|
def test_run_transport_reconnect_clears_in_progress_on_failure(self, tmp_path):
|
|
bot = self._make_bot(tmp_path)
|
|
bot._transport_reconnect_in_progress = True
|
|
|
|
async def fake_attempt():
|
|
return False
|
|
|
|
bot._attempt_reconnect = fake_attempt
|
|
asyncio.run(bot._run_transport_reconnect())
|
|
assert bot._transport_reconnect_in_progress is False
|
|
assert bot.connected is False
|
|
|
|
def test_disconnected_event_schedules_reconnect(self, tmp_path):
|
|
from meshcore.events import EventType
|
|
|
|
bot = self._make_bot(tmp_path)
|
|
bot.meshcore = MagicMock()
|
|
scheduled = []
|
|
|
|
async def fake_schedule(reason):
|
|
scheduled.append(reason)
|
|
|
|
bot._schedule_transport_reconnect = fake_schedule
|
|
bot.message_handler = MagicMock()
|
|
|
|
async def run_handlers():
|
|
await bot.setup_message_handlers()
|
|
disconnect_cb = None
|
|
for call in bot.meshcore.subscribe.call_args_list:
|
|
if call[0][0] == EventType.DISCONNECTED:
|
|
disconnect_cb = call[0][1]
|
|
break
|
|
assert disconnect_cb is not None
|
|
event = MagicMock()
|
|
event.payload = {"reason": "tcp_disconnect"}
|
|
await disconnect_cb(event)
|
|
|
|
async def noop_fetch():
|
|
return None
|
|
|
|
bot.meshcore.start_auto_message_fetching = noop_fetch
|
|
|
|
async def instant_sleep(*_args, **_kwargs):
|
|
return None
|
|
|
|
with patch("asyncio.sleep", instant_sleep):
|
|
asyncio.run(run_handlers())
|
|
|
|
assert scheduled == ["tcp_disconnect"]
|
|
|
|
def test_notify_services_transport_reconnected_only_running(self, tmp_path):
|
|
bot = self._make_bot(tmp_path)
|
|
running_svc = MagicMock()
|
|
running_svc.is_running.return_value = True
|
|
running_svc.on_transport_reconnected = AsyncMock()
|
|
|
|
stopped_svc = MagicMock()
|
|
stopped_svc.is_running.return_value = False
|
|
stopped_svc.on_transport_reconnected = AsyncMock()
|
|
|
|
bot.services = {"running": running_svc, "stopped": stopped_svc}
|
|
asyncio.run(bot._notify_services_transport_reconnected())
|
|
|
|
running_svc.on_transport_reconnected.assert_awaited_once()
|
|
stopped_svc.on_transport_reconnected.assert_not_awaited()
|
|
|
|
def test_keep_running_through_transport_reconnect(self, tmp_path):
|
|
"""A reconnect clears connected; the main loop and scheduler must not treat that as a stop."""
|
|
bot = self._make_bot(tmp_path, connection_type="tcp")
|
|
bot.meshcore = None
|
|
# _schedule_transport_reconnect sets this before handing off to the task
|
|
bot._transport_reconnect_in_progress = True
|
|
seen_during_connect = []
|
|
|
|
async def fake_connect():
|
|
seen_during_connect.append((bot.connected, bot.keep_running))
|
|
bot.connected = True
|
|
return True
|
|
|
|
bot.connect = fake_connect
|
|
asyncio.run(bot._run_transport_reconnect())
|
|
|
|
assert seen_during_connect == [(False, True)]
|
|
assert bot._transport_reconnect_in_progress is False
|
|
assert bot.keep_running is True
|
|
|
|
def test_keep_running_false_after_reconnect_gives_up(self, tmp_path):
|
|
bot = self._make_bot(tmp_path)
|
|
bot._transport_reconnect_in_progress = True
|
|
|
|
async def fake_attempt():
|
|
bot.connected = False
|
|
return False
|
|
|
|
bot._attempt_reconnect = fake_attempt
|
|
asyncio.run(bot._run_transport_reconnect())
|
|
assert bot.keep_running is False
|
|
|
|
def test_keep_running_false_on_shutdown_mid_reconnect(self, tmp_path):
|
|
bot = self._make_bot(tmp_path)
|
|
bot.connected = False
|
|
bot._transport_reconnect_in_progress = True
|
|
assert bot.keep_running is True
|
|
bot._shutdown_event.set()
|
|
assert bot.keep_running is False
|
|
|
|
def test_keep_running_through_web_viewer_reconnect(self, tmp_path):
|
|
bot = self._make_bot(tmp_path)
|
|
bot.meshcore = MagicMock()
|
|
bot.meshcore.disconnect = AsyncMock()
|
|
seen_during_connect = []
|
|
|
|
async def fake_connect():
|
|
seen_during_connect.append((bot.connected, bot.keep_running))
|
|
return False
|
|
|
|
bot.connect = fake_connect
|
|
assert asyncio.run(bot.reconnect_radio()) is False
|
|
assert seen_during_connect == [(False, True)]
|
|
assert bot._radio_relinks_in_progress == 0
|
|
# A failed manual reconnect still ends the loops, as before
|
|
assert bot.keep_running is False
|
|
|
|
def test_failed_post_connect_initialization_cleans_up_and_stops(self, tmp_path):
|
|
bot = self._make_bot(tmp_path)
|
|
old_meshcore = MagicMock()
|
|
old_meshcore.disconnect = AsyncMock()
|
|
bot.meshcore = old_meshcore
|
|
|
|
new_meshcore = MagicMock()
|
|
new_meshcore.is_connected = True
|
|
new_meshcore.self_info = {}
|
|
new_meshcore.disconnect = AsyncMock()
|
|
|
|
bot.wait_for_contacts = AsyncMock()
|
|
bot.channel_manager.fetch_channels = AsyncMock()
|
|
bot.setup_message_handlers = AsyncMock(
|
|
side_effect=ConnectionError("message setup failed")
|
|
)
|
|
|
|
with patch(
|
|
"modules.core.meshcore.MeshCore.create_serial",
|
|
AsyncMock(return_value=new_meshcore),
|
|
):
|
|
assert asyncio.run(bot.reconnect_radio()) is False
|
|
|
|
old_meshcore.disconnect.assert_awaited_once()
|
|
new_meshcore.disconnect.assert_awaited_once()
|
|
assert bot.meshcore is None
|
|
assert bot.connected is False
|
|
assert bot._radio_relinks_in_progress == 0
|
|
assert bot.keep_running is False
|
|
|
|
def test_connect_fails_when_channel_fetch_never_returns_channels(self, tmp_path):
|
|
bot = self._make_bot(tmp_path, connection_type="serial")
|
|
new_meshcore = MagicMock()
|
|
new_meshcore.is_connected = True
|
|
new_meshcore.self_info = {}
|
|
new_meshcore.disconnect = AsyncMock()
|
|
|
|
bot.wait_for_contacts = AsyncMock()
|
|
bot.channel_manager.fetch_channels = AsyncMock(return_value=False)
|
|
bot.setup_message_handlers = AsyncMock()
|
|
|
|
with patch(
|
|
"modules.core.meshcore.MeshCore.create_serial",
|
|
AsyncMock(return_value=new_meshcore),
|
|
):
|
|
assert asyncio.run(bot.connect()) is False
|
|
|
|
new_meshcore.disconnect.assert_awaited_once()
|
|
bot.setup_message_handlers.assert_not_awaited()
|
|
assert bot.meshcore is None
|
|
assert bot.connected is False
|
|
|
|
def test_keep_running_through_web_viewer_reboot(self, tmp_path):
|
|
bot = self._make_bot(tmp_path)
|
|
bot.meshcore = MagicMock()
|
|
bot.meshcore.is_connected = True
|
|
bot.meshcore.commands.reboot = AsyncMock()
|
|
bot.meshcore.disconnect = AsyncMock()
|
|
seen = []
|
|
|
|
async def fake_connect():
|
|
bot.connected = True
|
|
return True
|
|
|
|
async def instant_sleep(*_args, **_kwargs):
|
|
seen.append((bot.connected, bot.keep_running))
|
|
|
|
bot.connect = fake_connect
|
|
with patch("asyncio.sleep", instant_sleep):
|
|
assert asyncio.run(bot.reboot_radio()) is True
|
|
assert seen == [(False, True)]
|
|
assert bot._radio_relinks_in_progress == 0
|
|
assert bot.keep_running is True
|
|
|
|
|
|
class TestRadioOfflineState:
|
|
"""Tests for _record_send_failure / _record_send_success / is_radio_offline."""
|
|
|
|
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 test_is_radio_offline_defaults_to_false(self, tmp_path):
|
|
bot = self._make_bot(tmp_path)
|
|
assert bot.is_radio_offline is False
|
|
|
|
def test_record_send_failure_increments_counter(self, tmp_path):
|
|
bot = self._make_bot(tmp_path)
|
|
bot._record_send_failure()
|
|
assert bot._send_consecutive_failures == 1
|
|
|
|
def test_record_send_failure_sets_offline_at_threshold(self, tmp_path):
|
|
bot = self._make_bot(tmp_path)
|
|
# Default threshold is 3
|
|
for _ in range(3):
|
|
bot._record_send_failure()
|
|
assert bot.is_radio_offline is True
|
|
|
|
def test_record_send_success_clears_offline_flag(self, tmp_path):
|
|
bot = self._make_bot(tmp_path)
|
|
bot._radio_offline = True
|
|
bot._radio_offline_trial = 'in_flight'
|
|
bot._radio_offline_trial_id = 1
|
|
bot._send_consecutive_failures = 5
|
|
bot._record_send_success(1)
|
|
assert bot.is_radio_offline is False
|
|
assert bot._send_consecutive_failures == 0
|
|
|
|
def test_success_of_a_send_that_was_not_the_trial_leaves_the_outage(self, tmp_path):
|
|
bot = self._make_bot(tmp_path)
|
|
bot._radio_offline = True
|
|
bot._record_send_success()
|
|
assert bot.is_radio_offline is True
|
|
|
|
def test_record_send_success_no_op_when_already_clean(self, tmp_path):
|
|
bot = self._make_bot(tmp_path)
|
|
bot._record_send_success() # must not raise
|
|
assert bot.is_radio_offline is False
|
|
|
|
def test_offline_not_set_below_threshold(self, tmp_path):
|
|
bot = self._make_bot(tmp_path)
|
|
bot._record_send_failure()
|
|
bot._record_send_failure()
|
|
assert bot.is_radio_offline is False
|
|
|
|
def test_custom_threshold_from_config(self, tmp_path):
|
|
bot = self._make_bot(tmp_path)
|
|
bot.config.set('Bot', 'radio_offline_threshold', '2')
|
|
bot._record_send_failure()
|
|
assert bot.is_radio_offline is False
|
|
bot._record_send_failure()
|
|
assert bot.is_radio_offline is True
|
|
|
|
|
|
# Captured before any test patches threading.Thread.
|
|
_RealThread = threading.Thread
|
|
|
|
|
|
class TestRadioOfflineRecovery:
|
|
"""Radio-offline state must be leavable while every send is suppressed."""
|
|
|
|
def _offline_bot(self, tmp_path: Path) -> "MeshCoreBot":
|
|
config_file = tmp_path / "config.ini"
|
|
db_path = tmp_path / "bot.db"
|
|
_write_config(config_file, db_path)
|
|
bot = MeshCoreBot(config_file=str(config_file))
|
|
self.scheduler = MagicMock()
|
|
self.alerts: list = []
|
|
self._thread_patch = patch("threading.Thread")
|
|
thread_cls = self._thread_patch.start()
|
|
# Count alert-email threads; let every other thread (e.g. asyncio.to_thread workers) run.
|
|
thread_cls.side_effect = lambda *a, **k: (
|
|
self.alerts.append(k) or MagicMock()
|
|
if k.get("target") == self.scheduler.send_radio_offline_alert_email
|
|
else _RealThread(*a, **k)
|
|
)
|
|
for _ in range(3):
|
|
bot._record_send_failure(scheduler=self.scheduler)
|
|
assert bot.is_radio_offline is True
|
|
assert bot.db_manager.get_metadata('bot.radio_offline') == 'true'
|
|
assert len(self.alerts) == 1
|
|
return bot
|
|
|
|
def teardown_method(self):
|
|
patcher = getattr(self, "_thread_patch", None)
|
|
if patcher is not None:
|
|
patcher.stop()
|
|
self._thread_patch = None
|
|
|
|
def _probe_ok(self, bot):
|
|
from meshcore import EventType
|
|
|
|
bot.meshcore = MagicMock()
|
|
bot.meshcore.is_connected = True
|
|
bot.meshcore.commands.get_time = AsyncMock(return_value=MagicMock(type=EventType.CURRENT_TIME))
|
|
assert asyncio.run(bot._probe_radio_health()) is True
|
|
|
|
def _clear_by_trial(self, bot):
|
|
self._probe_ok(bot)
|
|
_, trial = bot._admit_measured_send()
|
|
bot._record_send_success(trial)
|
|
assert bot._radio_offline is False
|
|
|
|
def _scheduled_send(self, bot, send_result):
|
|
"""Run one scheduled message through the real scheduler on a live event loop."""
|
|
from modules.scheduler import MessageScheduler
|
|
|
|
seen = {}
|
|
|
|
async def fake_send(channel, message, **kwargs):
|
|
seen['offline_inside_send'] = bot.is_radio_offline
|
|
if isinstance(send_result, BaseException):
|
|
raise send_result
|
|
return send_result
|
|
|
|
bot.command_manager = MagicMock()
|
|
bot.command_manager.send_channel_message = fake_send
|
|
bot.config.set("Bot", "scheduled_message_max_stagger_seconds", "0")
|
|
loop = asyncio.new_event_loop()
|
|
runner = _RealThread(target=loop.run_forever, daemon=True)
|
|
runner.start()
|
|
bot.main_event_loop = loop
|
|
try:
|
|
with patch.object(MessageScheduler, "setup_scheduled_messages"):
|
|
scheduler = MessageScheduler(bot)
|
|
scheduler._effective_send_scope = lambda channel, scope: scope
|
|
scheduler._channel_body_budget = lambda scope: 1000
|
|
scheduler.send_scheduled_message("general", "hello")
|
|
finally:
|
|
loop.call_soon_threadsafe(loop.stop)
|
|
runner.join(timeout=5)
|
|
loop.close()
|
|
return seen
|
|
|
|
def test_probe_arms_one_trial_and_ordinary_sends_stay_suppressed(self, tmp_path):
|
|
bot = self._offline_bot(tmp_path)
|
|
self._probe_ok(bot)
|
|
assert bot.is_radio_offline is True # interactive sends still suppressed
|
|
assert bot._admit_measured_send() == (True, 1)
|
|
assert bot._admit_measured_send() == (False, False) # the trial is taken once
|
|
|
|
def test_concurrent_admission_lets_exactly_one_trial_through(self, tmp_path):
|
|
bot = self._offline_bot(tmp_path)
|
|
self._probe_ok(bot)
|
|
results = []
|
|
barrier = threading.Barrier(8)
|
|
|
|
def admit():
|
|
barrier.wait()
|
|
results.append(bot._admit_measured_send())
|
|
|
|
workers = []
|
|
for _ in range(8):
|
|
worker = _RealThread(target=admit)
|
|
workers.append(worker)
|
|
for worker in workers:
|
|
worker.start()
|
|
for worker in workers:
|
|
worker.join(timeout=5)
|
|
assert results.count((True, 1)) == 1
|
|
assert results.count((False, False)) == 7
|
|
|
|
def test_only_the_trial_send_sees_the_radio_online(self, tmp_path):
|
|
bot = self._offline_bot(tmp_path)
|
|
self._probe_ok(bot)
|
|
_, trial = bot._admit_measured_send()
|
|
|
|
async def inside():
|
|
return bot.is_radio_offline
|
|
|
|
assert asyncio.run(bot._as_offline_trial(inside(), trial)) is False
|
|
assert bot.is_radio_offline is True
|
|
|
|
def test_successful_trial_send_clears_the_outage(self, tmp_path):
|
|
bot = self._offline_bot(tmp_path)
|
|
self._probe_ok(bot)
|
|
seen = self._scheduled_send(bot, True)
|
|
assert seen['offline_inside_send'] is False
|
|
assert bot.is_radio_offline is False
|
|
assert bot._radio_offline is False
|
|
assert bot.db_manager.get_metadata('bot.radio_offline') == 'false'
|
|
|
|
def test_trial_send_reporting_failure_waits_for_the_next_probe(self, tmp_path):
|
|
# A False result can follow real radio attempts (exhausted retries), so it
|
|
# must not re-arm the trial without another answered probe.
|
|
bot = self._offline_bot(tmp_path)
|
|
self._probe_ok(bot)
|
|
self._scheduled_send(bot, False)
|
|
assert bot.is_radio_offline is True
|
|
assert bot._radio_offline_trial is None
|
|
assert bot.db_manager.get_metadata('bot.radio_offline') == 'true'
|
|
assert len(self.alerts) == 1
|
|
|
|
def test_cancelling_a_reply_does_not_strand_its_successful_trial(self, tmp_path):
|
|
import time as real_time
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
|
|
bot = self._offline_bot(tmp_path)
|
|
self._probe_ok(bot)
|
|
_, trial = bot._admit_measured_send()
|
|
|
|
async def send():
|
|
return True
|
|
|
|
async def main():
|
|
loop = asyncio.get_running_loop()
|
|
# One busy worker, so the settlement is queued behind it when the caller is cancelled.
|
|
loop.set_default_executor(ThreadPoolExecutor(max_workers=1, thread_name_prefix="busy"))
|
|
busy = loop.run_in_executor(None, real_time.sleep, 0.2)
|
|
task = asyncio.create_task(bot._send_as_offline_trial(trial, send))
|
|
await asyncio.sleep(0.05)
|
|
task.cancel()
|
|
with pytest.raises(asyncio.CancelledError):
|
|
await task
|
|
await busy
|
|
await asyncio.sleep(0.2)
|
|
|
|
asyncio.run(main())
|
|
assert bot.is_radio_offline is False
|
|
|
|
def test_timed_out_trial_waits_for_the_next_probe_without_a_second_alert(self, tmp_path):
|
|
bot = self._offline_bot(tmp_path)
|
|
for _ in range(3):
|
|
self._probe_ok(bot)
|
|
self._scheduled_send(bot, asyncio.TimeoutError())
|
|
assert bot.is_radio_offline is True
|
|
assert bot._radio_offline_trial is None
|
|
assert bot._admit_measured_send() == (False, False)
|
|
assert len(self.alerts) == 1 # only the one sent when the outage began
|
|
|
|
def test_suppressed_scheduled_send_never_reaches_the_radio(self, tmp_path):
|
|
bot = self._offline_bot(tmp_path)
|
|
seen = self._scheduled_send(bot, True)
|
|
assert seen == {}
|
|
assert bot.is_radio_offline is True
|
|
|
|
def test_failed_probe_leaves_offline(self, tmp_path):
|
|
from meshcore import EventType
|
|
|
|
bot = self._offline_bot(tmp_path)
|
|
bot.meshcore = MagicMock()
|
|
bot.meshcore.is_connected = True
|
|
bot.meshcore.commands.get_time = AsyncMock(return_value=MagicMock(type=EventType.ERROR))
|
|
asyncio.run(bot._probe_radio_health())
|
|
assert bot.is_radio_offline is True
|
|
assert bot._admit_measured_send() == (False, False)
|
|
|
|
def test_web_viewer_clear_is_picked_up(self, tmp_path):
|
|
bot = self._offline_bot(tmp_path)
|
|
bot._sync_radio_offline_from_metadata()
|
|
assert bot.is_radio_offline is True # metadata still says offline
|
|
bot.db_manager.set_metadata('bot.radio_offline', 'false')
|
|
bot._last_offline_metadata_check = 0.0
|
|
bot._sync_radio_offline_from_metadata()
|
|
assert bot.is_radio_offline is False
|
|
assert bot._send_consecutive_failures == 0
|
|
|
|
def test_unconfirmed_trip_is_retried_before_a_viewer_clear_is_trusted(self, tmp_path):
|
|
config_file = tmp_path / "config.ini"
|
|
db_path = tmp_path / "bot.db"
|
|
_write_config(config_file, db_path)
|
|
bot = MeshCoreBot(config_file=str(config_file))
|
|
real_get = bot.db_manager.get_metadata
|
|
bot.db_manager.get_metadata = MagicMock(side_effect=sqlite3.OperationalError("locked"))
|
|
with patch("threading.Thread"):
|
|
for _ in range(3):
|
|
bot._record_send_failure(scheduler=MagicMock())
|
|
bot.db_manager.get_metadata = real_get
|
|
assert bot.is_radio_offline is True
|
|
# A 'false' seen before the trip is confirmed is not trusted ...
|
|
bot.db_manager.set_metadata('bot.radio_offline', 'false')
|
|
bot._last_offline_metadata_check = 0.0
|
|
bot._sync_radio_offline_from_metadata()
|
|
assert bot.is_radio_offline is True
|
|
assert bot.db_manager.get_metadata('bot.radio_offline') == 'true' # retried and confirmed
|
|
# ... but once confirmed, the viewer's clear works.
|
|
bot.db_manager.set_metadata('bot.radio_offline', 'false')
|
|
bot._last_offline_metadata_check = 0.0
|
|
bot._sync_radio_offline_from_metadata()
|
|
assert bot.is_radio_offline is False
|
|
|
|
def test_a_delayed_clear_write_cannot_erase_a_newer_outage(self, tmp_path):
|
|
bot = self._offline_bot(tmp_path)
|
|
real_set = bot.db_manager.set_metadata
|
|
entered = threading.Event()
|
|
release = threading.Event()
|
|
|
|
def slow_set(key, value):
|
|
if key == 'bot.radio_offline' and value == 'false' and not entered.is_set():
|
|
entered.set()
|
|
release.wait(5)
|
|
return real_set(key, value)
|
|
|
|
bot.db_manager.set_metadata = slow_set
|
|
self._probe_ok(bot)
|
|
_, trial = bot._admit_measured_send()
|
|
clearer = _RealThread(target=lambda: bot._record_send_success(trial))
|
|
clearer.start()
|
|
assert entered.wait(5)
|
|
# The old outage is cleared in memory; a new one trips while that write is stalled.
|
|
tripper = _RealThread(
|
|
target=lambda: [bot._record_send_failure(scheduler=self.scheduler) for _ in range(3)]
|
|
)
|
|
tripper.start()
|
|
deadline = time.time() + 5
|
|
while not bot._radio_offline and time.time() < deadline:
|
|
time.sleep(0.01)
|
|
assert bot._radio_offline is True # the newer outage has tripped in memory
|
|
release.set()
|
|
clearer.join(timeout=5)
|
|
tripper.join(timeout=5)
|
|
assert bot.is_radio_offline is True
|
|
assert bot.db_manager.get_metadata('bot.radio_offline') == 'true'
|
|
bot._last_offline_metadata_check = 0.0
|
|
bot._sync_radio_offline_from_metadata()
|
|
assert bot.is_radio_offline is True
|
|
|
|
def test_trial_is_settled_when_submission_to_the_loop_fails(self, tmp_path):
|
|
from modules.scheduler import MessageScheduler
|
|
|
|
bot = self._offline_bot(tmp_path)
|
|
self._probe_ok(bot)
|
|
bot.main_event_loop = MagicMock()
|
|
bot.main_event_loop.is_running.return_value = True
|
|
|
|
def closed_loop(coro, loop):
|
|
coro.close()
|
|
raise RuntimeError("Event loop is closed")
|
|
|
|
with patch.object(MessageScheduler, "setup_scheduled_messages"):
|
|
scheduler = MessageScheduler(bot)
|
|
with patch("asyncio.run_coroutine_threadsafe", side_effect=closed_loop):
|
|
scheduler.send_scheduled_message("general", "hello")
|
|
scheduler.send_interval_advert()
|
|
# The scheduled message's loss of the loop is inconclusive and re-arms;
|
|
# the advert treats it as a failure, ending the trial until the next probe.
|
|
assert bot._radio_offline_trial is None
|
|
assert bot.is_radio_offline is True
|
|
|
|
def test_fallback_trial_send_is_settled_by_its_outcome(self, tmp_path):
|
|
from modules.scheduler import MessageScheduler
|
|
|
|
bot = self._offline_bot(tmp_path)
|
|
self._probe_ok(bot)
|
|
seen = {}
|
|
|
|
async def fake_send(channel, message, **kwargs):
|
|
seen['offline_inside_send'] = bot.is_radio_offline
|
|
return True
|
|
|
|
bot.command_manager = MagicMock()
|
|
bot.command_manager.send_channel_message = fake_send
|
|
bot.config.set("Bot", "scheduled_message_max_stagger_seconds", "0")
|
|
bot.main_event_loop = None
|
|
with patch.object(MessageScheduler, "setup_scheduled_messages"):
|
|
scheduler = MessageScheduler(bot)
|
|
scheduler._effective_send_scope = lambda channel, scope: scope
|
|
scheduler._channel_body_budget = lambda scope: 1000
|
|
scheduler.send_scheduled_message("general", "hello")
|
|
assert seen['offline_inside_send'] is False
|
|
assert bot.is_radio_offline is False
|
|
assert bot._radio_offline is False
|
|
|
|
def test_trial_is_settled_even_when_preparing_the_send_raises(self, tmp_path):
|
|
from modules.scheduler import MessageScheduler
|
|
|
|
bot = self._offline_bot(tmp_path)
|
|
self._probe_ok(bot)
|
|
_, trial = bot._admit_measured_send()
|
|
with patch.object(MessageScheduler, "setup_scheduled_messages"):
|
|
scheduler = MessageScheduler(bot)
|
|
|
|
def broken():
|
|
raise ValueError("cannot build the send")
|
|
|
|
with pytest.raises(ValueError):
|
|
scheduler._run_measured_send(
|
|
broken, trial, error_label="x", loop_gone_label=None,
|
|
reports_outcome=False, run_fallback=scheduler._run_in_temporary_loop,
|
|
)
|
|
assert bot._radio_offline_trial == 'armed'
|
|
|
|
def test_an_older_trial_cannot_act_during_a_newer_one(self, tmp_path):
|
|
bot = self._offline_bot(tmp_path)
|
|
self._probe_ok(bot)
|
|
_, first = bot._admit_measured_send()
|
|
bot._record_send_failure(scheduler=self.scheduler, trial=first) # timed out
|
|
self._probe_ok(bot)
|
|
_, second = bot._admit_measured_send()
|
|
assert second != first
|
|
|
|
async def inside():
|
|
return bot.is_radio_offline
|
|
|
|
# The timed-out first send, still running, is not let through ...
|
|
assert asyncio.run(bot._as_offline_trial(inside(), first)) is True
|
|
assert asyncio.run(bot._as_offline_trial(inside(), second)) is False
|
|
# ... and its late settlement does not touch the newer trial.
|
|
bot._record_send_inconclusive(first)
|
|
assert bot._radio_offline_trial == 'in_flight'
|
|
bot._record_send_failure(scheduler=self.scheduler, trial=first)
|
|
assert bot._radio_offline_trial == 'in_flight'
|
|
|
|
def test_a_late_success_from_an_old_trial_cannot_clear_a_newer_outage(self, tmp_path):
|
|
bot = self._offline_bot(tmp_path)
|
|
self._probe_ok(bot)
|
|
_, old_trial = bot._admit_measured_send()
|
|
# The old trial's send went out, but before it settles the viewer clears
|
|
# that outage and a new one trips and starts its own trial.
|
|
bot._clear_radio_offline_state(expected_generation=bot._radio_offline_generation)
|
|
for _ in range(3):
|
|
bot._record_send_failure(scheduler=self.scheduler)
|
|
self._probe_ok(bot)
|
|
_, new_trial = bot._admit_measured_send()
|
|
bot._record_send_success(old_trial)
|
|
assert bot.is_radio_offline is True
|
|
assert bot._radio_offline_trial == 'in_flight'
|
|
bot._record_send_success(new_trial)
|
|
assert bot.is_radio_offline is False
|
|
|
|
def test_a_timed_out_trial_send_is_cancelled(self, tmp_path):
|
|
from concurrent.futures import TimeoutError as FuturesTimeoutError
|
|
|
|
from modules.scheduler import MessageScheduler
|
|
|
|
bot = self._offline_bot(tmp_path)
|
|
self._probe_ok(bot)
|
|
bot.main_event_loop = MagicMock()
|
|
bot.main_event_loop.is_running.return_value = True
|
|
future = MagicMock()
|
|
future.result.side_effect = FuturesTimeoutError()
|
|
|
|
def submit(coro, loop):
|
|
asyncio.run(coro)
|
|
return future
|
|
|
|
with patch.object(MessageScheduler, "setup_scheduled_messages"):
|
|
scheduler = MessageScheduler(bot)
|
|
scheduler._send_scheduled_message_async = lambda *args, **kwargs: asyncio.sleep(0)
|
|
with patch("asyncio.run_coroutine_threadsafe", side_effect=submit):
|
|
scheduler.send_scheduled_message("general", "hello")
|
|
future.cancel.assert_called_once()
|
|
assert bot._radio_offline_trial is None
|
|
|
|
def test_trial_is_settled_when_the_wrapper_fails_before_sending(self, tmp_path):
|
|
from modules.scheduler import MessageScheduler
|
|
|
|
bot = self._offline_bot(tmp_path)
|
|
with patch.object(MessageScheduler, "setup_scheduled_messages"):
|
|
scheduler = MessageScheduler(bot)
|
|
scheduler.get_current_time = MagicMock(side_effect=RuntimeError("clock"))
|
|
for send in (lambda: scheduler.send_scheduled_message("general", "hello"), scheduler.send_interval_advert):
|
|
self._probe_ok(bot)
|
|
assert bot._radio_offline_trial == 'armed'
|
|
with pytest.raises(RuntimeError):
|
|
send()
|
|
assert bot._radio_offline_trial == 'armed' # not stranded in flight
|
|
|
|
def test_a_reply_can_take_the_trial_and_clear_the_outage(self, tmp_path):
|
|
bot = self._offline_bot(tmp_path)
|
|
self._probe_ok(bot)
|
|
_, trial = bot._admit_measured_send()
|
|
seen = {}
|
|
|
|
async def send():
|
|
seen['offline_inside_send'] = bot.is_radio_offline
|
|
return True
|
|
|
|
assert asyncio.run(bot._send_as_offline_trial(trial, send)) is True
|
|
assert seen['offline_inside_send'] is False
|
|
assert bot.is_radio_offline is False
|
|
assert bot.db_manager.get_metadata('bot.radio_offline') == 'false'
|
|
|
|
def test_a_failed_reply_trial_waits_for_the_next_probe(self, tmp_path):
|
|
bot = self._offline_bot(tmp_path)
|
|
self._probe_ok(bot)
|
|
_, trial = bot._admit_measured_send()
|
|
|
|
async def send():
|
|
return False
|
|
|
|
assert asyncio.run(bot._send_as_offline_trial(trial, send)) is False
|
|
assert bot.is_radio_offline is True
|
|
assert bot._radio_offline_trial is None
|
|
assert len(self.alerts) == 1
|
|
|
|
def test_command_manager_send_claims_the_trial_and_settles_it(self, tmp_path):
|
|
bot = self._offline_bot(tmp_path)
|
|
bot.connected = True
|
|
bot.meshcore = MagicMock()
|
|
refused = AsyncMock(return_value=(False, None)) # e.g. rate limited: the send returns False
|
|
with patch.object(type(bot.command_manager), "_check_rate_limits", refused):
|
|
# No armed trial: suppressed before reaching the rate limiter.
|
|
assert asyncio.run(bot.command_manager.send_channel_message("general", "hi")) is False
|
|
assert refused.await_count == 0
|
|
self._probe_ok(bot)
|
|
# Armed: this send takes the trial, passes the guard under it, and fails it.
|
|
assert asyncio.run(bot.command_manager.send_channel_message("general", "hi")) is False
|
|
assert refused.await_count == 1
|
|
assert bot._radio_offline_trial is None
|
|
# The next send is suppressed again until another probe.
|
|
assert asyncio.run(bot.command_manager.send_dm("someone", "hi")) is False
|
|
assert refused.await_count == 1
|
|
assert bot.is_radio_offline is True
|
|
|
|
def test_viewer_clear_of_an_old_outage_leaves_a_newer_one(self, tmp_path):
|
|
bot = self._offline_bot(tmp_path)
|
|
old_generation = bot._radio_offline_generation
|
|
self._clear_by_trial(bot)
|
|
for _ in range(3):
|
|
bot._record_send_failure(scheduler=self.scheduler)
|
|
assert bot._clear_radio_offline_state(expected_generation=old_generation) is False
|
|
assert bot.is_radio_offline is True
|
|
assert bot._clear_radio_offline_state(expected_generation=bot._radio_offline_generation) is True
|
|
assert bot.is_radio_offline is False
|
|
|
|
def test_viewer_check_is_throttled(self, tmp_path):
|
|
bot = self._offline_bot(tmp_path)
|
|
bot.db_manager.get_metadata = MagicMock(return_value='true')
|
|
bot._last_offline_metadata_check = 0.0
|
|
bot._sync_radio_offline_from_metadata()
|
|
bot._sync_radio_offline_from_metadata()
|
|
assert bot.db_manager.get_metadata.call_count == 1
|
|
|
|
def test_new_outage_after_a_clear_alerts_again(self, tmp_path):
|
|
bot = self._offline_bot(tmp_path)
|
|
self._clear_by_trial(bot)
|
|
for _ in range(3):
|
|
bot._record_send_failure(scheduler=self.scheduler)
|
|
assert len(self.alerts) == 2
|
|
|
|
|
|
class TestSendStartupAdvertTimeout:
|
|
"""Coverage for startup advert timeout handling."""
|
|
|
|
def test_timeout_records_send_failure(self, tmp_path):
|
|
config_file = tmp_path / "config.ini"
|
|
db_path = tmp_path / "bot.db"
|
|
_write_config(config_file, db_path)
|
|
bot = MeshCoreBot(config_file=str(config_file))
|
|
bot.config.set("Bot", "startup_advert", "flood")
|
|
bot.meshcore = MagicMock()
|
|
bot.meshcore.commands.send_advert = MagicMock()
|
|
bot._record_send_failure = MagicMock()
|
|
|
|
async def run():
|
|
async def fake_wait_for(coro, timeout):
|
|
coro.close()
|
|
raise asyncio.TimeoutError()
|
|
|
|
with patch("asyncio.wait_for", side_effect=fake_wait_for):
|
|
await bot.send_startup_advert()
|
|
|
|
asyncio.run(run())
|
|
|
|
bot._record_send_failure.assert_called_once()
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _BotAdminServer — admin HTTP API
|
|
# ---------------------------------------------------------------------------
|
|
class TestBotAdminServer:
|
|
"""Admin HTTP server: /api/admin/reload and /api/admin/health."""
|
|
|
|
@staticmethod
|
|
def _free_port() -> int:
|
|
"""Reserve a port from the ephemeral range and hand the number back.
|
|
|
|
Hardcoded ports make the suite unrunnable twice at once. The second
|
|
process fails to bind, but its request still reaches the first process's
|
|
server on that port and comes back with whatever that server happens to
|
|
be patched to return, which surfaces as an unrelated assertion failure.
|
|
"""
|
|
with socket.socket() as sock:
|
|
sock.bind(("127.0.0.1", 0))
|
|
return sock.getsockname()[1]
|
|
|
|
@staticmethod
|
|
def _wait_until_serving(port: int, timeout: float = 5.0) -> None:
|
|
"""Block until the admin server accepts connections, or fail loudly.
|
|
|
|
The server comes up in a thread; a fixed sleep is either too short on a
|
|
loaded machine or wasted time on an idle one. Polling also turns a server
|
|
that never binds into a clear message instead of a confusing failure at
|
|
the assertion further down.
|
|
"""
|
|
deadline = time.monotonic() + timeout
|
|
while time.monotonic() < deadline:
|
|
try:
|
|
with socket.create_connection(("127.0.0.1", port), timeout=0.2):
|
|
return
|
|
except OSError:
|
|
time.sleep(0.01)
|
|
raise AssertionError(f"admin server never started listening on port {port}")
|
|
|
|
def _make_bot_with_admin(self, tmp_path):
|
|
"""Write config with [Admin] enabled and return a bot + token."""
|
|
token = "test-secret-token"
|
|
port = self._free_port()
|
|
config_file = tmp_path / "config.ini"
|
|
db_path = tmp_path / "bot.db"
|
|
config_file.write_text(
|
|
f"""[Connection]
|
|
connection_type = serial
|
|
serial_port = /dev/ttyUSB0
|
|
timeout = 30
|
|
|
|
[Bot]
|
|
db_path = {db_path.as_posix()}
|
|
prefix_bytes = 1
|
|
|
|
[Channels]
|
|
monitor_channels = #general
|
|
|
|
[Admin]
|
|
enabled = true
|
|
port = {port}
|
|
token = {token}
|
|
""",
|
|
encoding="utf-8",
|
|
)
|
|
bot = MeshCoreBot(config_file=str(config_file))
|
|
return bot, token, port
|
|
|
|
def test_admin_server_created_when_enabled(self, tmp_path):
|
|
bot, _token, _port = self._make_bot_with_admin(tmp_path)
|
|
assert bot._admin_server is not None
|
|
|
|
def test_admin_server_none_when_disabled(self, tmp_path):
|
|
config_file = tmp_path / "config.ini"
|
|
db_path = tmp_path / "bot.db"
|
|
_write_config(config_file, db_path)
|
|
bot = MeshCoreBot(config_file=str(config_file))
|
|
assert bot._admin_server is None
|
|
|
|
def test_admin_server_none_when_token_missing(self, tmp_path):
|
|
config_file = tmp_path / "config.ini"
|
|
db_path = tmp_path / "bot.db"
|
|
config_file.write_text(
|
|
f"""[Connection]
|
|
connection_type = serial
|
|
serial_port = /dev/ttyUSB0
|
|
timeout = 30
|
|
|
|
[Bot]
|
|
db_path = {db_path.as_posix()}
|
|
prefix_bytes = 1
|
|
|
|
[Channels]
|
|
monitor_channels = #general
|
|
|
|
[Admin]
|
|
enabled = true
|
|
port = 15002
|
|
token =
|
|
""",
|
|
encoding="utf-8",
|
|
)
|
|
bot = MeshCoreBot(config_file=str(config_file))
|
|
assert bot._admin_server is None
|
|
|
|
def test_reload_endpoint_success(self, tmp_path):
|
|
"""POST /api/admin/reload returns 200 and success=true when reload succeeds."""
|
|
import urllib.request
|
|
|
|
bot, token, port = self._make_bot_with_admin(tmp_path)
|
|
|
|
with patch.object(bot, "reload_config", return_value=(True, "Configuration reloaded successfully")):
|
|
server = bot._admin_server
|
|
server.start()
|
|
self._wait_until_serving(port)
|
|
|
|
req = urllib.request.Request(
|
|
f"http://127.0.0.1:{port}/api/admin/reload",
|
|
method="POST",
|
|
headers={"Authorization": f"Bearer {token}"},
|
|
)
|
|
with urllib.request.urlopen(req, timeout=5) as resp:
|
|
import json
|
|
body = json.loads(resp.read())
|
|
assert body["success"] is True
|
|
assert "reloaded" in body["message"].lower()
|
|
|
|
def test_reload_endpoint_failure(self, tmp_path):
|
|
"""POST /api/admin/reload returns 409 when reload is rejected."""
|
|
import urllib.request
|
|
from urllib.error import HTTPError
|
|
|
|
bot, token, port = self._make_bot_with_admin(tmp_path)
|
|
|
|
with patch.object(bot, "reload_config", return_value=(False, "Radio settings changed")):
|
|
server = bot._admin_server
|
|
server.start()
|
|
self._wait_until_serving(port)
|
|
|
|
req = urllib.request.Request(
|
|
f"http://127.0.0.1:{port}/api/admin/reload",
|
|
method="POST",
|
|
headers={"Authorization": f"Bearer {token}"},
|
|
)
|
|
with pytest.raises(HTTPError) as exc_info:
|
|
urllib.request.urlopen(req, timeout=5)
|
|
assert exc_info.value.code == 409
|
|
|
|
def test_reload_endpoint_rejects_bad_token(self, tmp_path):
|
|
"""POST /api/admin/reload returns 401 with wrong token."""
|
|
import urllib.request
|
|
from urllib.error import HTTPError
|
|
|
|
bot, _token, port = self._make_bot_with_admin(tmp_path)
|
|
server = bot._admin_server
|
|
server.start()
|
|
self._wait_until_serving(port)
|
|
|
|
req = urllib.request.Request(
|
|
f"http://127.0.0.1:{port}/api/admin/reload",
|
|
method="POST",
|
|
headers={"Authorization": "Bearer wrong-token"},
|
|
)
|
|
with pytest.raises(HTTPError) as exc_info:
|
|
urllib.request.urlopen(req, timeout=5)
|
|
assert exc_info.value.code == 401
|
|
|
|
def test_health_endpoint_returns_ok(self, tmp_path):
|
|
"""GET /api/admin/health returns 200 and status=ok."""
|
|
import urllib.request
|
|
|
|
bot, token, port = self._make_bot_with_admin(tmp_path)
|
|
server = bot._admin_server
|
|
server.start()
|
|
self._wait_until_serving(port)
|
|
|
|
req = urllib.request.Request(
|
|
f"http://127.0.0.1:{port}/api/admin/health",
|
|
method="GET",
|
|
headers={"Authorization": f"Bearer {token}"},
|
|
)
|
|
with urllib.request.urlopen(req, timeout=5) as resp:
|
|
import json
|
|
body = json.loads(resp.read())
|
|
assert body["status"] == "ok"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helper: create a coroutine that returns a fixed value
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
async def _make_coro_async(value):
|
|
return value
|
|
|
|
|
|
def _make_coro(value):
|
|
"""Return a coroutine that immediately resolves to *value*."""
|
|
return _make_coro_async(value)
|
|
|
|
|
|
class TestHealthStateAvailableToPluginConstructors:
|
|
"""Plugins built in MeshCoreBot.__init__ may read radio-health properties."""
|
|
|
|
def test_command_constructor_can_read_health_properties(self, tmp_path):
|
|
from modules.command_manager import CommandManager
|
|
|
|
seen = {}
|
|
real_init = CommandManager.__init__
|
|
|
|
def spying_init(self, bot):
|
|
seen["offline"] = bot.is_radio_offline
|
|
seen["zombie"] = bot.is_radio_zombie
|
|
real_init(self, bot)
|
|
|
|
config_file = tmp_path / "config.ini"
|
|
_write_config(config_file, tmp_path / "bot.db")
|
|
with patch.object(CommandManager, "__init__", spying_init):
|
|
MeshCoreBot(config_file=str(config_file))
|
|
assert seen == {"offline": False, "zombie": False}
|