mirror of
https://github.com/agessaman/meshcore-bot.git
synced 2026-10-17 03:29:52 +00:00
agessaman
bb3bfe321b
Closes #279. A MeshCore client with no region configured sends every channel message as a plain FLOOD, which every repeater on the mesh rebroadcasts. This adds two things: free observation of how much of that the bot hears, and an opt-in warning to the senders. Observation classifies each channel message as scoped, global or unknown and tallies it per channel per local day. It costs one upsert and no airtime, and it is what lets an operator see the size of the problem before deciding to spend airtime on it. The classification runs ahead of the flood_scopes allowlist, because an unscoped message is exactly what that allowlist drops. Warnings only ever fire on positive RF evidence of an unscoped FLOOD. Absence of correlation is not proof that a sender omitted a region, so it classifies as unknown and stays quiet. They are off by default, start in dry run, and are fenced by min_unscoped_messages, a per-sender cooldown, a mesh-wide cooldown and a daily cap on attempts. Dry run consumes the same budget it previews, so the log is what going live would put on the air, not an upper bound. All three limits read from the database, so a restart cannot release a burst. Channel-delivered warnings go out at global scope on purpose: the recipient is by definition outside any region the bot replies under. New Settings -> Region Warnings page in the web viewer covers all of it. Its dark-mode striped rows exposed a pre-existing base.html bug where Bootstrap's light-theme text color survived on a dark row background (~1.3:1); fixed there for every table in the app.
181 lines
7.4 KiB
Python
181 lines
7.4 KiB
Python
"""Tests for the /region-warnings page and its API."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import configparser
|
|
from datetime import datetime
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
|
|
from modules import region_warning
|
|
|
|
|
|
@pytest.fixture
|
|
def viewer(tmp_path):
|
|
"""BotDataViewer against a temp config + empty DB (migrations create tables)."""
|
|
from modules.web_viewer.app import BotDataViewer
|
|
|
|
config = configparser.ConfigParser()
|
|
config.add_section("Bot")
|
|
config.set("Bot", "db_path", str(tmp_path / "meshcore_bot.db"))
|
|
config.set("Bot", "bot_name", "TestBot")
|
|
config.add_section("Web_Viewer")
|
|
for key, value in [
|
|
("host", "127.0.0.1"), ("port", "8080"), ("enabled", "false"),
|
|
("auto_start", "false"), ("debug", "false"),
|
|
("cors_allowed_origins", "*"), ("web_viewer_password", ""),
|
|
]:
|
|
config.set("Web_Viewer", key, value)
|
|
|
|
config_path = str(tmp_path / "config.ini")
|
|
with open(config_path, "w") as handle:
|
|
config.write(handle)
|
|
|
|
with patch.object(BotDataViewer, "_start_database_polling"), \
|
|
patch.object(BotDataViewer, "_start_log_tailing"), \
|
|
patch.object(BotDataViewer, "_start_cleanup_scheduler"), \
|
|
patch.object(BotDataViewer, "_start_dashboard_refresher"), \
|
|
patch.object(BotDataViewer, "_setup_socketio_handlers"), \
|
|
patch("modules.web_viewer.app.RepeaterManager"):
|
|
instance = BotDataViewer(
|
|
db_path=str(tmp_path / "meshcore_bot.db"), config_path=config_path)
|
|
instance.app.testing = True
|
|
return instance
|
|
|
|
|
|
def _seed_tally(viewer, channel="#general", scoped=30, globally=10, unknown=5):
|
|
today = datetime.now().date().isoformat()
|
|
with viewer.db_manager.connection() as conn:
|
|
conn.execute(
|
|
"INSERT INTO region_scope_daily "
|
|
"(date, channel, scoped_count, global_count, unknown_count) "
|
|
"VALUES (?, ?, ?, ?, ?)",
|
|
(today, channel, scoped, globally, unknown),
|
|
)
|
|
conn.commit()
|
|
|
|
|
|
class TestRegionWarningsPage:
|
|
def test_page_renders(self, viewer):
|
|
resp = viewer.app.test_client().get("/region-warnings")
|
|
assert resp.status_code == 200
|
|
assert b"Region Warnings" in resp.data
|
|
|
|
def test_page_script_carries_a_csp_nonce(self, viewer):
|
|
"""The endpoint is in the nonce-hardened set, so its script must be nonced."""
|
|
resp = viewer.app.test_client().get("/region-warnings")
|
|
body = resp.data.decode()
|
|
assert "<script nonce=" in body
|
|
assert "unsafe-inline" not in resp.headers["Content-Security-Policy"].split("style-src")[0]
|
|
|
|
|
|
class TestRegionWarningsApi:
|
|
def test_defaults_when_section_absent(self, viewer):
|
|
data = viewer.app.test_client().get("/api/region-warnings").get_json()
|
|
assert data["settings"]["enabled"] == "false"
|
|
assert data["settings"]["dry_run"] == "true"
|
|
assert data["traffic"]["channels"] == []
|
|
assert data["events"] == []
|
|
|
|
def test_limits_account_for_the_bot_name(self, viewer):
|
|
data = viewer.app.test_client().get("/api/region-warnings").get_json()
|
|
assert data["limits"]["dm"] == region_warning.DM_BODY_LIMIT
|
|
assert data["limits"]["channel"] == max(130, 160 - len("TestBot") - 2)
|
|
|
|
def test_traffic_reflects_seeded_tallies(self, viewer):
|
|
_seed_tally(viewer)
|
|
data = viewer.app.test_client().get("/api/region-warnings").get_json()
|
|
channel = data["traffic"]["channels"][0]
|
|
assert channel["channel"] == "#general"
|
|
assert channel["unscoped_pct"] == 25.0
|
|
assert len(data["series"]) == 14
|
|
|
|
def test_window_is_clamped(self, viewer):
|
|
data = viewer.app.test_client().get("/api/region-warnings?days=9999").get_json()
|
|
assert data["traffic"]["days"] == 90
|
|
|
|
def test_bad_window_falls_back(self, viewer):
|
|
data = viewer.app.test_client().get("/api/region-warnings?days=abc").get_json()
|
|
assert data["traffic"]["days"] == 14
|
|
|
|
|
|
class TestRegionWarningsSave:
|
|
def test_save_writes_config_and_queues_reload(self, viewer, tmp_path):
|
|
resp = viewer.app.test_client().post("/api/region-warnings/settings", json={
|
|
"enabled": True,
|
|
"dry_run": True,
|
|
"delivery": "channel",
|
|
"channels": "#General, general, weather",
|
|
"message": "Set a region, {sender}",
|
|
"min_unscoped_messages": 5,
|
|
"per_sender_cooldown_hours": 24,
|
|
"mesh_cooldown_minutes": 15,
|
|
"max_warnings_per_day": 3,
|
|
"track_traffic": False,
|
|
})
|
|
assert resp.status_code == 200
|
|
body = resp.get_json()
|
|
assert body["success"] is True
|
|
assert body["reload_queued"] is True
|
|
# Duplicates collapse and the leading # is dropped.
|
|
assert body["settings"]["channels"] == "general, weather"
|
|
|
|
text = (tmp_path / "config.ini").read_text(encoding="utf-8")
|
|
assert "[Region_Warnings]" in text
|
|
written = configparser.ConfigParser()
|
|
written.read(tmp_path / "config.ini", encoding="utf-8")
|
|
settings = region_warning.load_settings(written)
|
|
assert settings.enabled is True
|
|
assert settings.delivery == "channel"
|
|
assert settings.channels == ("general", "weather")
|
|
assert settings.min_unscoped_messages == 5
|
|
assert settings.track_traffic is False
|
|
|
|
def test_rejects_unknown_delivery(self, viewer):
|
|
resp = viewer.app.test_client().post(
|
|
"/api/region-warnings/settings", json={"delivery": "smoke signal"})
|
|
assert resp.status_code == 400
|
|
assert "delivery" in resp.get_json()["error"]
|
|
|
|
def test_rejects_out_of_range_number(self, viewer):
|
|
resp = viewer.app.test_client().post(
|
|
"/api/region-warnings/settings",
|
|
json={"delivery": "dm", "min_unscoped_messages": 0})
|
|
assert resp.status_code == 400
|
|
|
|
def test_rejects_non_numeric(self, viewer):
|
|
resp = viewer.app.test_client().post(
|
|
"/api/region-warnings/settings",
|
|
json={"delivery": "dm", "mesh_cooldown_minutes": "soon"})
|
|
assert resp.status_code == 400
|
|
|
|
def test_rejects_multiline_message(self, viewer):
|
|
"""A newline in an INI value would break the file open at the next read."""
|
|
resp = viewer.app.test_client().post(
|
|
"/api/region-warnings/settings",
|
|
json={"delivery": "dm", "message": "line one\nline two"})
|
|
assert resp.status_code == 400
|
|
|
|
def test_empty_message_falls_back_to_the_default(self, viewer):
|
|
resp = viewer.app.test_client().post(
|
|
"/api/region-warnings/settings", json={"delivery": "dm", "message": " "})
|
|
assert resp.status_code == 200
|
|
assert resp.get_json()["settings"]["message"] == region_warning.DEFAULT_MESSAGE
|
|
|
|
def test_channels_accepts_a_list(self, viewer):
|
|
resp = viewer.app.test_client().post(
|
|
"/api/region-warnings/settings",
|
|
json={"delivery": "dm", "channels": ["#a", "b"]})
|
|
assert resp.get_json()["settings"]["channels"] == "a, b"
|
|
|
|
def test_saved_settings_are_read_back_by_the_get(self, viewer):
|
|
viewer.app.test_client().post("/api/region-warnings/settings", json={
|
|
"enabled": True, "dry_run": False, "delivery": "dm",
|
|
"max_warnings_per_day": 2,
|
|
})
|
|
data = viewer.app.test_client().get("/api/region-warnings").get_json()
|
|
assert data["settings"]["enabled"] == "true"
|
|
assert data["settings"]["dry_run"] == "false"
|
|
assert data["budget"]["cap"] == 2
|