dandri/meshcore-bot
1
0 Fork 0
mirror of https://github.com/agessaman/meshcore-bot.git synced 2026-11-11 15:07:32 +00:00
Code Issues Packages Projects Releases Wiki Activity
Files
3ecdf1fb8dd67c504d1d3adbd16962d776a40dbd
meshcore-bot/tests/test_web_viewer_integration.py
T
agessaman 2a1c2f86d8 feat(web_viewer): rebuild Real-time as one tape of what the radio heard 
Real-time was three equal panels (commands, packets, messages) that each scrolled on their own, couldn't be held still, and showed the same transmission up to three times. It's now one tape with one row per transmission: a packet, the channel message decrypted from it and a command it triggered share a row, joined by packet hash, and the bot's reply hangs under the command with how many times a repeater was heard echoing it. Those echoes fold into the reply instead of getting rows of their own, and a packet heard several times is one row with a count and every reception (SNR, RSSI, hops, path) when opened.

A receiver strip at the top says whether the bot is listening or the radio is offline or not responding (from the banner poll base.html already runs, now published as a banner-status event), when it last heard a packet, the packet rate and a 10-minute activity bar. The tape holds still while you read it: Pause (P), scrolling down, a moving mouse pointer or keyboard focus queue new rows behind an "N new" button. Rows can be kept (safe from the 300-row limit and from Clear), filtered to messages or commands, by packet type or channel, and searched (/). Replayed history sits under a "Heard before you opened this page" divider, and adverts and keyed group text are decoded in the browser and labeled as such. Each row's glyph chip carries its packet type's icon in the dashboard chart's color for that type; decoded messages get a stronger tint, and the bot's commands and replies are violet.

Server side, command rows carry trigger_packet_hash (from the message's routing info) and, once the transmission tracker matches a repeat, reply_packet_hash. Replayed rows carry stream_ts and replayed, and live rows stream_ts. Commands stored before this pair with their message only on an exact sender, channel and text match within 20 seconds.

Real-time is admin-only now: its socket already was, so anonymous visitors got three "Admin login required" badges over spinners that never filled. They're redirected to login and the nav link is hidden from them.

The byte breakdown sliced multibyte paths by hop count, so 2- and 3-byte hops spilled into the payload; it uses the path's byte count and shows the raw path-length byte. Route types 0 and 2 were swapped, and the channel key store was often built before the decoder had loaded, so it never decrypted anything. js-sha256 is vendored under static/vendor instead of loading from cdnjs.
2026-10-17 19:40:23 -07:00

740 lines
30 KiB
Python
Raw Blame History

"""Tests for modules.web_viewer.integration — BotIntegration pure logic."""
import json
import queue
import time
from configparser import ConfigParser
from pathlib import Path
from unittest.mock import MagicMock, Mock, patch
import pytest
from modules.models import MeshMessage
# ---------------------------------------------------------------------------
# Factory helpers
# ---------------------------------------------------------------------------
def _make_bot():
"""Create minimal mock bot for BotIntegration tests."""
bot = MagicMock()
bot.logger = Mock()
config = ConfigParser()
config.add_section("Bot")
config.add_section("Web_Viewer")
config.set("Web_Viewer", "host", "127.0.0.1")
config.set("Web_Viewer", "port", "8080")
config.set("Web_Viewer", "enabled", "false")
config.set("Web_Viewer", "auto_start", "false")
config.set("Web_Viewer", "debug", "false")
bot.config = config
bot.bot_root = "/tmp"
bot.db_manager = MagicMock()
bot.db_manager.db_path = ":memory:"
bot.transmission_tracker = None
return bot
def _make_bot_integration(bot=None):
"""Create BotIntegration with all I/O patched out."""
if bot is None:
bot = _make_bot()
from modules.web_viewer.integration import BotIntegration
with patch.object(BotIntegration, "_init_http_session"), \
patch.object(BotIntegration, "_init_packet_stream_table"), \
patch.object(BotIntegration, "_start_drain_thread"):
obj = BotIntegration(bot)
# Give it a real write queue for testing
obj._write_queue = queue.Queue()
obj.http_session = None
return obj
# ---------------------------------------------------------------------------
# reset_circuit_breaker
# ---------------------------------------------------------------------------
class TestResetCircuitBreaker:
def test_clears_open_flag(self):
bi = _make_bot_integration()
bi.circuit_breaker_open = True
bi.circuit_breaker_failures = 5
bi.reset_circuit_breaker()
assert bi.circuit_breaker_open is False
assert bi.circuit_breaker_failures == 0
# ---------------------------------------------------------------------------
# _should_skip_web_viewer_send
# ---------------------------------------------------------------------------
class TestShouldSkipWebViewerSend:
def test_not_open_returns_false(self):
bi = _make_bot_integration()
bi.circuit_breaker_open = False
assert bi._should_skip_web_viewer_send() is False
def test_open_within_cooldown_returns_true(self):
bi = _make_bot_integration()
bi.circuit_breaker_open = True
bi.circuit_breaker_last_failure_time = time.time() # just now
assert bi._should_skip_web_viewer_send() is True
def test_open_beyond_cooldown_resets_and_returns_false(self):
bi = _make_bot_integration()
bi.circuit_breaker_open = True
bi.circuit_breaker_failures = 3
# Set last failure time far in the past
bi.circuit_breaker_last_failure_time = time.time() - bi.CIRCUIT_BREAKER_COOLDOWN_SEC - 1
assert bi._should_skip_web_viewer_send() is False
assert bi.circuit_breaker_open is False
# ---------------------------------------------------------------------------
# _record_web_viewer_result
# ---------------------------------------------------------------------------
class TestRecordWebViewerResult:
def test_success_resets_circuit(self):
bi = _make_bot_integration()
bi.circuit_breaker_failures = 2
bi.circuit_breaker_open = True
bi._record_web_viewer_result(True)
assert bi.circuit_breaker_failures == 0
assert bi.circuit_breaker_open is False
def test_failure_increments_counter(self):
bi = _make_bot_integration()
bi._record_web_viewer_result(False)
assert bi.circuit_breaker_failures == 1
def test_failure_opens_circuit_at_threshold(self):
bi = _make_bot_integration()
for _ in range(bi.CIRCUIT_BREAKER_THRESHOLD):
bi._record_web_viewer_result(False)
assert bi.circuit_breaker_open is True
def test_failure_below_threshold_does_not_open(self):
bi = _make_bot_integration()
for _ in range(bi.CIRCUIT_BREAKER_THRESHOLD - 1):
bi._record_web_viewer_result(False)
assert bi.circuit_breaker_open is False
# ---------------------------------------------------------------------------
# _make_json_serializable
# ---------------------------------------------------------------------------
class TestMakeJsonSerializable:
def setup_method(self):
self.bi = _make_bot_integration()
def test_none_passes_through(self):
assert self.bi._make_json_serializable(None) is None
def test_string_passes_through(self):
assert self.bi._make_json_serializable("hello") == "hello"
def test_int_passes_through(self):
assert self.bi._make_json_serializable(42) == 42
def test_float_passes_through(self):
assert self.bi._make_json_serializable(3.14) == 3.14
def test_bool_passes_through(self):
assert self.bi._make_json_serializable(True) is True
def test_list_recurses(self):
result = self.bi._make_json_serializable([1, "two", None])
assert result == [1, "two", None]
def test_tuple_becomes_list(self):
result = self.bi._make_json_serializable((1, 2))
assert result == [1, 2]
def test_dict_recurses(self):
result = self.bi._make_json_serializable({"a": 1, "b": [2, 3]})
assert result == {"a": 1, "b": [2, 3]}
def test_enum_like_uses_name(self):
obj = Mock(spec=["name"])
obj.name = "MY_ENUM"
result = self.bi._make_json_serializable(obj)
assert result == "MY_ENUM"
def test_value_attr_used_when_no_name(self):
obj = Mock(spec=["value"])
obj.value = 99
result = self.bi._make_json_serializable(obj)
assert result == 99
def test_object_with_dict_converted(self):
class Dummy:
def __init__(self):
self.x = 1
self.y = "a"
result = self.bi._make_json_serializable(Dummy())
assert result == {"x": 1, "y": "a"}
def test_unknown_object_stringified(self):
# An object with no __dict__, no name, no value
result = self.bi._make_json_serializable(object())
assert isinstance(result, str)
def test_max_depth_stringifies(self):
# At max_depth, return str
result = self.bi._make_json_serializable({"nested": [1]}, depth=4, max_depth=3)
assert isinstance(result, str)
def test_deeply_nested_dict(self):
d = {"a": {"b": {"c": "deep"}}}
result = self.bi._make_json_serializable(d)
assert result["a"]["b"]["c"] == "deep"
# ---------------------------------------------------------------------------
# _insert_packet_stream_row
# ---------------------------------------------------------------------------
class TestInsertPacketStreamRow:
def test_enqueues_tuple(self):
bi = _make_bot_integration()
bi._insert_packet_stream_row('{"x": 1}', "packet")
assert not bi._write_queue.empty()
ts, data, row_type, *_dims = bi._write_queue.get_nowait()
assert data == '{"x": 1}'
assert row_type == "packet"
def test_queue_exception_logged(self):
bi = _make_bot_integration()
bi._write_queue = Mock()
bi._write_queue.put.side_effect = Exception("full")
# Should not raise
bi._insert_packet_stream_row("{}", "packet")
bi.bot.logger.warning.assert_called_once()
# ---------------------------------------------------------------------------
# capture_full_packet_data
# ---------------------------------------------------------------------------
class TestCaptureFullPacketData:
def test_dict_packet_data_queued(self):
bi = _make_bot_integration()
bi.capture_full_packet_data({"snr": -5.0, "path_len": 2})
assert not bi._write_queue.empty()
ts, data, row_type, *_dims = bi._write_queue.get_nowait()
parsed = json.loads(data)
assert row_type == "packet"
assert parsed["hops"] == 2
def test_no_path_len_defaults_hops_to_0(self):
bi = _make_bot_integration()
bi.capture_full_packet_data({"snr": -5.0})
ts, data, row_type, *_dims = bi._write_queue.get_nowait()
parsed = json.loads(data)
assert parsed["hops"] == 0
def test_existing_hops_not_overwritten(self):
bi = _make_bot_integration()
bi.capture_full_packet_data({"hops": 3, "path_len": 2})
ts, data, row_type, *_dims = bi._write_queue.get_nowait()
parsed = json.loads(data)
assert parsed["hops"] == 3
def test_datetime_added(self):
bi = _make_bot_integration()
bi.capture_full_packet_data({"snr": 0})
ts, data, _, *_dims = bi._write_queue.get_nowait()
parsed = json.loads(data)
assert "datetime" in parsed
def test_non_dict_wrapped(self):
bi = _make_bot_integration()
# Pass a non-dict (e.g. a Mock with __dict__)
bi.capture_full_packet_data("not a dict")
# Should not raise; may enqueue something
assert True # reached here without exception
# ---------------------------------------------------------------------------
# capture_command
# ---------------------------------------------------------------------------
class TestCaptureCommand:
def test_basic_capture_queued(self):
bi = _make_bot_integration()
msg = Mock()
msg.sender_id = "aa:bb"
msg.channel = "general"
msg.content = "ping"
bi.capture_command(msg, "ping", "Pong!", True)
assert not bi._write_queue.empty()
ts, data, row_type, *_dims = bi._write_queue.get_nowait()
assert row_type == "command"
parsed = json.loads(data)
assert parsed["command"] == "ping"
assert parsed["success"] is True
def test_user_input_prefers_original_content(self):
"""#267: command stream must show the on-air body, not mention-stripped content."""
bi = _make_bot_integration()
msg = MeshMessage(
content="hi",
original_content="@[AlphaBot👾] hi",
sender_id="Howl 🦾",
channel="#alphabot",
)
bi.capture_command(msg, "hello", "Dobryy vecher, earthling.", True)
_, data, row_type, *_dims = bi._write_queue.get_nowait()
assert row_type == "command"
parsed = json.loads(data)
assert parsed["user_input"] == "@[AlphaBot👾] hi"
assert parsed["command"] == "hello"
def test_carries_trigger_and_reply_packet_hashes(self):
"""The realtime tape joins a command to the packet that triggered it and to its heard-back reply."""
bi = _make_bot_integration()
bi.bot.transmission_tracker = Mock()
bi.bot.transmission_tracker.get_repeat_info.return_value = {
"repeat_count": 2, "repeater_prefixes": ["7e"], "repeater_counts": {"7e": 2},
"packet_hash": "FEEDFACE00112233",
}
msg = MeshMessage(
content="wx", sender_id="Kim", channel="#bot",
routing_info={"packet_hash": "ABCDEF0123456789"},
)
bi.capture_command(msg, "wx", "54F", True, command_id="cmd-1")
parsed = json.loads(bi._write_queue.get_nowait()[1])
assert parsed["trigger_packet_hash"] == "ABCDEF0123456789"
assert parsed["reply_packet_hash"] == "FEEDFACE00112233"
def test_omits_packet_hashes_it_does_not_have(self):
bi = _make_bot_integration()
bi.bot.transmission_tracker = None
msg = MeshMessage(content="wx", sender_id="Kim", channel="#bot",
routing_info={"packet_hash": "0000000000000000"})
bi.capture_command(msg, "wx", "54F", True)
parsed = json.loads(bi._write_queue.get_nowait()[1])
assert "trigger_packet_hash" not in parsed
assert "reply_packet_hash" not in parsed
def test_no_transmission_tracker(self):
bi = _make_bot_integration()
bi.bot.transmission_tracker = None
msg = Mock()
msg.sender_id = "u1"
msg.channel = "ch"
msg.content = "cmd"
bi.capture_command(msg, "cmd", "resp", True)
ts, data, _, *_dims = bi._write_queue.get_nowait()
parsed = json.loads(data)
assert parsed["repeat_count"] == 0
# ---------------------------------------------------------------------------
# capture_channel_message
# ---------------------------------------------------------------------------
class TestCaptureChannelMessage:
def test_message_queued_with_type_message(self):
bi = _make_bot_integration()
msg = Mock()
msg.sender_id = "aa"
msg.channel = "general"
msg.content = "hello"
msg.snr = -5.0
msg.hops = 1
msg.path = "bb,cc"
msg.is_dm = False
bi.capture_channel_message(msg)
assert not bi._write_queue.empty()
ts, data, row_type, *_dims = bi._write_queue.get_nowait()
assert row_type == "message"
parsed = json.loads(data)
assert parsed["type"] == "message"
assert parsed["content"] == "hello"
def test_prefers_original_content_over_stripped_body(self):
"""#267: capture must store the on-air body, not mention-stripped content."""
bi = _make_bot_integration()
msg = MeshMessage(
content="ack | 9d12,aa11,4039 (3 hops)",
original_content="ack @[IU1IPB-1] | 9d12,aa11,4039 (3 hops)",
sender_id="IW1PTV-BOT-Savona",
channel="#bot",
hops=4,
is_dm=False,
)
bi.capture_channel_message(msg)
_, data, row_type, *_dims = bi._write_queue.get_nowait()
assert row_type == "message"
parsed = json.loads(data)
assert parsed["content"] == "ack @[IU1IPB-1] | 9d12,aa11,4039 (3 hops)"
def test_dm_message_captured(self):
bi = _make_bot_integration()
msg = Mock()
msg.sender_id = "aa"
msg.channel = ""
msg.content = "private"
msg.snr = -3.0
msg.hops = 0
msg.path = ""
msg.is_dm = True
msg.routing_info = None
bi.capture_channel_message(msg)
ts, data, _, *_dims = bi._write_queue.get_nowait()
parsed = json.loads(data)
assert parsed["is_dm"] is True
assert "packet_hash" not in parsed
def test_stores_packet_hash_from_routing_info(self):
"""#270: message rows carry packet_hash so routes can link back to the packet."""
bi = _make_bot_integration()
msg = MeshMessage(
content="hello",
sender_id="aa",
channel="general",
hops=1,
is_dm=False,
routing_info={"packet_hash": "ABCDEF0123456789", "path_length": 1},
)
bi.capture_channel_message(msg)
_, data, row_type, *_dims = bi._write_queue.get_nowait()
assert row_type == "message"
parsed = json.loads(data)
assert parsed["packet_hash"] == "ABCDEF0123456789"
def test_omits_missing_or_all_zero_packet_hash(self):
"""#270: do not invent a hash when routing_info lacks a real one."""
bi = _make_bot_integration()
for routing_info in (None, {}, {"packet_hash": ""}, {"packet_hash": "0000000000000000"}):
msg = MeshMessage(
content="hello",
sender_id="aa",
channel="general",
hops=0,
is_dm=False,
routing_info=routing_info,
)
bi.capture_channel_message(msg)
_, data, _, *_dims = bi._write_queue.get_nowait()
parsed = json.loads(data)
assert "packet_hash" not in parsed, routing_info
# ---------------------------------------------------------------------------
# capture_packet_routing
# ---------------------------------------------------------------------------
class TestCapturePacketRouting:
def test_routing_data_queued(self):
bi = _make_bot_integration()
bi.capture_packet_routing({"path_nodes": ["aa", "bb"]})
assert not bi._write_queue.empty()
ts, data, row_type, *_dims = bi._write_queue.get_nowait()
assert row_type == "routing"
# ---------------------------------------------------------------------------
# _get_web_viewer_db_path
# ---------------------------------------------------------------------------
class TestGetWebViewerDbPath:
def test_uses_bot_db_path_when_no_section(self):
bi = _make_bot_integration()
bi.bot.config.remove_section("Web_Viewer")
bi.bot.db_manager.db_path = "/tmp/test.db"
result = bi._get_web_viewer_db_path()
assert "test.db" in result
def test_uses_web_viewer_db_path_when_set(self):
bi = _make_bot_integration()
bi.bot.config.set("Web_Viewer", "db_path", "/tmp/viewer.db")
result = bi._get_web_viewer_db_path()
assert "viewer.db" in result
def test_falls_back_when_web_viewer_db_path_empty(self):
bi = _make_bot_integration()
bi.bot.config.set("Web_Viewer", "db_path", "")
bi.bot.db_manager.db_path = "/tmp/bot.db"
result = bi._get_web_viewer_db_path()
assert "bot.db" in result
# ---------------------------------------------------------------------------
# WebViewerIntegration validation
# ---------------------------------------------------------------------------
class TestWebViewerIntegrationValidation:
def test_invalid_host_raises(self):
from modules.web_viewer.integration import WebViewerIntegration
bot = _make_bot()
bot.config.set("Web_Viewer", "host", "evil.host")
bot.config.set("Web_Viewer", "enabled", "false")
with patch.object(WebViewerIntegration, "start_viewer"):
with patch("modules.web_viewer.integration.BotIntegration._init_http_session"), \
patch("modules.web_viewer.integration.BotIntegration._init_packet_stream_table"), \
patch("modules.web_viewer.integration.BotIntegration._start_drain_thread"):
with pytest.raises(ValueError, match="Invalid host"):
WebViewerIntegration(bot)
def test_invalid_port_raises(self):
from modules.web_viewer.integration import WebViewerIntegration
bot = _make_bot()
bot.config.set("Web_Viewer", "port", "80") # privileged port
with patch("modules.web_viewer.integration.BotIntegration._init_http_session"), \
patch("modules.web_viewer.integration.BotIntegration._init_packet_stream_table"), \
patch("modules.web_viewer.integration.BotIntegration._start_drain_thread"):
with pytest.raises(ValueError, match="Port must be"):
WebViewerIntegration(bot)
def test_valid_config_no_error(self):
from modules.web_viewer.integration import WebViewerIntegration
bot = _make_bot()
with patch("modules.web_viewer.integration.BotIntegration._init_http_session"), \
patch("modules.web_viewer.integration.BotIntegration._init_packet_stream_table"), \
patch("modules.web_viewer.integration.BotIntegration._start_drain_thread"):
wvi = WebViewerIntegration(bot)
assert wvi.host == "127.0.0.1"
assert wvi.port == 8080
def test_zero_host_without_password_logs_error_does_not_raise(self):
from modules.web_viewer.integration import WebViewerIntegration
bot = _make_bot()
bot.config.set("Web_Viewer", "enabled", "true")
bot.config.set("Web_Viewer", "host", "0.0.0.0")
bot.config.set("Web_Viewer", "web_viewer_password", "")
with patch.object(WebViewerIntegration, "start_viewer"):
with patch("modules.web_viewer.integration.BotIntegration._init_http_session"), \
patch("modules.web_viewer.integration.BotIntegration._init_packet_stream_table"), \
patch("modules.web_viewer.integration.BotIntegration._start_drain_thread"):
WebViewerIntegration(bot)
bot.logger.error.assert_called()
msg = bot.logger.error.call_args[0][0]
assert "0.0.0.0" in msg
assert "web_viewer_password" in msg
def test_zero_host_without_password_does_not_log_when_disabled(self):
from modules.web_viewer.integration import WebViewerIntegration
bot = _make_bot()
bot.config.set("Web_Viewer", "enabled", "false")
bot.config.set("Web_Viewer", "host", "0.0.0.0")
bot.config.set("Web_Viewer", "web_viewer_password", "")
with patch.object(WebViewerIntegration, "start_viewer"):
with patch("modules.web_viewer.integration.BotIntegration._init_http_session"), \
patch("modules.web_viewer.integration.BotIntegration._init_packet_stream_table"), \
patch("modules.web_viewer.integration.BotIntegration._start_drain_thread"):
WebViewerIntegration(bot)
bot.logger.error.assert_not_called()
def test_start_viewer_refuses_missing_db_parent(self, tmp_path):
from modules.web_viewer.integration import WebViewerIntegration
bot = _make_bot()
bot.config_file = str(tmp_path / "config.ini")
Path(bot.config_file).write_text("[Bot]\n", encoding="utf-8")
bot.config.set("Web_Viewer", "db_path", "missing/viewer.db")
bot.db_manager.db_path = str(tmp_path / "bot.db")
with patch("modules.web_viewer.integration.BotIntegration._init_http_session"), \
patch("modules.web_viewer.integration.BotIntegration._init_packet_stream_table"), \
patch("modules.web_viewer.integration.BotIntegration._start_drain_thread"):
wvi = WebViewerIntegration(bot)
with patch.object(wvi, "_run_viewer") as run_viewer:
wvi.start_viewer()
run_viewer.assert_not_called()
assert wvi.running is False
assert any(
"parent directory does not exist" in str(call.args[0])
for call in bot.logger.error.call_args_list
)
def test_start_viewer_refuses_db_parent_that_is_file(self, tmp_path):
from modules.web_viewer.integration import WebViewerIntegration
bot = _make_bot()
bot.config_file = str(tmp_path / "config.ini")
Path(bot.config_file).write_text("[Bot]\n", encoding="utf-8")
not_dir = tmp_path / "not_a_directory"
not_dir.write_text("not a directory", encoding="utf-8")
bot.config.set("Web_Viewer", "db_path", "not_a_directory/viewer.db")
bot.db_manager.db_path = str(tmp_path / "bot.db")
with patch("modules.web_viewer.integration.BotIntegration._init_http_session"), \
patch("modules.web_viewer.integration.BotIntegration._init_packet_stream_table"), \
patch("modules.web_viewer.integration.BotIntegration._start_drain_thread"):
wvi = WebViewerIntegration(bot)
with patch.object(wvi, "_run_viewer") as run_viewer:
wvi.start_viewer()
run_viewer.assert_not_called()
assert wvi.running is False
assert any(
"parent path is not a directory" in str(call.args[0])
for call in bot.logger.error.call_args_list
)
def test_start_viewer_allows_valid_db_parent(self, tmp_path):
from modules.web_viewer.integration import WebViewerIntegration
bot = _make_bot()
bot.config_file = str(tmp_path / "config.ini")
Path(bot.config_file).write_text("[Bot]\n", encoding="utf-8")
bot.config.set("Web_Viewer", "db_path", "viewer.db")
bot.db_manager.db_path = str(tmp_path / "bot.db")
with patch("modules.web_viewer.integration.BotIntegration._init_http_session"), \
patch("modules.web_viewer.integration.BotIntegration._init_packet_stream_table"), \
patch("modules.web_viewer.integration.BotIntegration._start_drain_thread"):
wvi = WebViewerIntegration(bot)
with patch("modules.web_viewer.integration.threading.Thread") as thread_cls:
thread = thread_cls.return_value
wvi.start_viewer()
thread.start.assert_called_once()
assert wvi.running is True
class TestNormalizedWebViewerPassword:
def test_blank_and_null_placeholders(self):
from modules.web_viewer.integration import normalized_web_viewer_password
c = ConfigParser()
c.add_section("Web_Viewer")
assert normalized_web_viewer_password(c) == ""
c.set("Web_Viewer", "web_viewer_password", "")
assert normalized_web_viewer_password(c) == ""
c.set("Web_Viewer", "web_viewer_password", " ")
assert normalized_web_viewer_password(c) == ""
c.set("Web_Viewer", "web_viewer_password", '""')
assert normalized_web_viewer_password(c) == ""
c.set("Web_Viewer", "web_viewer_password", "null")
assert normalized_web_viewer_password(c) == ""
c.set("Web_Viewer", "web_viewer_password", "NONE")
assert normalized_web_viewer_password(c) == ""
def test_real_password_preserved(self):
from modules.web_viewer.integration import normalized_web_viewer_password
c = ConfigParser()
c.add_section("Web_Viewer")
c.set("Web_Viewer", "web_viewer_password", "secret")
assert normalized_web_viewer_password(c) == "secret"
c.set("Web_Viewer", "web_viewer_password", '"quoted"')
assert normalized_web_viewer_password(c) == "quoted"
# ---------------------------------------------------------------------------
# shutdown
# ---------------------------------------------------------------------------
class TestShutdown:
def test_shutdown_sets_flag(self):
bi = _make_bot_integration()
bi._drain_thread = Mock()
bi._drain_thread.is_alive.return_value = False
bi.shutdown()
assert bi.is_shutting_down is True
def test_shutdown_stops_drain_thread(self):
bi = _make_bot_integration()
bi._drain_thread = Mock()
bi._drain_thread.is_alive.return_value = True
bi.shutdown()
bi._drain_thread.join.assert_called_once()
class TestIntegrationTimeoutConfig:
def test_bot_integration_loads_custom_timeouts(self):
bot = _make_bot()
bot.config.set("Web_Viewer", "edge_post_timeout_sec", "2.5")
bot.config.set("Web_Viewer", "node_post_timeout_sec", "1.25")
bot.config.set("Web_Viewer", "sqlite_connect_timeout_sec", "42")
bot.config.set("Web_Viewer", "requeue_put_timeout_sec", "7")
bot.config.set("Web_Viewer", "integration_shutdown_join_timeout_sec", "3")
bi = _make_bot_integration(bot)
assert bi.edge_post_timeout_sec == 2.5
assert bi.node_post_timeout_sec == 1.25
assert bi.sqlite_connect_timeout_sec == 42
assert bi.requeue_put_timeout_sec == 7
assert bi.shutdown_join_timeout_sec == 3
def test_web_viewer_integration_loads_custom_timeouts(self):
from modules.web_viewer.integration import WebViewerIntegration
bot = _make_bot()
bot.config.set("Web_Viewer", "viewer_stop_grace_timeout_sec", "9")
bot.config.set("Web_Viewer", "viewer_stop_force_timeout_sec", "4")
bot.config.set("Web_Viewer", "port_cleanup_lsof_timeout_sec", "8")
bot.config.set("Web_Viewer", "port_cleanup_kill_timeout_sec", "1")
with patch("modules.web_viewer.integration.BotIntegration._init_http_session"), \
patch("modules.web_viewer.integration.BotIntegration._init_packet_stream_table"), \
patch("modules.web_viewer.integration.BotIntegration._start_drain_thread"):
wvi = WebViewerIntegration(bot)
assert wvi.viewer_stop_grace_timeout_sec == 9
assert wvi.viewer_stop_force_timeout_sec == 4
assert wvi.port_cleanup_lsof_timeout_sec == 8
assert wvi.port_cleanup_kill_timeout_sec == 1
class TestViewerLogDir:
"""Web viewer subprocess log dir follows [Logging] log_file."""
def test_empty_log_file_returns_none(self):
from modules.web_viewer.integration import WebViewerIntegration
bot = _make_bot()
bot.config.add_section("Logging")
bot.config.set("Logging", "log_file", "")
with patch("modules.web_viewer.integration.BotIntegration._init_http_session"), \
patch("modules.web_viewer.integration.BotIntegration._init_packet_stream_table"), \
patch("modules.web_viewer.integration.BotIntegration._start_drain_thread"):
wvi = WebViewerIntegration(bot)
assert wvi._viewer_log_dir("/tmp/config.ini") is None
def test_missing_logging_section_returns_none(self):
from modules.web_viewer.integration import WebViewerIntegration
bot = _make_bot()
assert not bot.config.has_section("Logging")
with patch("modules.web_viewer.integration.BotIntegration._init_http_session"), \
patch("modules.web_viewer.integration.BotIntegration._init_packet_stream_table"), \
patch("modules.web_viewer.integration.BotIntegration._start_drain_thread"):
wvi = WebViewerIntegration(bot)
assert wvi._viewer_log_dir("/tmp/config.ini") is None
def test_absolute_log_file_uses_parent(self, tmp_path: Path):
from modules.web_viewer.integration import WebViewerIntegration
bot = _make_bot()
log_dir = tmp_path / "var" / "log" / "meshcore-bot"
log_dir.mkdir(parents=True)
bot.config.add_section("Logging")
bot.config.set("Logging", "log_file", str(log_dir / "meshcore_bot.log"))
with patch("modules.web_viewer.integration.BotIntegration._init_http_session"), \
patch("modules.web_viewer.integration.BotIntegration._init_packet_stream_table"), \
patch("modules.web_viewer.integration.BotIntegration._start_drain_thread"):
wvi = WebViewerIntegration(bot)
assert wvi._viewer_log_dir(str(tmp_path / "config.ini")) == log_dir.resolve()
Reference in new issue View Git Blame Copy Permalink