Add unit tests for HTTP server, main daemon, service utilities, SQLite handler, and update endpoints

repeater/web/update_endpoints.py

@@ -25,7 +25,7 @@ import threading
 import time
 import urllib.error
 import urllib.request
-from datetime import datetime
+from datetime import datetime, timezone
 from typing import List, Optional
 
 import cherrypy
@@ -384,7 +384,7 @@ class _UpdateState:
             if fresh != "unknown":
                 self.current_version = fresh
             self.has_update = _has_update(self.current_version, latest)
-            self.last_checked = datetime.utcnow()
+            self.last_checked = datetime.now(timezone.utc)
             self.state = "idle"
             self.error_message = None
 
@@ -392,7 +392,7 @@ class _UpdateState:
         with self._lock:
             self.state = "error"
             self.error_message = msg
-            self.last_checked = datetime.utcnow()
+            self.last_checked = datetime.now(timezone.utc)
 
     def _fail_check_ratelimit(self, msg: str, reset_at: Optional[datetime]) -> None:
         """Like _fail_check but keeps existing version data intact and records
@@ -401,7 +401,7 @@ class _UpdateState:
             # Keep state as idle so the UI still shows version info
             self.state = "idle"
             self.error_message = msg
-            self.last_checked = datetime.utcnow()
+            self.last_checked = datetime.now(timezone.utc)
             self.rate_limit_until = reset_at
 
     def start_install(self, thread: threading.Thread) -> bool:
@@ -463,7 +463,7 @@ def _fetch_url(url: str, timeout: int = 10) -> str:
             try:
                 reset_ts = exc.headers.get("X-RateLimit-Reset")
                 if reset_ts:
-                    reset_at = datetime.utcfromtimestamp(int(reset_ts))
+                    reset_at = datetime.fromtimestamp(int(reset_ts), timezone.utc)
             except Exception:
                 pass
             reset_str = reset_at.strftime("%H:%M UTC") if reset_at else "a short while"
@@ -982,7 +982,7 @@ class UpdateAPIEndpoints:
 
         # Respect GitHub rate-limit backoff window
         if not force and _state.rate_limit_until is not None:
-            remaining = (_state.rate_limit_until - datetime.utcnow()).total_seconds()
+            remaining = (_state.rate_limit_until - datetime.now(timezone.utc)).total_seconds()
             if remaining > 0:
                 reset_str = _state.rate_limit_until.strftime("%H:%M UTC")
                 return self._ok({
@@ -992,7 +992,7 @@ class UpdateAPIEndpoints:
                 })
 
         if not force and snap["last_checked"] is not None:
-            age = (datetime.utcnow() - _state.last_checked).total_seconds()
+            age = (datetime.now(timezone.utc) - _state.last_checked).total_seconds()
             if age < CHECK_CACHE_TTL and snap["latest_version"] is not None:
                 return self._ok({"message": "Using cached result", "state": snap["state"], **snap})
 

tests/test_api_endpoints_core_coverage.py

@@ -12,6 +12,9 @@ def _make_api(config=None):
     api = APIEndpoints.__new__(APIEndpoints)
     api.config = config or {}
     api.daemon_instance = None
+    api.send_advert_func = None
+    api.event_loop = None
+    api.stats_getter = None
     api._config_path = "/tmp/test-config.yaml"
     api.config_manager = MagicMock()
     return api
@@ -1159,3 +1162,647 @@ def test_config_import_identity_redaction_preserves_by_name_for_room_servers(che
         assert by_name["main-room"] == bytes.fromhex("ABCD")
         # Unknown existing room keeps empty value when imported as redacted.
         assert by_name["new-room"] == ""
+
+
+def test_stats_includes_versions_and_buildroot_image_info(cherrypy_ctx):
+    del cherrypy_ctx
+    api = _make_api({"radio_type": "sx1262", "web": {"site_name": "Field"}})
+    api.stats_getter = lambda: {"uptime": 10}
+
+    with patch("repeater.web.api_endpoints.get_buildroot_image_info", return_value={"image_name": "pyMC", "image_version": "1.2.3"}):
+        out = api.stats()
+
+    assert out["uptime"] == 10
+    assert out["radio_type"] == "sx1262"
+    assert out["site_name"] == "Field"
+    assert out["version"]
+    assert out["image_name"] == "pyMC"
+    assert out["image_version"] == "1.2.3"
+
+
+def test_gps_snapshot_when_service_present_and_default_when_absent(cherrypy_ctx):
+    del cherrypy_ctx
+    api = _make_api({"gps": {"enabled": True}})
+
+    api.daemon_instance = SimpleNamespace(gps_service=SimpleNamespace(get_snapshot=lambda: {"running": True}))
+    out = api.gps()
+    assert out == {"success": True, "data": {"running": True}}
+
+    api.daemon_instance = SimpleNamespace(gps_service=None)
+    out2 = api.gps()
+    assert out2["success"] is True
+    assert out2["data"]["status"]["state"] == "disabled"
+
+
+def test_check_pymc_console_and_mqtt_status_and_broker_presets(cherrypy_ctx):
+    request, _ = cherrypy_ctx
+    api = _make_api()
+
+    request.method = "OPTIONS"
+    assert api.check_pymc_console() == ""
+
+    request.method = "GET"
+    with patch("os.path.isdir", return_value=True):
+        out = api.check_pymc_console()
+    assert out["success"] is True
+    assert out["data"]["exists"] is True
+
+    # mqtt status when no handler reachable
+    api.daemon_instance = None
+    status = api.mqtt_status()
+    assert status["success"] is True
+    assert status["data"]["handler_active"] is False
+
+    # mqtt status with active connections
+    conn = SimpleNamespace(
+        enabled=True,
+        broker={"name": "main", "host": "mqtt.local"},
+        is_connected=lambda: True,
+        has_pending_reconnect=lambda: False,
+        format="json",
+    )
+    _attach_storage(api, SimpleNamespace(mqtt_handler=SimpleNamespace(connections=[conn])))
+    status2 = api.mqtt_status()
+    assert status2["success"] is True
+    assert status2["data"]["brokers"][0]["name"] == "main"
+
+    with patch("repeater.presets.list_presets", return_value=["waev"]), patch(
+        "repeater.presets.get_preset",
+        return_value={"display_name": "Waev", "website": "https://waev.app", "brokers": [{"host": "h"}]},
+    ):
+        presets = api.broker_presets()
+    assert presets["success"] is True
+    assert presets["data"][0]["id"] == "waev"
+
+
+def test_send_advert_paths(cherrypy_ctx):
+    request, _ = cherrypy_ctx
+    api = _make_api()
+
+    request.method = "OPTIONS"
+    assert api.send_advert() == ""
+
+    request.method = "POST"
+    no_func = api.send_advert()
+    assert no_func["success"] is False
+    assert "not configured" in no_func["error"]
+
+    api.send_advert_func = MagicMock()
+    no_loop = api.send_advert()
+    assert no_loop["success"] is False
+    assert "Event loop not available" in no_loop["error"]
+
+    api.event_loop = object()
+    api.send_advert_func = MagicMock()
+    fake_future = SimpleNamespace(result=lambda timeout: True)
+    with patch("asyncio.run_coroutine_threadsafe", return_value=fake_future):
+        ok = api.send_advert()
+    assert ok == {"success": True, "data": "Advert sent successfully"}
+
+    fake_future_fail = SimpleNamespace(result=lambda timeout: False)
+    with patch("asyncio.run_coroutine_threadsafe", return_value=fake_future_fail):
+        bad = api.send_advert()
+    assert bad["success"] is False
+
+
+def test_set_mode_and_set_duty_cycle_paths(cherrypy_ctx):
+    request, _ = cherrypy_ctx
+    api = _make_api({"repeater": {}, "duty_cycle": {}})
+
+    request.method = "OPTIONS"
+    assert api.set_mode() == ""
+    assert api.set_duty_cycle() == ""
+
+    request.method = "POST"
+    request.json = {"mode": "invalid"}
+    invalid = api.set_mode()
+    assert invalid["success"] is False
+
+    request.json = {"mode": "monitor"}
+    ok = api.set_mode()
+    assert ok == {"success": True, "mode": "monitor"}
+    assert api.config["repeater"]["mode"] == "monitor"
+
+    request.json = {"enabled": False}
+    duty = api.set_duty_cycle()
+    assert duty == {"success": True, "enabled": False}
+    assert api.config["duty_cycle"]["enforcement_enabled"] is False
+
+
+def test_update_duty_cycle_config_branches(cherrypy_ctx):
+    request, _ = cherrypy_ctx
+    api = _make_api({"duty_cycle": {}})
+
+    request.method = "OPTIONS"
+    assert api.update_duty_cycle_config() == ""
+
+    request.method = "POST"
+    request.json = {"max_airtime_percent": 0.01}
+    bad = api.update_duty_cycle_config()
+    assert bad["success"] is False
+    assert "0.1-100.0" in bad["error"]
+
+    request.json = {}
+    none = api.update_duty_cycle_config()
+    assert none["success"] is False
+    assert "No valid settings" in none["error"]
+
+    request.json = {"max_airtime_percent": 10, "enforcement_enabled": True}
+    api.config_manager.update_and_save.return_value = {"saved": False, "error": "disk"}
+    save_fail = api.update_duty_cycle_config()
+    assert save_fail["success"] is False
+    assert "disk" in save_fail["error"]
+
+    api.config_manager.update_and_save.return_value = {"saved": True, "live_updated": True}
+    ok = api.update_duty_cycle_config()
+    assert ok["success"] is True
+    assert ok["data"]["persisted"] is True
+    assert api.config["duty_cycle"]["max_airtime_per_minute"] == 6000
+
+
+def test_update_advert_rate_limit_config_branches(cherrypy_ctx):
+    request, _ = cherrypy_ctx
+    api = _make_api({"repeater": {}})
+
+    request.method = "OPTIONS"
+    assert api.update_advert_rate_limit_config() == ""
+
+    request.method = "POST"
+    request.json = {}
+    none = api.update_advert_rate_limit_config()
+    assert none["success"] is False
+    assert "No valid settings" in none["error"]
+
+    request.json = {
+        "rate_limit_enabled": True,
+        "bucket_capacity": 0,
+        "refill_tokens": 0,
+        "refill_interval_seconds": 1,
+        "min_interval_seconds": -10,
+        "penalty_enabled": True,
+        "violation_threshold": 0,
+        "violation_decay_seconds": 1,
+        "base_penalty_seconds": 1,
+        "penalty_multiplier": 0.1,
+        "max_penalty_seconds": 1,
+        "adaptive_enabled": True,
+        "ewma_alpha": 99,
+        "hysteresis_seconds": -1,
+        "quiet_max": 0.05,
+        "normal_max": 0.2,
+        "busy_max": 0.5,
+    }
+    api.config_manager.update_and_save.return_value = {"saved": True, "live_updated": False}
+    ok = api.update_advert_rate_limit_config()
+    assert ok["success"] is True
+    rate = api.config["repeater"]["advert_rate_limit"]
+    pen = api.config["repeater"]["advert_penalty_box"]
+    ad = api.config["repeater"]["advert_adaptive"]
+    assert rate["bucket_capacity"] == 1
+    assert rate["refill_tokens"] == 1
+    assert rate["refill_interval_seconds"] == 60
+    assert rate["min_interval_seconds"] == 0
+    assert pen["violation_threshold"] == 1
+    assert pen["base_penalty_seconds"] == 60
+    assert pen["penalty_multiplier"] == 1.0
+    assert pen["max_penalty_seconds"] == 60
+    assert ad["ewma_alpha"] == 1.0
+    assert ad["hysteresis_seconds"] == 0
+    assert ad["thresholds"]["quiet_max"] == 0.05
+
+
+def test_logs_hardware_stats_and_hardware_processes(cherrypy_ctx):
+    del cherrypy_ctx
+    api = _make_api()
+
+    with patch("repeater.web.http_server._log_buffer", SimpleNamespace(logs=[])):
+        logs = api.logs()
+    assert "logs" in logs
+    assert logs["logs"][0]["message"] == "No logs available"
+
+    storage = SimpleNamespace(
+        get_hardware_stats=MagicMock(return_value={"cpu": 10}),
+        get_hardware_processes=MagicMock(return_value=[{"pid": 1}]),
+    )
+    _attach_storage(api, storage)
+
+    hs = api.hardware_stats()
+    hp = api.hardware_processes()
+    assert hs == {"success": True, "data": {"cpu": 10}}
+    assert hp == {"success": True, "data": [{"pid": 1}]}
+
+    storage.get_hardware_stats.return_value = None
+    assert api.hardware_stats()["success"] is False
+
+    storage.get_hardware_processes.return_value = None
+    assert api.hardware_processes()["success"] is False
+
+
+def test_noise_floor_and_crc_endpoints(cherrypy_ctx):
+    del cherrypy_ctx
+    api = _make_api()
+    storage = SimpleNamespace(
+        get_noise_floor_history=MagicMock(return_value=[{"v": -110}]),
+        get_noise_floor_stats=MagicMock(return_value={"avg": -105}),
+        get_noise_floor_rrd=MagicMock(return_value=[[1, -100]]),
+        get_crc_error_count=MagicMock(return_value=3),
+        get_crc_error_history=MagicMock(return_value=[{"id": 1}]),
+    )
+    _attach_storage(api, storage)
+
+    h = api.noise_floor_history(hours="24", limit="5")
+    s = api.noise_floor_stats(hours="12")
+    c = api.noise_floor_chart_data(hours="2")
+    cc = api.crc_error_count(hours="6")
+    ch = api.crc_error_history(hours="6", limit="10")
+
+    assert h["success"] is True and h["data"]["count"] == 1
+    assert s["success"] is True and s["data"]["stats"]["avg"] == -105
+    assert c["success"] is True and c["data"]["chart_data"] == [[1, -100]]
+    assert cc["success"] is True and cc["data"]["crc_error_count"] == 3
+    assert ch["success"] is True and ch["data"]["count"] == 1
+
+    err = api.crc_error_count(hours="bad")
+    assert err["success"] is False
+
+
+def test_advert_contact_and_rate_limit_stats_endpoints(cherrypy_ctx):
+    del cherrypy_ctx
+    api = _make_api()
+
+    miss = api.adverts_by_contact_type()
+    assert miss["success"] is False
+
+    storage = SimpleNamespace(
+        sqlite_handler=SimpleNamespace(
+            get_adverts_by_contact_type=MagicMock(return_value=[{"id": 1}]),
+            get_adverts_count_by_contact_type=MagicMock(return_value=7),
+        )
+    )
+    _attach_storage(api, storage)
+
+    out = api.adverts_by_contact_type(contact_type="room_server", limit="2", offset="0", hours="24")
+    count = api.adverts_count_by_contact_type(contact_type="room_server", hours="24")
+    assert out["success"] is True and out["count"] == 1
+    assert count["success"] is True and count["data"]["count"] == 7
+
+    bad_fmt = api.adverts_count_by_contact_type(contact_type="room_server", hours="bad")
+    assert bad_fmt["success"] is False
+
+    no_daemon = api.advert_rate_limit_stats()
+    assert no_daemon["success"] is False
+
+    api.daemon_instance = SimpleNamespace(advert_helper=None)
+    no_helper = api.advert_rate_limit_stats()
+    assert no_helper["success"] is False
+
+    api.daemon_instance = SimpleNamespace(advert_helper=SimpleNamespace())
+    no_method = api.advert_rate_limit_stats()
+    assert no_method["success"] is False
+
+    api.daemon_instance = SimpleNamespace(
+        advert_helper=SimpleNamespace(get_rate_limit_stats=lambda: {"tier": "normal"})
+    )
+    ok = api.advert_rate_limit_stats()
+    assert ok == {"success": True, "data": {"tier": "normal"}}
+
+
+def test_transport_keys_and_transport_key_and_unscoped_policy(cherrypy_ctx):
+    request, _ = cherrypy_ctx
+    api = _make_api({"mesh": {}})
+    storage = SimpleNamespace(
+        get_transport_keys=MagicMock(return_value=[{"id": 1}]),
+        create_transport_key=MagicMock(return_value=10),
+        get_transport_key_by_id=MagicMock(return_value={"id": 1}),
+        update_transport_key=MagicMock(return_value=True),
+        delete_transport_key=MagicMock(return_value=True),
+    )
+    _attach_storage(api, storage)
+
+    request.method = "GET"
+    keys = api.transport_keys()
+    assert keys["success"] is True and keys["count"] == 1
+
+    request.method = "POST"
+    request.json = {"name": "", "flood_policy": "allow"}
+    assert api.transport_keys()["success"] is False
+
+    request.json = {"name": "k1", "flood_policy": "invalid"}
+    assert api.transport_keys()["success"] is False
+
+    request.json = {"name": "k1", "flood_policy": "allow", "last_used": "bad"}
+    created = api.transport_keys()
+    assert created["success"] is True
+
+    request.method = "GET"
+    assert api.transport_key("x")["success"] is False
+    assert api.transport_key("1")["success"] is True
+
+    request.method = "PUT"
+    request.json = {"flood_policy": "maybe"}
+    assert api.transport_key("1")["success"] is False
+
+    request.json = {"name": "new", "flood_policy": "deny", "last_used": "not-ts"}
+    updated = api.transport_key("1")
+    assert updated["success"] is True
+
+    request.method = "DELETE"
+    deleted = api.transport_key("1")
+    assert deleted["success"] is True
+
+    request.method = "GET"
+    assert api.unscoped_flood_policy()["success"] is False
+
+    request.method = "POST"
+    request.json = {}
+    assert api.unscoped_flood_policy()["success"] is False
+
+    request.json = {"unscoped_flood_allow": "yes"}
+    assert api.unscoped_flood_policy()["success"] is False
+
+    api.config_manager.save_to_file.return_value = True
+    request.json = {"unscoped_flood_allow": True}
+    ok = api.unscoped_flood_policy()
+    assert ok["success"] is True
+    assert api.config["mesh"]["unscoped_flood_allow"] is True
+
+
+class _FakeIdentityObj:
+    def __init__(self, first=0x42):
+        self._pk = bytes([first]) + (b"A" * 31)
+
+    def get_public_key(self):
+        return self._pk
+
+    def get_address_bytes(self):
+        return b"\x12\x34"
+
+
+class _FakeClient:
+    def __init__(self, key_hex: str, admin: bool):
+        self.id = SimpleNamespace(get_public_key=lambda: bytes.fromhex(key_hex))
+        self._admin = admin
+        self.last_activity = 1.0
+        self.last_login_success = 2.0
+        self.last_timestamp = 3.0
+
+    def is_admin(self):
+        return self._admin
+
+
+class _FakeACL:
+    def __init__(self, clients, admin_password="a", guest_password="g"):
+        self._clients = list(clients)
+        self.max_clients = 10
+        self.admin_password = admin_password
+        self.guest_password = guest_password
+        self.allow_read_only = True
+
+    def get_num_clients(self):
+        return len(self._clients)
+
+    def get_all_clients(self):
+        return list(self._clients)
+
+    def remove_client(self, pubkey):
+        before = len(self._clients)
+        self._clients = [c for c in self._clients if c.id.get_public_key() != pubkey]
+        return len(self._clients) < before
+
+
+def test_identity_endpoints_paths(cherrypy_ctx):
+    request, response = cherrypy_ctx
+    api = _make_api({"identities": {"room_servers": [], "companions": []}})
+
+    request.method = "OPTIONS"
+    assert api.identities() == ""
+
+    request.method = "GET"
+    assert api.identities()["success"] is False
+
+    id_mgr = SimpleNamespace(
+        list_identities=lambda: [{"name": "room_server:main", "hash": "0x42", "address": "1234"}],
+        get_identities_by_type=lambda t: (
+            [("main", _FakeIdentityObj(0x42), {"settings": {"x": 1}})]
+            if t == "room_server"
+            else [("comp1", _FakeIdentityObj(0x51), {"settings": {"tcp_port": 5000}})]
+        ),
+        get_identity_by_name=lambda n: (_FakeIdentityObj(0x42), {}, "room_server") if n == "main" else None,
+        named_identities={"comp1": 1, "main": 1},
+    )
+    api.daemon_instance = SimpleNamespace(identity_manager=id_mgr)
+    api.config = {
+        "identities": {
+            "room_servers": [{"name": "main", "identity_key": "a" * 64, "settings": {"x": 1}}],
+            "companions": [{"name": "comp1", "identity_key": "b" * 64, "settings": {"tcp_port": 5000}}],
+        }
+    }
+    api.config_manager.save_to_file.return_value = True
+
+    ids = api.identities()
+    assert ids["success"] is True
+    assert ids["data"]["total_configured"] == 1
+    assert ids["data"]["total_configured_companions"] == 1
+
+    assert api.identity()["success"] is False
+    assert api.identity(name="missing")["success"] is False
+    one = api.identity(name="main")
+    assert one["success"] is True
+    assert one["data"]["runtime"]["registered"] is True
+
+    # create identity validation + success path
+    request.method = "POST"
+    request.json = {}
+    assert api.create_identity()["success"] is False
+    request.json = {"name": "x", "type": "invalid"}
+    assert api.create_identity()["success"] is False
+    request.json = {"name": "x", "type": "room_server", "settings": {"admin_password": "p", "guest_password": "p"}}
+    assert api.create_identity()["success"] is False
+    request.json = {"name": "comp1", "type": "companion", "identity_key": "aa" * 32}
+    assert api.create_identity()["success"] is False
+
+    request.json = {"name": "new-comp", "type": "companion", "identity_key": "cc" * 32, "settings": {"node_name": "N"}}
+    api.event_loop = object()
+    api.daemon_instance = SimpleNamespace(add_companion_from_config=MagicMock())
+    with patch("asyncio.run_coroutine_threadsafe", return_value=SimpleNamespace(result=lambda timeout: True)):
+        created = api.create_identity()
+    assert created["success"] is True
+
+    # update identity method guard and room_server success
+    request.method = "GET"
+    with pytest.raises(cherrypy.HTTPError):
+        api.update_identity()
+    request.method = "PUT"
+    api.daemon_instance = None
+    request.json = {"name": "main", "settings": {"node_name": "updated"}}
+    upd = api.update_identity()
+    assert upd["success"] is True
+
+    # delete identity paths
+    request.method = "GET"
+    with pytest.raises(cherrypy.HTTPError):
+        api.delete_identity(name="main")
+    request.method = "DELETE"
+    assert api.delete_identity(name="", type="room_server")["success"] is False
+    deleted = api.delete_identity(name="main", type="room_server")
+    assert deleted["success"] is True
+
+    # companion delete
+    api.config["identities"]["companions"] = [{"name": "comp1", "identity_key": "11" * 32}]
+    api.daemon_instance = SimpleNamespace(identity_manager=id_mgr)
+    d2 = api.delete_identity(name="comp1", type="companion")
+    assert d2["success"] is True
+    assert "comp1" not in id_mgr.named_identities
+    assert response.status in (200, 405)
+
+
+def test_acl_endpoints_paths(cherrypy_ctx):
+    request, _ = cherrypy_ctx
+    api = _make_api()
+
+    request.method = "OPTIONS"
+    assert api.acl_info() == ""
+    assert api.acl_clients() == ""
+    assert api.acl_remove_client() == ""
+    assert api.acl_stats() == ""
+
+    request.method = "GET"
+    assert api.acl_info()["success"] is False
+
+    clients = [_FakeClient("aa" * 32, True), _FakeClient("bb" * 32, False)]
+    acl = _FakeACL(clients)
+    login_helper = SimpleNamespace(get_acl_dict=lambda: {0x42: acl, 0x51: _FakeACL([])})
+    id_mgr = SimpleNamespace(
+        get_identities_by_type=lambda t: (
+            [("room1", _FakeIdentityObj(0x42), {})] if t == "room_server" else [("comp1", _FakeIdentityObj(0x51), {})]
+        )
+    )
+    local = _FakeIdentityObj(0x42)
+    frame_server = SimpleNamespace(companion_hash="0x51", _client_writer=SimpleNamespace(get_extra_info=lambda k: ("10.0.0.2", 1234)))
+    api.daemon_instance = SimpleNamespace(
+        login_helper=login_helper,
+        identity_manager=id_mgr,
+        local_identity=local,
+        companion_bridges={0x51: object()},
+        companion_frame_servers=[frame_server],
+    )
+
+    info = api.acl_info()
+    assert info["success"] is True
+    assert info["data"]["total_identities"] >= 2
+
+    all_clients = api.acl_clients()
+    assert all_clients["success"] is True
+    assert all_clients["data"]["count"] >= 1
+    assert api.acl_clients(identity_hash="bad")["success"] is False
+    assert api.acl_clients(identity_name="missing")["success"] is False
+
+    request.method = "POST"
+    request.json = {}
+    assert api.acl_remove_client()["success"] is False
+    request.json = {"public_key": "zz"}
+    assert api.acl_remove_client()["success"] is False
+    request.json = {"public_key": "aa" * 32, "identity_hash": "0x42"}
+    removed = api.acl_remove_client()
+    assert removed["success"] is True
+
+    request.method = "GET"
+    st = api.acl_stats()
+    assert st["success"] is True
+    assert st["data"]["total_identities"] >= 1
+
+
+def test_room_endpoint_slice(cherrypy_ctx):
+    request, _ = cherrypy_ctx
+    api = _make_api()
+
+    request.method = "OPTIONS"
+    assert api.room_messages(room_name="x") == ""
+    assert api.room_stats(room_name="x") == ""
+    assert api.room_clients(room_name="x") == ""
+    assert api.room_message(room_name="x") == ""
+    assert api.room_messages_clear(room_name="x") == ""
+
+    # Basic room messages success path via helper patch
+    request.method = "GET"
+    db = SimpleNamespace(
+        get_room_message_count=MagicMock(return_value=1),
+        get_room_messages=MagicMock(return_value=[{"id": 1, "author_pubkey": "aa" * 32, "post_timestamp": 1.0, "sender_timestamp": 1, "message_text": "m", "txt_type": 0}]),
+        get_messages_since=MagicMock(return_value=[]),
+        delete_room_message=MagicMock(return_value=True),
+        clear_room_messages=MagicMock(return_value=1),
+        get_all_room_clients=MagicMock(return_value=[]),
+    )
+    room = SimpleNamespace(db=db, max_posts=10, _running=True, next_push_time=0, last_cleanup_time=0)
+    with patch.object(api, "_get_room_server_by_name_or_hash", return_value={"room_server": room, "name": "room", "hash": 0x42, "identity": None, "config": {}}):
+        _attach_storage(api, SimpleNamespace(get_node_name_by_pubkey=lambda _pk: "Node"))
+        msgs = api.room_messages(room_name="room")
+        assert msgs["success"] is True
+        assert msgs["data"]["count"] == 1
+
+        request.method = "DELETE"
+        one = api.room_message(room_name="room", message_id="1")
+        assert one["success"] is True
+        cleared = api.room_messages_clear(room_name="room")
+        assert cleared["success"] is True
+
+
+def test_update_mqtt_config_validation_and_success(cherrypy_ctx):
+    request, _ = cherrypy_ctx
+    api = _make_api()
+
+    request.method = "OPTIONS"
+    assert api.update_mqtt_config() == ""
+
+    request.method = "POST"
+    request.json = {}
+    assert api.update_mqtt_config()["success"] is False
+
+    request.json = {"brokers": "not-list"}
+    assert api.update_mqtt_config()["success"] is False
+
+    request.json = {"brokers": ["bad"]}
+    assert "must be an object" in api.update_mqtt_config()["error"]
+
+    request.json = {"brokers": [{"name": "n", "host": "h", "port": "x", "format": "json"}]}
+    assert "invalid port" in api.update_mqtt_config()["error"]
+
+    request.json = {
+        "iata_code": "SFO",
+        "status_interval": 20,
+        "brokers": [
+            {"preset": "waev"},
+            {"name": "a", "host": "h", "port": 443, "format": "json", "enabled": True},
+        ],
+    }
+    api.config_manager.update_and_save.return_value = {"success": True, "saved": True}
+    out = api.update_mqtt_config()
+    assert out["success"] is True
+    assert out["data"]["restart_required"] is True
+
+    api.config_manager.update_and_save.return_value = {"success": False, "error": "save failed"}
+    out2 = api.update_mqtt_config()
+    assert out2["success"] is False
+    assert "save failed" in out2["error"]
+
+
+def test_restart_service_options_method_and_result_paths(cherrypy_ctx):
+    request, _ = cherrypy_ctx
+    api = _make_api()
+
+    request.method = "OPTIONS"
+    assert api.restart_service() == ""
+
+    request.method = "GET"
+    with pytest.raises(cherrypy.HTTPError):
+        api.restart_service()
+
+    request.method = "POST"
+    with patch("repeater.service_utils.restart_service", return_value=(True, "ok")):
+        ok = api.restart_service()
+    assert ok == {"success": True, "message": "ok"}
+
+    with patch("repeater.service_utils.restart_service", return_value=(False, "nope")):
+        err = api.restart_service()
+    assert err["success"] is False
+    assert "nope" in err["error"]

tests/test_auth_components.py

@@ -0,0 +1,127 @@
+from types import SimpleNamespace
+from unittest.mock import MagicMock
+
+import cherrypy
+import jwt
+import pytest
+
+from repeater.web.auth.api_tokens import APITokenManager
+from repeater.web.auth.cherrypy_tool import check_auth
+from repeater.web.auth.jwt_handler import JWTHandler
+
+
+def test_jwt_handler_create_and_verify_and_invalid_cases():
+    secret = "test-secret-key-minimum-32-bytes!!"
+    h = JWTHandler(secret, expiry_minutes=15)
+    token = h.create_jwt("admin", "client-1")
+
+    payload = h.verify_jwt(token)
+    assert payload is not None
+    assert payload["sub"] == "admin"
+    assert payload["client_id"] == "client-1"
+
+    expired = jwt.encode({"sub": "admin", "client_id": "c", "iat": 1, "exp": 1}, secret, algorithm="HS256")
+    assert h.verify_jwt(expired) is None
+    assert h.verify_jwt("not-a-token") is None
+
+
+def test_api_token_manager_happy_paths_and_revoke_false():
+    db = SimpleNamespace(
+        create_api_token=MagicMock(return_value=10),
+        verify_api_token=MagicMock(return_value={"id": 10, "name": "n1"}),
+        revoke_api_token=MagicMock(side_effect=[True, False]),
+        list_api_tokens=MagicMock(return_value=[{"id": 10, "name": "n1"}]),
+    )
+
+    mgr = APITokenManager(sqlite_handler=db, secret_key="k")
+
+    token_id, plaintext = mgr.create_token("n1")
+    assert token_id == 10
+    assert isinstance(plaintext, str)
+    assert len(plaintext) == 64
+
+    verified = mgr.verify_token(plaintext)
+    assert verified["id"] == 10
+
+    assert mgr.revoke_token(10) is True
+    assert mgr.revoke_token(11) is False
+    assert mgr.list_tokens()[0]["name"] == "n1"
+
+
+def _set_cp(monkeypatch, method="GET", path="/api/private", headers=None, params=None, cfg=None):
+    req = SimpleNamespace(
+        method=method,
+        path_info=path,
+        headers=headers or {},
+        params=params or {},
+        user=None,
+    )
+    resp = SimpleNamespace(status=200, headers={})
+    monkeypatch.setattr(cherrypy, "request", req, raising=False)
+    monkeypatch.setattr(cherrypy, "response", resp, raising=False)
+    monkeypatch.setattr(cherrypy, "config", cfg or {}, raising=False)
+    return req, resp
+
+
+def test_check_auth_skips_options_and_login(monkeypatch):
+    _set_cp(monkeypatch, method="OPTIONS")
+    assert check_auth() is None
+
+    _set_cp(monkeypatch, method="GET", path="/auth/login")
+    assert check_auth() is None
+
+
+def test_check_auth_missing_handlers_returns_500_json(monkeypatch):
+    _set_cp(monkeypatch, cfg={})
+    out = check_auth()
+    assert out["success"] is False
+    assert cherrypy.response.status == 500
+
+
+def test_check_auth_accepts_bearer_token(monkeypatch):
+    jwt_handler = SimpleNamespace(verify_jwt=lambda _t: {"sub": "admin", "client_id": "c1"})
+    token_manager = SimpleNamespace(verify_token=lambda _k: None)
+    req, _resp = _set_cp(
+        monkeypatch,
+        headers={"Authorization": "Bearer abc"},
+        cfg={"jwt_handler": jwt_handler, "token_manager": token_manager},
+    )
+
+    assert check_auth() is None
+    assert req.user["auth_type"] == "jwt"
+
+
+def test_check_auth_accepts_query_token_and_removes_it(monkeypatch):
+    jwt_handler = SimpleNamespace(verify_jwt=lambda _t: {"sub": "admin", "client_id": "c2"})
+    token_manager = SimpleNamespace(verify_token=lambda _k: None)
+    req, _resp = _set_cp(
+        monkeypatch,
+        params={"token": "xyz", "x": "1"},
+        cfg={"jwt_handler": jwt_handler, "token_manager": token_manager},
+    )
+
+    assert check_auth() is None
+    assert req.user["auth_type"] == "jwt_query"
+    assert "token" not in req.params
+
+
+def test_check_auth_accepts_api_key(monkeypatch):
+    jwt_handler = SimpleNamespace(verify_jwt=lambda _t: None)
+    token_manager = SimpleNamespace(verify_token=lambda _k: {"id": 3, "name": "svc"})
+    req, _resp = _set_cp(
+        monkeypatch,
+        headers={"X-API-Key": "k"},
+        cfg={"jwt_handler": jwt_handler, "token_manager": token_manager},
+    )
+
+    assert check_auth() is None
+    assert req.user["auth_type"] == "api_token"
+
+
+def test_check_auth_unauthorized_raises_http_error(monkeypatch):
+    jwt_handler = SimpleNamespace(verify_jwt=lambda _t: None)
+    token_manager = SimpleNamespace(verify_token=lambda _k: None)
+    _set_cp(monkeypatch, cfg={"jwt_handler": jwt_handler, "token_manager": token_manager})
+
+    with pytest.raises(cherrypy.HTTPError):
+        check_auth()

tests/test_auth_endpoints.py

@@ -0,0 +1,363 @@
+import io
+import json
+from types import SimpleNamespace
+from unittest.mock import MagicMock
+
+import cherrypy
+import pytest
+
+from repeater.web.auth_endpoints import AuthAPIEndpoints, AuthEndpoints, TokensAPIEndpoint
+
+
+@pytest.fixture
+def cp_ctx(monkeypatch):
+    def _set(method="GET", headers=None, body=b"", path="/api/auth"):
+        req = SimpleNamespace(
+            method=method,
+            headers=headers or {},
+            body=io.BytesIO(body),
+            path_info=path,
+            user=None,
+        )
+        resp = SimpleNamespace(status=200, headers={})
+        cfg = {}
+        monkeypatch.setattr(cherrypy, "request", req, raising=False)
+        monkeypatch.setattr(cherrypy, "response", resp, raising=False)
+        monkeypatch.setattr(cherrypy, "config", cfg, raising=False)
+        return req, resp, cfg
+
+    return _set
+
+
+def _jwt_ok_payload():
+    return {"sub": "admin", "client_id": "cli-1"}
+
+
+def _jwt_handler(ok=True):
+    if ok:
+        return SimpleNamespace(verify_jwt=lambda _token: _jwt_ok_payload(), create_jwt=lambda u, c: "jwt-new", expiry_minutes=15)
+    return SimpleNamespace(verify_jwt=lambda _token: None, create_jwt=lambda u, c: "jwt-new", expiry_minutes=15)
+
+
+def _token_mgr():
+    return SimpleNamespace(
+        verify_token=lambda _k: {"id": 7, "name": "tok"},
+        list_tokens=lambda: [{"id": 1, "name": "a"}],
+        create_token=lambda name: (3, "plain-token"),
+        revoke_token=lambda _id: True,
+    )
+
+
+def test_auth_api_endpoints_constructs_tokens_endpoint():
+    api = AuthAPIEndpoints()
+    assert isinstance(api.tokens, TokensAPIEndpoint)
+
+
+def test_tokens_index_options_and_missing_manager(cp_ctx):
+    endpoint = TokensAPIEndpoint()
+
+    cp_ctx(method="OPTIONS")
+    assert endpoint.index() == {}
+
+    cp_ctx(method="GET", headers={"Authorization": "Bearer x"})
+    with pytest.raises(cherrypy.HTTPError):
+        endpoint.index()
+
+
+def test_tokens_index_get_post_and_error_paths(cp_ctx):
+    endpoint = TokensAPIEndpoint()
+
+    # Authenticated GET success
+    _req, _resp, cfg = cp_ctx(method="GET", headers={"Authorization": "Bearer ok"})
+    cfg["jwt_handler"] = _jwt_handler(ok=True)
+    cfg["token_manager"] = _token_mgr()
+    out = endpoint.index()
+    assert out["success"] is True
+    assert out["tokens"][0]["id"] == 1
+
+    # GET exception
+    _req, _resp, cfg = cp_ctx(method="GET", headers={"Authorization": "Bearer ok"})
+    cfg["jwt_handler"] = _jwt_handler(ok=True)
+    cfg["token_manager"] = SimpleNamespace(list_tokens=lambda: (_ for _ in ()).throw(RuntimeError("db")))
+    out = endpoint.index()
+    assert out["success"] is False
+    assert cherrypy.response.status == 500
+
+    # POST missing name
+    _req, _resp, cfg = cp_ctx(
+        method="POST",
+        headers={"Authorization": "Bearer ok"},
+        body=json.dumps({"name": ""}).encode(),
+    )
+    cfg["jwt_handler"] = _jwt_handler(ok=True)
+    cfg["token_manager"] = _token_mgr()
+    out = endpoint.index()
+    assert out["success"] is False
+    assert cherrypy.response.status == 400
+
+    # POST success
+    _req, _resp, cfg = cp_ctx(
+        method="POST",
+        headers={"Authorization": "Bearer ok"},
+        body=json.dumps({"name": "build-bot"}).encode(),
+    )
+    cfg["jwt_handler"] = _jwt_handler(ok=True)
+    cfg["token_manager"] = _token_mgr()
+    out = endpoint.index()
+    assert out["success"] is True
+    assert out["token"] == "plain-token"
+
+
+def test_tokens_default_delete_paths(cp_ctx):
+    endpoint = TokensAPIEndpoint()
+
+    # Missing token_id
+    _req, _resp, cfg = cp_ctx(method="DELETE", headers={"Authorization": "Bearer ok"})
+    cfg["jwt_handler"] = _jwt_handler(ok=True)
+    cfg["token_manager"] = _token_mgr()
+    out = endpoint.default(token_id=None)
+    assert out["success"] is False
+    assert cherrypy.response.status == 400
+
+    # Invalid token id
+    _req, _resp, cfg = cp_ctx(method="DELETE", headers={"Authorization": "Bearer ok"})
+    cfg["jwt_handler"] = _jwt_handler(ok=True)
+    cfg["token_manager"] = _token_mgr()
+    out = endpoint.default(token_id="abc")
+    assert out["success"] is False
+    assert cherrypy.response.status == 400
+
+    # Not found
+    _req, _resp, cfg = cp_ctx(method="DELETE", headers={"Authorization": "Bearer ok"})
+    cfg["jwt_handler"] = _jwt_handler(ok=True)
+    cfg["token_manager"] = SimpleNamespace(revoke_token=lambda _id: False)
+    out = endpoint.default(token_id="9")
+    assert out["success"] is False
+    assert cherrypy.response.status == 404
+
+    # Success
+    _req, _resp, cfg = cp_ctx(method="DELETE", headers={"Authorization": "Bearer ok"})
+    cfg["jwt_handler"] = _jwt_handler(ok=True)
+    cfg["token_manager"] = _token_mgr()
+    out = endpoint.default(token_id="9")
+    assert out["success"] is True
+
+
+def test_login_paths(cp_ctx):
+    auth = AuthEndpoints(config={"repeater": {"security": {"admin_password": "pw"}}}, jwt_handler=_jwt_handler(ok=True), token_manager=_token_mgr())
+
+    cp_ctx(method="OPTIONS")
+    assert auth.login() == b""
+
+    cp_ctx(method="POST", body=b"{}")
+    out = json.loads(auth.login().decode())
+    assert out["success"] is False
+
+    cp_ctx(method="POST", body=json.dumps({"username": "admin", "password": "pw", "client_id": "abc"}).encode())
+    out = json.loads(auth.login().decode())
+    assert out["success"] is True
+    assert out["token"] == "jwt-new"
+
+    cp_ctx(method="POST", body=json.dumps({"username": "admin", "password": "bad", "client_id": "abc"}).encode())
+    out = json.loads(auth.login().decode())
+    assert out["success"] is False
+
+
+@pytest.mark.asyncio
+async def test_verify_requires_get_and_auth(cp_ctx):
+    auth = AuthEndpoints(config={}, jwt_handler=_jwt_handler(ok=True), token_manager=_token_mgr())
+
+    _req, _resp, cfg = cp_ctx(method="GET", headers={"Authorization": "Bearer ok"})
+    cfg["jwt_handler"] = _jwt_handler(ok=True)
+    cfg["token_manager"] = _token_mgr()
+    out = auth.verify()
+    assert out["success"] is True
+
+    _req, _resp, cfg = cp_ctx(method="POST", headers={"Authorization": "Bearer ok"})
+    cfg["jwt_handler"] = _jwt_handler(ok=True)
+    cfg["token_manager"] = _token_mgr()
+    with pytest.raises(cherrypy.HTTPError):
+        auth.verify()
+
+
+def test_refresh_paths(cp_ctx):
+    auth = AuthEndpoints(config={}, jwt_handler=_jwt_handler(ok=True), token_manager=_token_mgr())
+
+    cp_ctx(method="OPTIONS")
+    assert auth.refresh() == b""
+
+    # unauthorized
+    _req, _resp, cfg = cp_ctx(method="POST", body=b"{}")
+    cfg["jwt_handler"] = _jwt_handler(ok=False)
+    cfg["token_manager"] = SimpleNamespace(verify_token=lambda _k: None)
+    out = json.loads(auth.refresh().decode())
+    assert out["success"] is False
+
+    # missing client id
+    _req, _resp, cfg = cp_ctx(method="POST", headers={"Authorization": "Bearer ok"}, body=b"{}")
+    cfg["jwt_handler"] = _jwt_handler(ok=True)
+    cfg["token_manager"] = _token_mgr()
+    out = json.loads(auth.refresh().decode())
+    assert out["success"] is True  # falls back to payload client_id
+
+    # api token path
+    _req, _resp, cfg = cp_ctx(method="POST", headers={"X-API-Key": "k"}, body=json.dumps({"client_id": "z"}).encode())
+    cfg["jwt_handler"] = _jwt_handler(ok=False)
+    cfg["token_manager"] = _token_mgr()
+    out = json.loads(auth.refresh().decode())
+    assert out["success"] is True
+
+
+def test_change_password_paths(cp_ctx):
+    config = {"repeater": {"security": {"admin_password": "old-password"}}}
+    auth = AuthEndpoints(
+        config=config,
+        jwt_handler=_jwt_handler(ok=True),
+        token_manager=_token_mgr(),
+        config_manager=SimpleNamespace(save_to_file=MagicMock(return_value=True)),
+    )
+
+    cp_ctx(method="OPTIONS")
+    assert auth.change_password() == b""
+
+    # no auth handlers configured in cherrypy config
+    cp_ctx(method="POST", headers={})
+    with pytest.raises(cherrypy.HTTPError):
+        auth.change_password()
+
+    # unauthorized
+    _req, _resp, cfg = cp_ctx(method="POST", headers={}, body=b"{}")
+    cfg["jwt_handler"] = _jwt_handler(ok=False)
+    cfg["token_manager"] = SimpleNamespace(verify_token=lambda _k: None)
+    out = json.loads(auth.change_password().decode())
+    assert out["success"] is False
+    assert cherrypy.response.status == 401
+
+    # missing fields
+    _req, _resp, cfg = cp_ctx(method="POST", headers={"Authorization": "Bearer ok"}, body=b"{}")
+    cfg["jwt_handler"] = _jwt_handler(ok=True)
+    cfg["token_manager"] = _token_mgr()
+    out = json.loads(auth.change_password().decode())
+    assert out["success"] is False
+    assert cherrypy.response.status == 400
+
+    # weak new password
+    _req, _resp, cfg = cp_ctx(
+        method="POST",
+        headers={"Authorization": "Bearer ok"},
+        body=json.dumps({"current_password": "old-password", "new_password": "short"}).encode(),
+    )
+    cfg["jwt_handler"] = _jwt_handler(ok=True)
+    cfg["token_manager"] = _token_mgr()
+    out = json.loads(auth.change_password().decode())
+    assert out["success"] is False
+    assert cherrypy.response.status == 400
+
+    # wrong current password
+    _req, _resp, cfg = cp_ctx(
+        method="POST",
+        headers={"Authorization": "Bearer ok"},
+        body=json.dumps({"current_password": "wrong", "new_password": "new-password"}).encode(),
+    )
+    cfg["jwt_handler"] = _jwt_handler(ok=True)
+    cfg["token_manager"] = _token_mgr()
+    out = json.loads(auth.change_password().decode())
+    assert out["success"] is False
+    assert cherrypy.response.status == 401
+
+    # success
+    _req, _resp, cfg = cp_ctx(
+        method="POST",
+        headers={"Authorization": "Bearer ok"},
+        body=json.dumps({"current_password": "old-password", "new_password": "new-password"}).encode(),
+    )
+    cfg["jwt_handler"] = _jwt_handler(ok=True)
+    cfg["token_manager"] = _token_mgr()
+    out = json.loads(auth.change_password().decode())
+    assert out["success"] is True
+
+    # save fails
+    auth_fail_save = AuthEndpoints(
+        config={"repeater": {"security": {"admin_password": "old-password"}}},
+        jwt_handler=_jwt_handler(ok=True),
+        token_manager=_token_mgr(),
+        config_manager=SimpleNamespace(save_to_file=MagicMock(return_value=False)),
+    )
+    _req, _resp, cfg = cp_ctx(
+        method="POST",
+        headers={"Authorization": "Bearer ok"},
+        body=json.dumps({"current_password": "old-password", "new_password": "new-password"}).encode(),
+    )
+    cfg["jwt_handler"] = _jwt_handler(ok=True)
+    cfg["token_manager"] = _token_mgr()
+    out = json.loads(auth_fail_save.change_password().decode())
+    assert out["success"] is False
+    assert cherrypy.response.status == 500
+
+
+def test_protected_auth_urls_block_unauthenticated_access(cp_ctx):
+    auth = AuthEndpoints(config={}, jwt_handler=_jwt_handler(ok=True), token_manager=_token_mgr())
+    no_auth_cfg = {
+        "jwt_handler": _jwt_handler(ok=False),
+        "token_manager": SimpleNamespace(verify_token=lambda _k: None),
+    }
+
+    # /api/auth/tokens requires auth
+    endpoint = TokensAPIEndpoint()
+    _req, _resp, cfg = cp_ctx(method="GET", path="/api/auth/tokens", headers={})
+    cfg.update(no_auth_cfg)
+    out = endpoint.index()
+    assert out["success"] is False
+    assert cherrypy.response.status == 401
+
+    # /api/auth/tokens/<id> requires auth
+    _req, _resp, cfg = cp_ctx(method="DELETE", path="/api/auth/tokens/1", headers={})
+    cfg.update(no_auth_cfg)
+    out = endpoint.default(token_id="1")
+    assert out["success"] is False
+    assert cherrypy.response.status == 401
+
+    # /api/auth/verify requires auth
+    _req, _resp, cfg = cp_ctx(method="GET", path="/api/auth/verify", headers={})
+    cfg.update(no_auth_cfg)
+    out = auth.verify()
+    assert out["success"] is False
+    assert cherrypy.response.status == 401
+
+    # /api/auth/change_password requires auth
+    _req, _resp, cfg = cp_ctx(
+        method="POST",
+        path="/api/auth/change_password",
+        headers={},
+        body=json.dumps({"current_password": "x", "new_password": "new-password"}).encode(),
+    )
+    cfg.update(no_auth_cfg)
+    out = json.loads(auth.change_password().decode())
+    assert out["success"] is False
+    assert cherrypy.response.status == 401
+
+
+def test_public_and_restricted_auth_url_methods(cp_ctx):
+    auth = AuthEndpoints(
+        config={"repeater": {"security": {"admin_password": "pw"}}},
+        jwt_handler=_jwt_handler(ok=True),
+        token_manager=_token_mgr(),
+    )
+
+    # /api/auth/login is public but only for POST/OPTIONS.
+    cp_ctx(method="GET", path="/api/auth/login")
+    with pytest.raises(cherrypy.HTTPError):
+        auth.login()
+
+    cp_ctx(
+        method="POST",
+        path="/api/auth/login",
+        body=json.dumps({"username": "admin", "password": "pw", "client_id": "client-a"}).encode(),
+    )
+    out = json.loads(auth.login().decode())
+    assert out["success"] is True
+
+    # /api/auth/refresh is not publicly readable.
+    cp_ctx(method="GET", path="/api/auth/refresh")
+    with pytest.raises(cherrypy.HTTPError):
+        auth.refresh()

tests/test_companion_bridge_frame_utils.py

@@ -0,0 +1,208 @@
+from dataclasses import dataclass
+from enum import Enum
+from types import SimpleNamespace
+from typing import Any, cast
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+from pymc_core.companion.constants import RESP_CODE_NO_MORE_MESSAGES
+
+from repeater.companion.bridge import RepeaterCompanionBridge, _to_json_safe
+from repeater.companion.frame_server import CompanionFrameServer
+from repeater.companion.utils import normalize_companion_identity_key, validate_companion_node_name
+
+
+class _Mode(Enum):
+    A = "a"
+
+
+@dataclass
+class _Dc:
+    n: int
+    b: bytes
+
+
+def test_to_json_safe_handles_enums_bytes_collections_and_dataclass():
+    payload = {
+        "enum": _Mode.A,
+        "bytes": b"\x01\x02",
+        "tuple": (1, _Mode.A, b"x"),
+        "dc": _Dc(3, b"\xff"),
+        "nested": {"k": _Mode.A},
+    }
+
+    out = _to_json_safe(payload)
+    assert out["enum"] == "a"
+    assert out["bytes"] == "0102"
+    assert out["tuple"] == [1, "a", "78"]
+    assert out["dc"] == {"n": 3, "b": "ff"}
+    assert out["nested"]["k"] == "a"
+
+
+def test_bridge_save_prefs_persists_and_calls_callback():
+    @dataclass
+    class _Prefs:
+        node_name: str
+        retry: int
+
+    sqlite = SimpleNamespace(companion_save_prefs=MagicMock())
+    callback = MagicMock()
+
+    bridge = object.__new__(RepeaterCompanionBridge)
+    bridge._sqlite_handler = sqlite
+    bridge._companion_hash = "abc123"
+    bridge._on_prefs_saved = callback
+    bridge.prefs = cast(Any, _Prefs(node_name="node-1", retry=2))
+
+    bridge._save_prefs()
+
+    sqlite.companion_save_prefs.assert_called_once()
+    args = sqlite.companion_save_prefs.call_args[0]
+    assert args[0] == "abc123"
+    assert args[1]["node_name"] == "node-1"
+    callback.assert_called_once_with("node-1")
+
+
+def test_bridge_load_prefs_merges_known_fields_with_type_conversion():
+    @dataclass
+    class _Prefs:
+        node_name: str = "orig"
+        retries: int = 1
+        enabled: bool = False
+        ratio: float = 0.5
+
+    stored = {
+        "node_name": "new-name",
+        "retries": "7",
+        "enabled": 1,
+        "ratio": "1.25",
+        "unknown": "ignore",
+        "retries_bad": "NaN",
+    }
+    sqlite = SimpleNamespace(companion_load_prefs=lambda _h: stored)
+
+    bridge = object.__new__(RepeaterCompanionBridge)
+    bridge._sqlite_handler = sqlite
+    bridge._companion_hash = "hash"
+    bridge.prefs = cast(Any, _Prefs())
+
+    bridge._load_prefs()
+
+    assert bridge.prefs.node_name == "new-name"
+    assert cast(Any, bridge.prefs).retries == 7
+    assert cast(Any, bridge.prefs).enabled is True
+    assert cast(Any, bridge.prefs).ratio == 1.25
+
+
+def test_bridge_load_prefs_ignores_invalid_or_missing_backend():
+    @dataclass
+    class _Prefs:
+        node_name: str = "orig"
+
+    bridge = object.__new__(RepeaterCompanionBridge)
+    bridge._sqlite_handler = None
+    bridge._companion_hash = ""
+    bridge.prefs = cast(Any, _Prefs())
+    bridge._load_prefs()
+    assert bridge.prefs.node_name == "orig"
+
+
+@pytest.mark.asyncio
+async def test_frame_server_persistence_paths_and_stop():
+    sqlite = SimpleNamespace(
+        companion_push_message=MagicMock(),
+        companion_pop_message=MagicMock(
+            return_value={
+                "sender_key": b"k",
+                "txt_type": 1,
+                "timestamp": 2,
+                "text": "hello",
+                "is_channel": True,
+                "channel_idx": 3,
+                "path_len": 1,
+            }
+        ),
+        companion_save_contacts=MagicMock(),
+        companion_save_channels=MagicMock(),
+        companion_upsert_contact=MagicMock(),
+    )
+    bridge = SimpleNamespace(
+        message_queue=SimpleNamespace(pop_last=MagicMock()),
+        sync_next_message=lambda: None,
+        get_contacts=lambda: [],
+        channels=SimpleNamespace(max_channels=2),
+        get_channel=lambda idx: None,
+    )
+
+    with patch("repeater.companion.frame_server._BaseFrameServer.__init__", lambda self, **kwargs: None), patch(
+        "repeater.companion.frame_server._BaseFrameServer.stop", AsyncMock()
+    ) as base_stop:
+        srv = CompanionFrameServer(bridge=bridge, companion_hash="h", sqlite_handler=sqlite)
+        srv.bridge = bridge
+        srv.companion_hash = "h"
+        srv._write_frame = MagicMock()
+        srv._build_message_frame = MagicMock(return_value=b"frame")
+
+        await srv._persist_companion_message({"text": "x"})
+        sqlite.companion_push_message.assert_called_once_with("h", {"text": "x"})
+        bridge.message_queue.pop_last.assert_called_once()
+
+        msg = srv._sync_next_from_persistence()
+        assert msg is not None
+        assert msg.text == "hello"
+
+        await srv._cmd_sync_next_message(b"")
+        srv._write_frame.assert_called_once_with(b"frame")
+
+        contact = SimpleNamespace(
+            public_key=b"\x01\x02",
+            name="n",
+            adv_type=1,
+            flags=0,
+            out_path_len=1,
+            out_path=b"\x03",
+            last_advert_timestamp=4,
+            lastmod=5,
+            gps_lat=1.1,
+            gps_lon=2.2,
+            sync_since=6,
+        )
+        await srv._persist_contact(contact)
+        sqlite.companion_upsert_contact.assert_called_once()
+
+        bridge.get_contacts = lambda: [contact]
+        bridge.get_channel = lambda idx: (SimpleNamespace(name="c1", secret="s") if idx == 1 else None)
+        await srv.stop()
+
+        sqlite.companion_save_contacts.assert_called_once()
+        sqlite.companion_save_channels.assert_called_once_with(
+            "h", [{"channel_idx": 1, "name": "c1", "secret": "s"}]
+        )
+        base_stop.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_frame_server_no_more_messages_response_when_empty():
+    bridge = SimpleNamespace(sync_next_message=lambda: None)
+
+    with patch("repeater.companion.frame_server._BaseFrameServer.__init__", lambda self, **kwargs: None):
+        srv = CompanionFrameServer(bridge=bridge, companion_hash="h", sqlite_handler=None)
+        srv.bridge = bridge
+        srv._write_frame = MagicMock()
+        await srv._cmd_sync_next_message(b"")
+        # RESP_CODE_NO_MORE_MESSAGES is encoded as a single-byte frame.
+        assert srv._write_frame.call_args[0][0] == bytes([RESP_CODE_NO_MORE_MESSAGES])
+
+
+def test_companion_utils_validation_and_normalization():
+    assert normalize_companion_identity_key(" 0xAABB ") == "AABB"
+    assert validate_companion_node_name("  node-1  ") == "node-1"
+
+    with pytest.raises(ValueError):
+        validate_companion_node_name(cast(Any, 123))
+    with pytest.raises(ValueError):
+        validate_companion_node_name("   ")
+    with pytest.raises(ValueError):
+        validate_companion_node_name("x" * 32)
+    with pytest.raises(ValueError):
+        validate_companion_node_name("bad\nname")

tests/test_companion_ws_proxy.py

@@ -0,0 +1,172 @@
+from types import SimpleNamespace
+from unittest.mock import MagicMock
+
+import cherrypy
+import pytest
+
+from repeater.web import companion_ws_proxy as proxy
+
+
+@pytest.fixture
+def cp_cfg(monkeypatch):
+    cfg = {}
+    monkeypatch.setattr(cherrypy, "config", cfg, raising=False)
+    return cfg
+
+
+def _ws(query_string):
+    ws = object.__new__(proxy.CompanionFrameWebSocket)
+    ws.environ = {"QUERY_STRING": query_string}
+    ws.close = MagicMock()
+    ws.send = MagicMock()
+    ws._teardown = MagicMock()
+    return ws
+
+
+def test_opened_rejects_missing_jwt_handler(cp_cfg):
+    ws = _ws("token=t&companion_name=c1")
+    ws.opened()
+    ws.close.assert_called_once_with(code=1011, reason="server configuration error")
+
+
+def test_opened_rejects_missing_token(cp_cfg):
+    cp_cfg["jwt_handler"] = SimpleNamespace(verify_jwt=lambda _t: {"sub": "u"})
+    ws = _ws("companion_name=c1")
+    ws.opened()
+    ws.close.assert_called_once_with(code=1008, reason="unauthorized")
+
+
+def test_opened_rejects_invalid_token(cp_cfg):
+    cp_cfg["jwt_handler"] = SimpleNamespace(verify_jwt=lambda _t: None)
+    ws = _ws("token=t&companion_name=c1")
+    ws.opened()
+    ws.close.assert_called_once_with(code=1008, reason="unauthorized")
+
+
+def test_opened_rejects_missing_companion_name(cp_cfg):
+    cp_cfg["jwt_handler"] = SimpleNamespace(verify_jwt=lambda _t: {"sub": "u"})
+    ws = _ws("token=t")
+    ws.opened()
+    ws.close.assert_called_once_with(code=1008, reason="missing companion_name")
+
+
+def test_opened_rejects_missing_companion_endpoint(cp_cfg):
+    cp_cfg["jwt_handler"] = SimpleNamespace(verify_jwt=lambda _t: {"sub": "u"})
+    ws = _ws("token=t&companion_name=c1")
+    ws._resolve_tcp_endpoint = MagicMock(return_value=None)
+    ws.opened()
+    ws.close.assert_called_once_with(code=1008, reason="companion not found")
+
+
+def test_opened_tcp_connect_failure(cp_cfg, monkeypatch):
+    cp_cfg["jwt_handler"] = SimpleNamespace(verify_jwt=lambda _t: {"sub": "u"})
+    ws = _ws("token=t&companion_name=c1")
+    ws._resolve_tcp_endpoint = MagicMock(return_value=("127.0.0.1", 5000))
+
+    fake_socket = MagicMock()
+    fake_socket.connect.side_effect = RuntimeError("nope")
+    monkeypatch.setattr(proxy.socket, "socket", lambda *_args, **_kwargs: fake_socket)
+
+    ws.opened()
+    ws.close.assert_called_once_with(code=1011, reason="TCP connect failed")
+
+
+def test_opened_success_starts_reader_thread(cp_cfg, monkeypatch):
+    cp_cfg["jwt_handler"] = SimpleNamespace(verify_jwt=lambda _t: {"sub": "u"})
+    ws = _ws("token=t&companion_name=c1")
+    ws._resolve_tcp_endpoint = MagicMock(return_value=("127.0.0.1", 5000))
+
+    fake_socket = MagicMock()
+    monkeypatch.setattr(proxy.socket, "socket", lambda *_args, **_kwargs: fake_socket)
+
+    thread_started = {"started": False}
+
+    class _T:
+        def __init__(self, *args, **kwargs):
+            self.args = args
+            self.kwargs = kwargs
+
+        def start(self):
+            thread_started["started"] = True
+
+    monkeypatch.setattr(proxy.threading, "Thread", _T)
+
+    ws.opened()
+    assert ws._closing is False
+    assert ws._companion_name == "c1"
+    assert thread_started["started"] is True
+
+
+def test_resolve_tcp_endpoint_paths(monkeypatch):
+    ws = _ws("token=t")
+
+    # no daemon
+    proxy.set_daemon(None)
+    assert ws._resolve_tcp_endpoint("c1") is None
+
+    # daemon missing identity manager
+    proxy.set_daemon(SimpleNamespace(companion_bridges={1: object()}, config={}))
+    assert ws._resolve_tcp_endpoint("c1") is None
+
+    # daemon with empty bridges
+    daemon = SimpleNamespace(
+        identity_manager=SimpleNamespace(get_identities_by_type=lambda _t: [("c1", SimpleNamespace(get_public_key=lambda: b"\x01"), {})]),
+        companion_bridges={},
+        config={"identities": {"companions": []}},
+    )
+    proxy.set_daemon(daemon)
+    assert ws._resolve_tcp_endpoint("c1") is None
+
+    # found in identity+bridge and in config, bind 0.0.0.0 => loopback
+    daemon = SimpleNamespace(
+        identity_manager=SimpleNamespace(get_identities_by_type=lambda _t: [("c1", SimpleNamespace(get_public_key=lambda: b"\x01"), {})]),
+        companion_bridges={1: object()},
+        config={"identities": {"companions": [{"name": "c1", "settings": {"tcp_port": 6000, "bind_address": "0.0.0.0"}}]}},
+    )
+    proxy.set_daemon(daemon)
+    assert ws._resolve_tcp_endpoint("c1") == ("127.0.0.1", 6000)
+
+    # found bridge but missing in config
+    daemon.config = {"identities": {"companions": []}}
+    assert ws._resolve_tcp_endpoint("c1") is None
+
+
+def test_received_message_and_closed_paths():
+    ws = _ws("token=t")
+    ws._closing = False
+    ws._tcp = MagicMock()
+
+    ws.received_message(SimpleNamespace(data="abc"))
+    ws._tcp.sendall.assert_called_once_with(b"abc")
+
+    ws._tcp.sendall.side_effect = RuntimeError("sendfail")
+    ws.received_message(SimpleNamespace(data=b"x"))
+    ws._teardown.assert_called_once()
+
+    ws.closed(1000, "done")
+    assert ws._teardown.call_count == 2
+
+
+def test_tcp_to_ws_and_teardown():
+    ws = _ws("token=t")
+    ws._teardown = MagicMock()
+    ws._companion_name = "c1"
+    ws._closing = False
+
+    tcp = MagicMock()
+    tcp.recv.side_effect = [b"a", b""]
+    ws._tcp = tcp
+    ws._tcp_to_ws()
+    ws.send.assert_called_once_with(b"a", binary=True)
+    ws._teardown.assert_called_once()
+
+    # teardown closes tcp and closes websocket when active
+    ws2 = _ws("token=t")
+    ws2._closing = False
+    ws2._companion_name = "c2"
+    tcp_ref = MagicMock()
+    ws2._tcp = tcp_ref
+    ws2._teardown = proxy.CompanionFrameWebSocket._teardown.__get__(ws2, proxy.CompanionFrameWebSocket)
+    ws2._teardown()
+    tcp_ref.close.assert_called_once()
+    ws2.close.assert_called_once()

tests/test_handler_helpers_mesh_cli.py

@@ -0,0 +1,209 @@
+from types import SimpleNamespace
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from repeater.handler_helpers.mesh_cli import MeshCLI
+
+
+def _base_config():
+    return {
+        "version": "3.2.1",
+        "repeater": {
+            "name": "node-a",
+            "mode": "forward",
+            "latitude": 1.2,
+            "longitude": 3.4,
+            "airtime_factor": 1.1,
+            "advert_interval_minutes": 120,
+            "flood_advert_interval_hours": 24,
+            "max_flood_hops": 20,
+            "rx_delay_base": 0.2,
+            "tx_delay_factor": 1.3,
+            "direct_tx_delay_factor": 0.6,
+            "multi_acks": 2,
+            "interference_threshold": -115,
+            "agc_reset_interval": 8,
+        },
+        "radio": {
+            "frequency": 915000000,
+            "bandwidth": 125000,
+            "spreading_factor": 7,
+            "coding_rate": 5,
+            "tx_power": 22,
+        },
+        "security": {"guest_password": "guest", "allow_read_only": True},
+    }
+
+
+def _cfg_mgr(save_ok=True, err=None):
+    return SimpleNamespace(
+        save_to_file=MagicMock(return_value=(save_ok, err)),
+        live_update_daemon=MagicMock(),
+    )
+
+
+def test_handle_command_admin_and_prefix_behavior():
+    cli = MeshCLI("/tmp/cfg.yaml", _base_config(), _cfg_mgr())
+
+    assert cli.handle_command(b"a", "help", is_admin=False) == "Error: Admin permission required"
+    assert cli.handle_command(b"a", "12|help set", is_admin=True).startswith("12|")
+
+
+def test_help_routing_and_basic_unknown_paths():
+    cli = MeshCLI("/tmp/cfg.yaml", _base_config(), _cfg_mgr(), enable_regions=False)
+
+    assert "pyMC CLI Commands" in cli._route_command("help")
+    assert "No detailed help" in cli._route_command("help nope")
+    assert cli._route_command("start ota").startswith("Error:")
+    assert cli._route_command("sensor read").startswith("Error:")
+    assert cli._route_command("gps on").startswith("Error:")
+    assert cli._route_command("stats-foo").startswith("Error:")
+    assert cli._route_command("region load x").startswith("Error: Region commands not available")
+    assert cli._route_command("unknown") == "Unknown command"
+
+
+def test_cmd_advert_branches_and_success_schedule():
+    cli = MeshCLI("/tmp/cfg.yaml", _base_config(), _cfg_mgr(), send_advert_callback=MagicMock())
+
+    # No callback configured.
+    cli_no_cb = MeshCLI("/tmp/cfg.yaml", _base_config(), _cfg_mgr(), send_advert_callback=None)
+    assert cli_no_cb._cmd_advert().startswith("Error: Advert functionality")
+
+    # Callback present but no event loop.
+    cli._event_loop = None
+    assert cli._cmd_advert() == "Error: Event loop not available"
+
+    # Event loop available/running and schedule succeeds.
+    fake_loop = SimpleNamespace(is_running=lambda: True)
+    cli._event_loop = fake_loop
+
+    with patch("asyncio.run_coroutine_threadsafe", side_effect=lambda coro, _loop: coro.close()) as run_ts:
+        out = cli._cmd_advert()
+
+    assert out == "OK - Advert sent"
+    run_ts.assert_called_once()
+
+
+def test_cmd_password_save_success_failure_and_exception():
+    cfg = _base_config()
+    ok_mgr = _cfg_mgr(save_ok=True)
+    cli_ok = MeshCLI("/tmp/cfg.yaml", cfg, ok_mgr)
+
+    assert cli_ok._cmd_password("password   ") == "Error: Password cannot be empty"
+    assert cli_ok._cmd_password("password newpw") == "password now: newpw"
+    ok_mgr.live_update_daemon.assert_called_once_with(["security"])
+
+    bad_mgr = _cfg_mgr(save_ok=False, err="disk")
+    cli_bad = MeshCLI("/tmp/cfg.yaml", _base_config(), bad_mgr)
+    assert "Failed to save config" in cli_bad._cmd_password("password x")
+
+    ex_mgr = SimpleNamespace(
+        save_to_file=MagicMock(side_effect=RuntimeError("boom")),
+        live_update_daemon=MagicMock(),
+    )
+    cli_ex = MeshCLI("/tmp/cfg.yaml", _base_config(), ex_mgr)
+    assert cli_ex._cmd_password("password x") == "Error: Failed to save password"
+
+
+def test_cmd_get_public_key_and_neighbor_branches():
+    cli = MeshCLI("/tmp/cfg.yaml", _base_config(), _cfg_mgr())
+
+    assert cli._cmd_get("public.key") == "Error: Identity not available"
+
+    cli.identity = SimpleNamespace(get_public_key=lambda: b"\x01" * 32)
+    assert cli._cmd_get("public.key") == "> " + (b"\x01" * 32).hex()
+
+    cli.identity = SimpleNamespace(get_public_key=MagicMock(side_effect=RuntimeError("bad")))
+    assert cli._cmd_get("public.key").startswith("Error:")
+
+    # neighbors: no storage
+    assert cli._cmd_neighbors() == "Error: Storage not available"
+
+    # neighbors: empty, filtered empty, then formatted output
+    storage = SimpleNamespace(get_neighbors=lambda: {})
+    cli.storage_handler = storage
+    assert cli._cmd_neighbors() == "No neighbors discovered yet"
+
+    storage.get_neighbors = lambda: {"aa": {"is_repeater": False, "zero_hop": False, "last_seen": 1}}
+    assert "No repeaters or zero hop" in cli._cmd_neighbors()
+
+    storage.get_neighbors = lambda: {
+        "abcdef12feed": {"is_repeater": True, "zero_hop": False, "last_seen": 10, "snr": 4.9},
+        "11223344aabb": {"is_repeater": False, "zero_hop": True, "last_seen": 20, "snr": 1.2},
+    }
+    with patch("time.time", return_value=30):
+        out = cli._cmd_neighbors()
+
+    assert "abcdef12:20:4" in out
+    assert "11223344:10:1" in out
+
+    cli.storage_handler = SimpleNamespace(get_neighbors=MagicMock(side_effect=RuntimeError("db fail")))
+    assert cli._cmd_neighbors().startswith("Error:")
+
+
+def test_cmd_set_updates_and_validation_errors():
+    cfg = _base_config()
+    mgr = _cfg_mgr()
+    cli = MeshCLI("/tmp/cfg.yaml", cfg, mgr)
+
+    assert cli._cmd_set("af 2.5") == "OK"
+    assert cfg["repeater"]["airtime_factor"] == 2.5
+
+    assert cli._cmd_set("name node-z") == "OK"
+    assert cfg["repeater"]["node_name"] == "node-z"
+
+    assert cli._cmd_set("repeat off").endswith("OFF")
+    assert cfg["repeater"]["mode"] == "monitor"
+
+    assert cli._cmd_set("radio 900000000 250000 9 6").startswith("OK")
+    assert cfg["radio"]["frequency"] == 900000000.0
+
+    assert cli._cmd_set("freq 868000000").startswith("OK")
+    assert cli._cmd_set("tx 17") == "OK"
+    assert cli._cmd_set("guest.password g") == "OK"
+    assert cli._cmd_set("allow.read.only off") == "OK"
+
+    assert cli._cmd_set("advert.interval 59").startswith("Error: interval range")
+    assert cli._cmd_set("flood.advert.interval 2").startswith("Error: interval range")
+    assert cli._cmd_set("flood.max 100") == "Error: max 64"
+    assert cli._cmd_set("rxdelay -1") == "Error: cannot be negative"
+    assert cli._cmd_set("txdelay -1") == "Error: cannot be negative"
+    assert cli._cmd_set("direct.txdelay -1") == "Error: cannot be negative"
+
+    assert cli._cmd_set("agc.reset.interval 10") == "OK - interval rounded to 8"
+    assert cli._cmd_set("bad") == "Error: Missing value"
+    assert cli._cmd_set("tx nope").startswith("Error: invalid value")
+    assert cli._cmd_set("unknown.key 1") == "unknown config: unknown.key"
+
+
+def test_misc_commands_and_routes():
+    cli = MeshCLI("/tmp/cfg.yaml", _base_config(), _cfg_mgr(), enable_regions=True)
+
+    assert cli._cmd_region("region") .startswith("Error:")
+    assert cli._cmd_region("region load us").startswith("Error:")
+    assert cli._cmd_region("region save").startswith("Error:")
+    assert cli._cmd_region("region remove x").startswith("Error:")
+    assert cli._cmd_region("region unknown").startswith("Err -")
+
+    assert cli._cmd_setperm("setperm") == "Err - bad params"
+    assert cli._cmd_setperm("setperm abc zz") == "Err - invalid permissions"
+    assert cli._cmd_setperm("setperm abc 2").startswith("Error:")
+
+    assert cli._cmd_tempradio("tempradio 1 2 3").startswith("Error: Expected")
+    assert cli._cmd_tempradio("tempradio 299 125 7 5 10") == "Error: invalid frequency"
+    assert cli._cmd_tempradio("tempradio 915 6 7 5 10") == "Error: invalid bandwidth"
+    assert cli._cmd_tempradio("tempradio 915 125 4 5 10") == "Error: invalid spreading factor"
+    assert cli._cmd_tempradio("tempradio 915 125 7 9 10") == "Error: invalid coding rate"
+    assert cli._cmd_tempradio("tempradio 915 125 7 5 0") == "Error: invalid timeout"
+    assert cli._cmd_tempradio("tempradio 915 125 7 5 nope") == "Error, invalid params"
+    assert cli._cmd_tempradio("tempradio 915 125 7 5 10").startswith("Error:")
+
+    assert cli._cmd_neighbor_remove("neighbor.remove   ") == "ERR: Missing pubkey"
+    assert cli._cmd_neighbor_remove("neighbor.remove abc").startswith("Error:")
+
+    assert cli._cmd_log("log start").startswith("Error:")
+    assert cli._cmd_log("log stop").startswith("Error:")
+    assert cli._cmd_log("log erase").startswith("Error:")
+    assert cli._cmd_log("log") == "Error: Use journalctl to view logs"
+    assert cli._cmd_log("log whatever") == "Unknown log command"

tests/test_http_server_unit.py

@@ -0,0 +1,129 @@
+import io
+import logging
+from pathlib import Path
+from types import SimpleNamespace
+from unittest.mock import MagicMock
+
+import cherrypy
+import pytest
+
+from repeater.web import http_server as hs
+
+
+def test_log_buffer_emit_collects_messages():
+    buf = hs.LogBuffer(max_lines=2)
+    rec1 = logging.LogRecord("x", logging.INFO, __file__, 1, "hello", (), None)
+    rec2 = logging.LogRecord("x", logging.ERROR, __file__, 2, "boom", (), None)
+    rec3 = logging.LogRecord("x", logging.WARNING, __file__, 3, "warn", (), None)
+
+    buf.emit(rec1)
+    buf.emit(rec2)
+    buf.emit(rec3)
+
+    assert len(buf.logs) == 2
+    assert buf.logs[-1]["level"] == "WARNING"
+    assert "warn" in buf.logs[-1]["message"]
+
+
+def test_doc_endpoint_routes_and_openapi_json_paths(monkeypatch):
+    api = SimpleNamespace(docs=lambda: "docs-html")
+    doc = hs.DocEndpoint(api)
+
+    assert doc.index() == "docs-html"
+    assert doc.docs() == "docs-html"
+
+    monkeypatch.setattr(cherrypy, "response", SimpleNamespace(headers={}, status=200), raising=False)
+
+    # success path
+    monkeypatch.setattr("builtins.open", lambda *args, **kwargs: io.StringIO("openapi: 3.0.0\n"))
+    out = doc.openapi_json()
+    assert cherrypy.response.headers["Content-Type"] == "application/json"
+    assert b"openapi" in out
+
+    # not found
+    def _missing(*_args, **_kwargs):
+        raise FileNotFoundError
+
+    monkeypatch.setattr("builtins.open", _missing)
+    out = doc.openapi_json()
+    assert cherrypy.response.status == 404
+    assert b"not found" in out
+
+    # generic error
+    def _err(*_args, **_kwargs):
+        raise RuntimeError("bad")
+
+    monkeypatch.setattr("builtins.open", _err)
+    out = doc.openapi_json()
+    assert cherrypy.response.status == 500
+    assert b"Error loading OpenAPI spec" in out
+
+
+def test_stats_app_index_and_default_routing(monkeypatch, tmp_path):
+    index_html = tmp_path / "index.html"
+    index_html.write_text("<html>ok</html>", encoding="utf-8")
+
+    fake_api = SimpleNamespace(config_manager=object(), docs=lambda: "d")
+    monkeypatch.setattr(hs, "APIEndpoints", lambda *args, **kwargs: fake_api)
+
+    app = hs.StatsApp(config={"web": {"web_path": str(tmp_path)}})
+
+    monkeypatch.setattr(cherrypy, "request", SimpleNamespace(method="GET"), raising=False)
+    assert app.index() == "<html>ok</html>"
+
+    monkeypatch.setattr(cherrypy, "request", SimpleNamespace(method="OPTIONS"), raising=False)
+    assert app.default("anything") == ""
+
+    monkeypatch.setattr(cherrypy, "request", SimpleNamespace(method="GET"), raising=False)
+    with pytest.raises(cherrypy.NotFound):
+        app.default("api")
+
+    assert app.default("ws", "packets") == ""
+    assert app.default("route") == "<html>ok</html>"
+
+
+def test_stats_app_index_error_paths(monkeypatch, tmp_path):
+    fake_api = SimpleNamespace(config_manager=object(), docs=lambda: "d")
+    monkeypatch.setattr(hs, "APIEndpoints", lambda *args, **kwargs: fake_api)
+
+    app = hs.StatsApp(config={"web": {"web_path": str(tmp_path)}})
+
+    with pytest.raises(cherrypy.HTTPError):
+        app.index()
+
+
+    # Force generic open() exception branch
+    def _explode(*_args, **_kwargs):
+        raise RuntimeError("boom")
+
+    monkeypatch.setattr("builtins.open", _explode)
+    (tmp_path / "index.html").write_text("ignored", encoding="utf-8")
+    with pytest.raises(cherrypy.HTTPError):
+        app.index()
+
+
+def test_http_server_utility_methods(monkeypatch, tmp_path):
+    def _fake_init_auth(self):
+        self.jwt_handler = object()
+        self.token_manager = object()
+
+    monkeypatch.setattr(hs.HTTPStatsServer, "_init_auth_handlers", _fake_init_auth)
+    monkeypatch.setattr(hs, "StatsApp", lambda *args, **kwargs: SimpleNamespace(api=SimpleNamespace(config_manager=object())))
+    monkeypatch.setattr(hs, "AuthEndpoints", lambda *args, **kwargs: object())
+    monkeypatch.setattr(hs, "DocEndpoint", lambda *_args, **_kwargs: object())
+
+    server = hs.HTTPStatsServer(config={"web": {"cors_enabled": False}}, config_path=str(Path(tmp_path) / "cfg.yml"))
+
+    monkeypatch.setattr(cherrypy, "response", SimpleNamespace(headers={}), raising=False)
+    out = server._json_error_handler(401, "no", "", "")
+    assert "\"success\": false" in out
+
+    install_called = {"v": False}
+    monkeypatch.setattr(hs.cherrypy_cors, "install", lambda: install_called.__setitem__("v", True))
+    server._setup_server_cors()
+    assert install_called["v"] is True
+
+    exited = {"v": False}
+    monkeypatch.setattr(cherrypy, "engine", SimpleNamespace(exit=lambda: exited.__setitem__("v", True)), raising=False)
+    server.stop()
+    assert exited["v"] is True

tests/test_main_py_more.py

@@ -0,0 +1,79 @@
+import asyncio
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from repeater.main import RepeaterDaemon
+
+
+class _FakeLocalIdentity:
+    def __init__(self, seed: bytes):
+        self._seed = seed
+
+    def get_public_key(self):
+        # Keep deterministic first-byte hash behavior.
+        return bytes([self._seed[0]]) + (b"P" * 31)
+
+    def get_address_bytes(self):
+        return b"\xAB\xCD"
+
+
+def _base_config():
+    return {
+        "repeater": {"node_name": "n1", "mode": "forward", "identity_key": b"k" * 32},
+        "logging": {"level": "INFO"},
+        "http": {"host": "127.0.0.1", "port": 8123},
+    }
+
+
+@pytest.mark.asyncio
+async def test_load_additional_identities_valid_and_invalid_entries():
+    cfg = _base_config()
+    cfg["identities"] = {
+        "room_servers": [
+            {},  # missing fields
+            {"name": "bad-hex", "identity_key": "zz-not-hex"},
+            {"name": "bad-len", "identity_key": "aa"},
+            {"name": "bad-type", "identity_key": 12345},
+            {"name": "good-bytes", "identity_key": b"\x10" * 32},
+            {"name": "good-hex", "identity_key": ("11" * 32)},
+        ]
+    }
+
+    daemon = RepeaterDaemon(cfg, radio=object())
+    daemon.identity_manager = SimpleNamespace(list_identities=lambda: [1, 2])
+    daemon._register_identity_everywhere = MagicMock(return_value=True)
+
+    with patch("pymc_core.LocalIdentity", _FakeLocalIdentity):
+        await daemon._load_additional_identities()
+
+    # Only the two valid entries should be registered.
+    assert daemon._register_identity_everywhere.call_count == 2
+    names = [c.kwargs["name"] for c in daemon._register_identity_everywhere.call_args_list]
+    assert names == ["good-bytes", "good-hex"]
+
+
+@pytest.mark.asyncio
+async def test_run_starts_http_and_handles_dispatcher_cancelled_gracefully():
+    daemon = RepeaterDaemon(_base_config(), radio=SimpleNamespace(cleanup=MagicMock()))
+
+    async def _init_stub():
+        daemon.local_identity = SimpleNamespace(get_public_key=lambda: b"\x22" * 32)
+        daemon.dispatcher = SimpleNamespace(run_forever=AsyncMock(side_effect=asyncio.CancelledError()))
+
+    daemon.initialize = _init_stub
+
+    fake_http_instance = SimpleNamespace(start=MagicMock(), stop=MagicMock())
+
+    fake_loop_for_signals = SimpleNamespace(add_signal_handler=MagicMock())
+
+    with (
+        patch("asyncio.get_running_loop", return_value=fake_loop_for_signals),
+        patch("repeater.main.HTTPStatsServer", return_value=fake_http_instance),
+        patch("os.path.exists", return_value=False),
+    ):
+        await daemon.run()
+
+    fake_http_instance.start.assert_called_once()
+    daemon.dispatcher.run_forever.assert_awaited_once()

tests/test_presets.py

@@ -320,6 +320,77 @@ def test_connect_failure_schedules_reconnect_with_actual_error_reason(monkeypatc
     assert captured["reason"] == "Not authorized (JWT signature/format invalid)"
 
 
+def test_schedule_reconnect_uses_exponential_backoff_and_cap(monkeypatch):
+    conn = _make_broker_connection("letsmesh")
+
+    captured = {"delay": None, "started": False}
+
+    class _Timer:
+        def __init__(self, delay, cb):
+            captured["delay"] = delay
+            self.daemon = False
+            self._cb = cb
+
+        def start(self):
+            captured["started"] = True
+
+        def cancel(self):
+            return None
+
+    monkeypatch.setattr("repeater.data_acquisition.mqtt_handler.threading.Timer", _Timer)
+
+    conn._reconnect_attempts = 0
+    conn._schedule_reconnect("first")
+    assert captured["delay"] == 5
+    assert captured["started"] is True
+
+    # Large attempt count should clamp to max delay.
+    conn._reconnect_attempts = 99
+    conn._schedule_reconnect("later")
+    assert captured["delay"] == conn._max_reconnect_delay
+
+
+def test_on_disconnect_duplicate_callback_does_not_schedule_reconnect(monkeypatch):
+    conn = _make_broker_connection("letsmesh")
+    conn._running = False
+
+    called = {"count": 0}
+
+    def _fake_schedule(reason="connection lost"):
+        called["count"] += 1
+
+    monkeypatch.setattr(conn, "_schedule_reconnect", _fake_schedule)
+
+    # Unexpected disconnect while already disconnected = duplicate callback.
+    conn._on_disconnect(client=None, userdata=None, rc=1)
+    assert called["count"] == 0
+
+
+def test_attempt_reconnect_failure_reschedules(monkeypatch):
+    conn = _make_broker_connection("letsmesh")
+    conn._running = False
+    conn._reconnect_timer = object()
+
+    monkeypatch.setattr(conn, "_set_credentials", lambda: None)
+    monkeypatch.setattr(conn.client, "loop_stop", lambda: None)
+    monkeypatch.setattr(conn.client, "loop_start", lambda: None)
+
+    def _boom_connect(*args, **kwargs):
+        raise RuntimeError("connect failed")
+
+    monkeypatch.setattr(conn.client, "connect", _boom_connect)
+
+    called = {"count": 0}
+
+    def _fake_schedule(reason="connection lost"):
+        called["count"] += 1
+
+    monkeypatch.setattr(conn, "_schedule_reconnect", _fake_schedule)
+
+    conn._attempt_reconnect("network")
+    assert called["count"] == 1
+
+
 def test_on_pre_connect_refreshes_jwt_credentials(monkeypatch):
     """JWT credentials should be refreshed on each (re)connect attempt."""
     conn = _make_broker_connection("letsmesh")

tests/test_service_utils.py

@@ -0,0 +1,224 @@
+import io
+import subprocess
+from unittest.mock import MagicMock
+
+import pytest
+
+from repeater import service_utils as su
+
+
+def test_is_buildroot_via_metadata_file(monkeypatch):
+    monkeypatch.setattr(su.os.path, "exists", lambda p: p == su.BUILDROOT_METADATA_PATH)
+    assert su.is_buildroot() is True
+
+
+def test_is_buildroot_via_os_release(monkeypatch):
+    monkeypatch.setattr(
+        su.os.path,
+        "exists",
+        lambda p: p == "/etc/os-release",
+    )
+    monkeypatch.setattr(
+        "builtins.open",
+        lambda *args, **kwargs: io.StringIO("NAME=x\nID=buildroot\n"),
+    )
+    assert su.is_buildroot() is True
+
+
+def test_get_buildroot_image_info_parse_and_error(monkeypatch):
+    monkeypatch.setattr(
+        "builtins.open",
+        lambda *args, **kwargs: io.StringIO("\nfoo=bar\ninvalid\nimage_version=1.2.3\n"),
+    )
+    info = su.get_buildroot_image_info()
+    assert info["foo"] == "bar"
+    assert info["image_version"] == "1.2.3"
+    assert su.get_buildroot_image_version() == "1.2.3"
+
+    def _raise(*_args, **_kwargs):
+        raise OSError("nope")
+
+    monkeypatch.setattr("builtins.open", _raise)
+    assert su.get_buildroot_image_info() == {}
+
+
+def test_is_container_detection_paths(monkeypatch):
+    # /.dockerenv path
+    monkeypatch.setattr(su.os.path, "exists", lambda p: p == "/.dockerenv")
+    monkeypatch.delenv("container", raising=False)
+    assert su.is_container() is True
+
+    # env var path
+    monkeypatch.setattr(su.os.path, "exists", lambda _p: False)
+    monkeypatch.setenv("container", "docker")
+    assert su.is_container() is True
+
+
+@pytest.mark.parametrize(
+    "environ_bytes,cgroup_text,host_path,expected",
+    [
+        (b"abc\x00container=docker\x00", "", False, True),
+        (b"abc", "1:name=systemd:/docker/abc", False, True),
+        (b"abc", "1:name=systemd:/", True, True),
+        (b"abc", "1:name=systemd:/", False, False),
+    ],
+)
+def test_is_container_proc_and_host_paths(monkeypatch, environ_bytes, cgroup_text, host_path, expected):
+    monkeypatch.setattr(su.os.path, "exists", lambda p: p == "/run/host/container-manager" and host_path)
+    monkeypatch.delenv("container", raising=False)
+
+    def _open(path, mode="r", encoding=None):
+        if path == "/proc/1/environ":
+            return io.BytesIO(environ_bytes)
+        if path == "/proc/1/cgroup":
+            return io.StringIO(cgroup_text)
+        raise OSError("unexpected")
+
+    monkeypatch.setattr("builtins.open", _open)
+    assert su.is_container() is expected
+
+
+def test_get_container_restart_message():
+    msg = su.get_container_restart_message()
+    assert "Container restart initiated" in msg
+    assert "Docker or Home Assistant" in msg
+
+
+def test_restart_service_container_path(monkeypatch):
+    monkeypatch.setattr(su, "is_container", lambda: True)
+    sched = MagicMock()
+    monkeypatch.setattr(su, "_schedule_container_exit", sched)
+
+    ok, msg = su.restart_service()
+    assert ok is True
+    assert "Container restart initiated" in msg
+    sched.assert_called_once()
+
+
+def test_restart_service_buildroot_paths(monkeypatch):
+    monkeypatch.setattr(su, "is_container", lambda: False)
+    monkeypatch.setattr(su, "is_buildroot", lambda: True)
+
+    # missing init script
+    monkeypatch.setattr(su.os.path, "exists", lambda _p: False)
+    ok, msg = su.restart_service()
+    assert ok is False
+    assert "init script not found" in msg
+
+    # popen success
+    monkeypatch.setattr(su.os.path, "exists", lambda p: p == su.INIT_SCRIPT)
+    monkeypatch.setattr(su.subprocess, "Popen", MagicMock())
+    ok, msg = su.restart_service()
+    assert ok is True
+    assert "Service restart initiated" in msg
+
+    # popen failure
+    def _raise(*_args, **_kwargs):
+        raise RuntimeError("boom")
+
+    monkeypatch.setattr(su.subprocess, "Popen", _raise)
+    ok, msg = su.restart_service()
+    assert ok is False
+    assert "Restart failed" in msg
+
+
+def test_restart_service_systemctl_and_sudo_paths(monkeypatch):
+    monkeypatch.setattr(su, "is_container", lambda: False)
+    monkeypatch.setattr(su, "is_buildroot", lambda: False)
+
+    def _result(code=0, err=""):
+        return subprocess.CompletedProcess(args=[], returncode=code, stdout="", stderr=err)
+
+    # systemctl success
+    monkeypatch.setattr(su.subprocess, "run", lambda *args, **kwargs: _result(0))
+    ok, msg = su.restart_service()
+    assert ok is True
+    assert "Service restart initiated" in msg
+
+    # systemctl timeout
+    def _timeout(*_args, **_kwargs):
+        raise subprocess.TimeoutExpired(cmd="x", timeout=5)
+
+    monkeypatch.setattr(su.subprocess, "run", _timeout)
+    ok, msg = su.restart_service()
+    assert ok is True
+    assert "timeout" in msg
+
+    # systemctl missing binary
+    def _missing(*_args, **_kwargs):
+        raise FileNotFoundError("systemctl")
+
+    monkeypatch.setattr(su.subprocess, "run", _missing)
+    ok, msg = su.restart_service()
+    assert ok is False
+    assert "systemctl not available" in msg
+
+    # systemctl denied then sudo success
+    calls = {"n": 0}
+
+    def _denied_then_sudo(*args, **kwargs):
+        calls["n"] += 1
+        if calls["n"] == 1:
+            return _result(1, "Access denied")
+        return _result(0)
+
+    monkeypatch.setattr(su.subprocess, "run", _denied_then_sudo)
+    ok, msg = su.restart_service()
+    assert ok is True
+    assert "Service restart initiated" in msg
+
+    # systemctl generic fail then sudo fail
+    calls = {"n": 0}
+
+    def _fail_then_fail(*args, **kwargs):
+        calls["n"] += 1
+        if calls["n"] == 1:
+            return _result(1, "broken")
+        return _result(2, "sudo denied")
+
+    monkeypatch.setattr(su.subprocess, "run", _fail_then_fail)
+    ok, msg = su.restart_service()
+    assert ok is False
+    assert "Restart failed" in msg
+
+    # systemctl generic fail then sudo timeout
+    calls = {"n": 0}
+
+    def _fail_then_timeout(*args, **kwargs):
+        calls["n"] += 1
+        if calls["n"] == 1:
+            return _result(1, "broken")
+        raise subprocess.TimeoutExpired(cmd="sudo", timeout=5)
+
+    monkeypatch.setattr(su.subprocess, "run", _fail_then_timeout)
+    ok, msg = su.restart_service()
+    assert ok is True
+    assert "timeout" in msg
+
+    # systemctl generic fail then sudo missing
+    calls = {"n": 0}
+
+    def _fail_then_sudo_missing(*args, **kwargs):
+        calls["n"] += 1
+        if calls["n"] == 1:
+            return _result(1, "broken")
+        raise FileNotFoundError("sudo")
+
+    monkeypatch.setattr(su.subprocess, "run", _fail_then_sudo_missing)
+    ok, msg = su.restart_service()
+    assert ok is False
+    assert "Neither polkit nor sudo" in msg
+
+    # systemctl generic fail then sudo unexpected exception
+    calls = {"n": 0}
+
+    def _fail_then_exception(*args, **kwargs):
+        calls["n"] += 1
+        if calls["n"] == 1:
+            return _result(1, "broken")
+        raise RuntimeError("bad")
+
+    monkeypatch.setattr(su.subprocess, "run", _fail_then_exception)
+    ok, msg = su.restart_service()
+    assert ok is False
+    assert "Restart command failed" in msg

tests/test_sqlite_handler_easy.py

@@ -0,0 +1,305 @@
+from pathlib import Path
+
+import pytest
+
+from repeater.data_acquisition.sqlite_handler import SQLiteHandler
+
+
+def _make_handler(tmp_path: Path) -> SQLiteHandler:
+    return SQLiteHandler(tmp_path)
+
+
+def test_api_token_crud_cycle(tmp_path):
+    h = _make_handler(tmp_path)
+
+    token_id = h.create_api_token("svc-a", "hash-a")
+    assert isinstance(token_id, int)
+
+    verified = h.verify_api_token("hash-a")
+    assert verified is not None
+    assert verified["id"] == token_id
+    assert verified["name"] == "svc-a"
+
+    listed = h.list_api_tokens()
+    assert any(t["id"] == token_id for t in listed)
+
+    assert h.revoke_api_token(token_id) is True
+    assert h.verify_api_token("hash-a") is None
+    assert h.revoke_api_token(token_id) is False
+
+
+def test_transport_key_crud_cycle(tmp_path):
+    h = _make_handler(tmp_path)
+
+    key_id = h.create_transport_key(
+        name="root",
+        flood_policy="allow",
+        transport_key="dGVzdC1rZXk=",  # base64('test-key')
+    )
+    assert isinstance(key_id, int)
+
+    row = h.get_transport_key_by_id(key_id)
+    assert row is not None
+    assert row["name"] == "root"
+    assert row["flood_policy"] == "allow"
+
+    # No fields to update returns False by design.
+    assert h.update_transport_key(key_id) is False
+
+    assert h.update_transport_key(key_id, name="child", flood_policy="deny") is True
+    row2 = h.get_transport_key_by_id(key_id)
+    assert row2 is not None
+    assert row2["name"] == "child"
+    assert row2["flood_policy"] == "deny"
+
+    all_rows = h.get_transport_keys()
+    assert len(all_rows) == 1
+    assert all_rows[0]["id"] == key_id
+
+    assert h.delete_transport_key(key_id) is True
+    assert h.get_transport_key_by_id(key_id) is None
+    assert h.delete_transport_key(key_id) is False
+
+
+def test_room_messages_and_sync_flow(tmp_path):
+    h = _make_handler(tmp_path)
+
+    room_hash = "0x42"
+    a_pub = "a" * 64
+    b_pub = "b" * 64
+
+    m1 = h.insert_room_message(room_hash, a_pub, "hello", post_timestamp=100.0)
+    m2 = h.insert_room_message(room_hash, b_pub, "world", post_timestamp=200.0)
+    assert isinstance(m1, int)
+    assert isinstance(m2, int)
+
+    assert h.get_room_message_count(room_hash) == 2
+
+    # get_room_messages sorts by post_timestamp DESC
+    msgs = h.get_room_messages(room_hash, limit=10, offset=0)
+    assert len(msgs) == 2
+    assert msgs[0]["message_text"] == "world"
+
+    since = h.get_messages_since(room_hash, since_timestamp=150.0, limit=10)
+    assert len(since) == 1
+    assert since[0]["message_text"] == "world"
+
+    unsynced_for_a = h.get_unsynced_messages(room_hash, client_pubkey=a_pub, sync_since=0.0)
+    assert len(unsynced_for_a) == 1
+    assert unsynced_for_a[0]["author_pubkey"] == b_pub
+
+    assert h.get_unsynced_count(room_hash, client_pubkey=a_pub, sync_since=0.0) == 1
+
+    # Client sync upsert/get/list
+    assert h.upsert_client_sync(room_hash, a_pub, sync_since=50.0, last_activity=123.0) is True
+    sync = h.get_client_sync(room_hash, a_pub)
+    assert sync is not None
+    assert sync["sync_since"] == 50.0
+
+    clients = h.get_all_room_clients(room_hash)
+    assert len(clients) == 1
+    assert clients[0]["client_pubkey"] == a_pub
+
+    assert h.delete_room_message(room_hash, int(m1)) is True
+    assert h.delete_room_message(room_hash, int(m1)) is False
+
+    deleted = h.clear_room_messages(room_hash)
+    assert deleted == 1
+    assert h.get_room_message_count(room_hash) == 0
+
+
+def test_store_and_delete_advert(tmp_path):
+    h = _make_handler(tmp_path)
+
+    h.store_advert(
+        {
+            "timestamp": 123.0,
+            "pubkey": "pk1",
+            "node_name": "node-1",
+            "is_repeater": True,
+            "route_type": 1,
+            "contact_type": "neighbor",
+            "latitude": 1.0,
+            "longitude": 2.0,
+            "rssi": -88,
+            "snr": 7.5,
+            "is_new_neighbor": True,
+            "zero_hop": True,
+        }
+    )
+
+    with h._connect() as conn:
+        row = conn.execute("SELECT id FROM adverts WHERE pubkey = ?", ("pk1",)).fetchone()
+    assert row is not None
+    advert_id = int(row[0])
+
+    assert h.delete_advert(advert_id) is True
+    assert h.delete_advert(advert_id) is False
+
+
+def test_verify_api_token_last_used_throttle(tmp_path, monkeypatch):
+    h = _make_handler(tmp_path)
+    h._api_token_last_used_interval_sec = 300
+
+    now = {"v": 1000.0}
+
+    monkeypatch.setattr(
+        "repeater.data_acquisition.sqlite_handler.time.time", lambda: now["v"]
+    )
+
+    token_id = h.create_api_token("svc-throttle", "hash-throttle")
+    assert token_id > 0
+
+    assert h.verify_api_token("hash-throttle") is not None
+    with h._connect() as conn:
+        first = conn.execute(
+            "SELECT last_used FROM api_tokens WHERE id = ?", (token_id,)
+        ).fetchone()[0]
+    assert first == 1000.0
+
+    now["v"] = 1010.0
+    assert h.verify_api_token("hash-throttle") is not None
+    with h._connect() as conn:
+        second = conn.execute(
+            "SELECT last_used FROM api_tokens WHERE id = ?", (token_id,)
+        ).fetchone()[0]
+    assert second == 1000.0
+
+    now["v"] = 1401.0
+    assert h.verify_api_token("hash-throttle") is not None
+    with h._connect() as conn:
+        third = conn.execute(
+            "SELECT last_used FROM api_tokens WHERE id = ?", (token_id,)
+        ).fetchone()[0]
+    assert third == 1401.0
+
+
+def test_store_advert_zero_hop_signal_handling(tmp_path):
+    h = _make_handler(tmp_path)
+
+    h.store_advert(
+        {
+            "timestamp": 10.0,
+            "pubkey": "pk-z",
+            "node_name": "node-z",
+            "is_repeater": False,
+            "route_type": 1,
+            "contact_type": "neighbor",
+            "rssi": -80,
+            "snr": 5.0,
+            "is_new_neighbor": True,
+            "zero_hop": True,
+        }
+    )
+
+    # Multi-hop update must preserve previous zero-hop signal quality.
+    h.store_advert(
+        {
+            "timestamp": 20.0,
+            "pubkey": "pk-z",
+            "node_name": "node-z-2",
+            "is_repeater": False,
+            "route_type": 2,
+            "contact_type": "neighbor",
+            "rssi": -50,
+            "snr": 9.0,
+            "is_new_neighbor": False,
+            "zero_hop": False,
+        }
+    )
+    with h._connect() as conn:
+        row = conn.execute(
+            "SELECT rssi, snr, zero_hop, advert_count FROM adverts WHERE pubkey = ?",
+            ("pk-z",),
+        ).fetchone()
+    assert row is not None
+    assert row[0] == -80
+    assert row[1] == 5.0
+    assert bool(row[2]) is True
+    assert row[3] == 2
+
+    # New zero-hop update should refresh signal quality.
+    h.store_advert(
+        {
+            "timestamp": 30.0,
+            "pubkey": "pk-z",
+            "node_name": "node-z-3",
+            "is_repeater": False,
+            "route_type": 1,
+            "contact_type": "neighbor",
+            "rssi": -60,
+            "snr": 6.5,
+            "is_new_neighbor": False,
+            "zero_hop": True,
+        }
+    )
+    with h._connect() as conn:
+        row2 = conn.execute(
+            "SELECT rssi, snr, zero_hop, advert_count FROM adverts WHERE pubkey = ?",
+            ("pk-z",),
+        ).fetchone()
+    assert row2 is not None
+    assert row2[0] == -60
+    assert row2[1] == 6.5
+    assert bool(row2[2]) is True
+    assert row2[3] == 3
+
+
+def test_sync_transport_keys_validation_and_tree_apply(tmp_path, monkeypatch):
+    h = _make_handler(tmp_path)
+
+    with pytest.raises(ValueError, match="must be a list"):
+        h.sync_transport_keys({"bad": True})
+
+    with pytest.raises(ValueError, match="Duplicate node_id"):
+        h.sync_transport_keys(
+            [
+                {"node_id": "1", "name": "a", "flood_policy": "allow"},
+                {"node_id": "1", "name": "b", "flood_policy": "deny"},
+            ]
+        )
+
+    with pytest.raises(ValueError, match="Parent node 'missing'"):
+        h.sync_transport_keys(
+            [
+                {
+                    "node_id": "c1",
+                    "name": "child",
+                    "flood_policy": "allow",
+                    "parent_node_id": "missing",
+                }
+            ]
+        )
+
+    monkeypatch.setattr(h, "generate_transport_key", lambda _name: "GEN-KEY")
+    applied = h.sync_transport_keys(
+        [
+            {
+                "node_id": "root",
+                "name": "root-name",
+                "flood_policy": "allow",
+                "transport_key": "ROOT-KEY",
+            },
+            {
+                "node_id": "child",
+                "name": "child-name",
+                "flood_policy": "deny",
+                "parent_node_id": "root",
+                "transport_key": None,
+            },
+        ]
+    )
+
+    assert applied == {"applied_nodes": 2, "generated_keys": 1}
+
+    with h._connect() as conn:
+        rows = conn.execute(
+            "SELECT id, name, flood_policy, transport_key, parent_id FROM transport_keys ORDER BY id"
+        ).fetchall()
+    assert len(rows) == 2
+    assert rows[0][1] == "root-name"
+    assert rows[0][3] == "ROOT-KEY"
+    assert rows[1][1] == "child-name"
+    assert rows[1][2] == "deny"
+    assert rows[1][3] == "GEN-KEY"
+    assert rows[1][4] == rows[0][0]

tests/test_update_endpoints_unit.py

@@ -0,0 +1,454 @@
+from datetime import datetime, timedelta, timezone
+from types import SimpleNamespace
+from unittest.mock import patch
+
+import cherrypy
+import pytest
+
+import repeater.web.update_endpoints as ue
+
+
+@pytest.fixture
+def cherrypy_ctx(monkeypatch):
+    request = SimpleNamespace(method="GET", json={}, params={})
+    response = SimpleNamespace(headers={}, status=200)
+    monkeypatch.setattr(cherrypy, "request", request, raising=False)
+    monkeypatch.setattr(cherrypy, "response", response, raising=False)
+    return request, response
+
+
+@pytest.fixture
+def isolated_state(monkeypatch, tmp_path):
+    channel_file = tmp_path / "update_channel"
+    monkeypatch.setattr(ue, "_CHANNELS_FILE", str(channel_file), raising=False)
+    monkeypatch.setattr(ue, "_detect_channel_from_dist_info", lambda: None)
+    monkeypatch.setattr(ue, "_get_installed_version", lambda force_refresh=False: "1.0.0")
+    st = ue._UpdateState()
+    monkeypatch.setattr(ue, "_state", st, raising=False)
+    return st
+
+
+def _fake_thread(*args, **kwargs):
+    return SimpleNamespace(start=lambda: None, name=kwargs.get("name", "t"))
+
+
+def test_jwt_warning_fix_guard():
+    # Guard test file import path and ensure this module executes in suite.
+    assert ue.PACKAGE_NAME == "pymc_repeater"
+
+
+def test_has_update_paths():
+    assert ue._has_update("1.2.3", "1.2.3") is False
+    assert ue._has_update("1.2.3", "1.2.4") is True
+    assert ue._has_update("1.2.4.dev10", "1.2.4.dev12") is True
+
+
+def test_fetch_url_success_and_rate_limit(monkeypatch):
+    class _Resp:
+        def __enter__(self):
+            return self
+
+        def __exit__(self, *_):
+            return False
+
+        def read(self):
+            return b"ok"
+
+    monkeypatch.setattr(ue.urllib.request, "urlopen", lambda *args, **kwargs: _Resp())
+    assert ue._fetch_url("https://example.com") == "ok"
+
+    reset = int((datetime.now(timezone.utc) + timedelta(minutes=5)).timestamp())
+    hdrs = {"X-RateLimit-Reset": str(reset)}
+
+    def _raise(*args, **kwargs):
+        raise ue.urllib.error.HTTPError("u", 403, "forbidden", hdrs, None)
+
+    monkeypatch.setattr(ue.urllib.request, "urlopen", _raise)
+    with pytest.raises(ue._RateLimitError) as exc:
+        ue._fetch_url("https://api.github.com/test")
+    assert "rate limit" in str(exc.value).lower()
+
+
+def test_update_state_snapshot_and_mutators(isolated_state, monkeypatch):
+    st = isolated_state
+    monkeypatch.setattr(ue, "_get_installed_version", lambda force_refresh=False: "1.0.1")
+
+    st.latest_version = "1.0.2"
+    snap = st.snapshot()
+    assert snap["current_version"] == "1.0.1"
+    assert snap["has_update"] is True
+
+    st.set_channel("dev")
+    assert st.channel == "dev"
+    assert st.latest_version is None
+    assert st.has_update is False
+
+    assert st._set_checking() is True
+    assert st._set_checking() is False
+
+    t = _fake_thread(name="install")
+    assert st.start_install(t) is True
+    assert st.start_install(t) is False
+
+    st.finish_install(True, "done")
+    assert st.state == "complete"
+
+
+def test_update_state_append_line_trim(isolated_state):
+    st = isolated_state
+    for i in range(510):
+        st.append_line(f"l-{i}")
+    assert len(st.progress_lines) == 500
+    assert st.progress_lines[0].startswith("l-")
+
+
+def test_status_endpoint_options_and_ok(cherrypy_ctx, isolated_state):
+    request, _ = cherrypy_ctx
+    api = ue.UpdateAPIEndpoints()
+
+    request.method = "OPTIONS"
+    assert api.status() == ""
+
+    request.method = "GET"
+    out = api.status()
+    assert out["success"] is True
+    assert out["current_version"] == "1.0.0"
+
+
+def test_check_endpoint_paths(cherrypy_ctx, isolated_state, monkeypatch):
+    request, _ = cherrypy_ctx
+    api = ue.UpdateAPIEndpoints()
+
+    request.method = "OPTIONS"
+    assert api.check() == ""
+
+    request.method = "PUT"
+    with pytest.raises(cherrypy.HTTPError):
+        api.check()
+
+    request.method = "GET"
+    isolated_state.state = "checking"
+    busy = api.check()
+    assert busy["success"] is True
+    assert busy["state"] == "checking"
+
+    isolated_state.state = "idle"
+    isolated_state.latest_version = "1.0.2"
+    isolated_state.last_checked = datetime.now(timezone.utc)
+    cached = api.check()
+    assert cached["success"] is True
+    assert "cached" in cached["message"].lower()
+
+    isolated_state.last_checked = None
+    isolated_state.latest_version = None
+    request.method = "POST"
+    request.json = {"force": True}
+    monkeypatch.setattr(ue.threading, "Thread", _fake_thread)
+    started = api.check()
+    assert started["success"] is True
+    assert started["state"] == "checking"
+
+
+def test_check_endpoint_rate_limit_window(cherrypy_ctx, isolated_state):
+    request, _ = cherrypy_ctx
+    api = ue.UpdateAPIEndpoints()
+    request.method = "POST"
+    request.json = {}
+
+    isolated_state.rate_limit_until = datetime.now(timezone.utc) + timedelta(minutes=1)
+    out = api.check()
+    assert out["success"] is True
+    assert "rate limit" in out["message"].lower()
+
+
+def test_install_endpoint_paths(cherrypy_ctx, isolated_state, monkeypatch):
+    request, response = cherrypy_ctx
+    api = ue.UpdateAPIEndpoints()
+
+    request.method = "GET"
+    with pytest.raises(cherrypy.HTTPError):
+        api.install()
+
+    request.method = "POST"
+    request.json = {}
+    isolated_state.state = "installing"
+    out = api.install()
+    assert out["success"] is False
+    assert response.status == 409
+
+    isolated_state.state = "idle"
+    isolated_state.latest_version = "1.0.0"
+    isolated_state.has_update = False
+    up_to_date = api.install()
+    assert up_to_date["success"] is False
+    assert response.status == 409
+
+    request.json = {"force": True}
+    monkeypatch.setattr(ue.threading, "Thread", _fake_thread)
+    isolated_state.state = "idle"
+    isolated_state.latest_version = None
+    isolated_state.has_update = False
+    ok = api.install()
+    assert ok["success"] is True
+    assert ok["state"] == "installing"
+
+
+def test_progress_endpoint_stream(cherrypy_ctx, isolated_state):
+    _, response = cherrypy_ctx
+    api = ue.UpdateAPIEndpoints()
+
+    isolated_state.state = "complete"
+    isolated_state.progress_lines = ["line-1"]
+
+    stream = api.progress()
+    chunks = list(stream)
+
+    assert response.headers["Content-Type"] == "text/event-stream"
+    joined = "".join(chunks)
+    assert "connected" in joined
+    assert "line-1" in joined
+    assert "done" in joined
+
+
+def test_channels_set_channel_and_changelog(cherrypy_ctx, isolated_state, monkeypatch):
+    request, response = cherrypy_ctx
+    api = ue.UpdateAPIEndpoints()
+
+    request.method = "OPTIONS"
+    assert api.channels() == ""
+    assert api.set_channel() == ""
+    assert api.changelog() == ""
+
+    request.method = "GET"
+    monkeypatch.setattr(ue, "_fetch_branches", lambda: ["main", "dev"])
+    ch = api.channels()
+    assert ch["success"] is True
+    assert ch["channels"][0] == "main"
+
+    request.method = "POST"
+    request.json = {}
+    bad = api.set_channel()
+    assert bad["success"] is False
+    assert response.status == 400
+
+    request.json = {"channel": "dev"}
+    isolated_state.state = "installing"
+    blocked = api.set_channel()
+    assert blocked["success"] is False
+    assert response.status == 409
+
+    isolated_state.state = "idle"
+    ok = api.set_channel()
+    assert ok["success"] is True
+    assert ok["channel"] == "dev"
+
+    request.method = "GET"
+    monkeypatch.setattr(ue, "_fetch_changelog", lambda channel, installed, max_commits: [{"title": "t"}])
+    c = api.changelog(channel="dev", max="5")
+    assert c["success"] is True
+    assert c["commits"][0]["title"] == "t"
+
+
+def test_cors_headers_and_error_helpers(cherrypy_ctx):
+    _, response = cherrypy_ctx
+    api = ue.UpdateAPIEndpoints()
+
+    api._set_cors_headers({"web": {"cors_enabled": True}})
+    assert response.headers["Access-Control-Allow-Origin"] == "*"
+
+    response.status = 200
+    err = api._err("nope", status=418)
+    assert err["success"] is False
+    assert response.status == 418
+
+
+def test_do_check_success_rate_limit_and_generic_error(isolated_state, monkeypatch):
+    st = isolated_state
+
+    monkeypatch.setattr(ue, "_fetch_latest_version", lambda _channel: "1.0.2")
+    ue._do_check()
+    assert st.latest_version == "1.0.2"
+    assert st.state == "idle"
+    assert st.has_update is True
+
+    reset_at = datetime.now(timezone.utc) + timedelta(minutes=2)
+    monkeypatch.setattr(
+        ue,
+        "_fetch_latest_version",
+        lambda _channel: (_ for _ in ()).throw(ue._RateLimitError("limited", reset_at=reset_at)),
+    )
+    ue._do_check()
+    assert st.state == "idle"
+    assert st.rate_limit_until == reset_at
+    assert "limited" in (st.error_message or "")
+
+    monkeypatch.setattr(
+        ue,
+        "_fetch_latest_version",
+        lambda _channel: (_ for _ in ()).throw(RuntimeError("boom")),
+    )
+    ue._do_check()
+    assert st.state == "error"
+    assert "boom" in (st.error_message or "")
+
+
+def test_fetch_branches_priority_and_fallback(monkeypatch):
+    monkeypatch.setattr(
+        ue,
+        "_fetch_url",
+        lambda _url, timeout=8: '[{"name":"feature"},{"name":"dev"},{"name":"main"}]',
+    )
+    out = ue._fetch_branches()
+    assert out[:2] == ["main", "dev"]
+    assert "feature" in out
+
+    monkeypatch.setattr(
+        ue,
+        "_fetch_url",
+        lambda _url, timeout=8: (_ for _ in ()).throw(RuntimeError("net down")),
+    )
+    out2 = ue._fetch_branches()
+    assert out2 == ["main"]
+
+
+def test_fetch_latest_version_dynamic_and_static(monkeypatch):
+    monkeypatch.setattr(ue, "_get_latest_tag", lambda: "1.0.5")
+
+    # Dynamic branch path uses compare ahead_by -> next dev version.
+    monkeypatch.setattr(ue, "_branch_is_dynamic", lambda _ch: True)
+    monkeypatch.setattr(ue, "_fetch_url", lambda _url, timeout=10: '{"ahead_by": 3}')
+    dyn = ue._fetch_latest_version("dev")
+    assert dyn == "1.0.6.dev3"
+
+    # Static branch path parses version from pyproject content.
+    monkeypatch.setattr(ue, "_branch_is_dynamic", lambda _ch: False)
+    monkeypatch.setattr(
+        ue,
+        "_fetch_url",
+        lambda _url, timeout=8: 'name = "x"\nversion = "2.3.4"\n',
+    )
+    stat = ue._fetch_latest_version("main")
+    assert stat == "2.3.4"
+
+
+def test_do_install_non_root_wrapper_missing_finishes_error(isolated_state, monkeypatch):
+    st = isolated_state
+    st.channel = "main"
+    st.latest_version = "1.2.3"
+
+    monkeypatch.setattr(ue.os, "geteuid", lambda: 1000)
+    monkeypatch.setattr(ue, "is_buildroot", lambda: False)
+    monkeypatch.setattr(ue.os.path, "isfile", lambda p: False)
+
+    ue._do_install()
+
+    assert st.state == "error"
+    assert "Upgrade wrapper not found" in (st.error_message or "")
+
+
+def test_do_install_root_buildroot_helper_missing(isolated_state, monkeypatch):
+    st = isolated_state
+    st.channel = "dev"
+    st.latest_version = "2.0.0"
+
+    monkeypatch.setattr(ue.os, "geteuid", lambda: 0)
+    monkeypatch.setattr(ue, "is_buildroot", lambda: True)
+    monkeypatch.setattr(ue, "_find_buildroot_upgrade_helper", lambda: None)
+
+    ue._do_install()
+
+    assert st.state == "error"
+    assert "Buildroot upgrade helper not found" in (st.error_message or "")
+
+
+def test_do_install_root_install_command_failure_sets_error(isolated_state, monkeypatch):
+    st = isolated_state
+    st.channel = "main"
+    st.latest_version = "3.1.4"
+
+    monkeypatch.setattr(ue.os, "geteuid", lambda: 0)
+    monkeypatch.setattr(ue, "is_buildroot", lambda: False)
+    monkeypatch.setattr(ue, "_migrate_service_unit", lambda: None)
+    monkeypatch.setattr(ue.os.path, "isfile", lambda p: True)
+    monkeypatch.setattr(ue.os.path, "isdir", lambda p: False)
+
+    class _Proc:
+        def __init__(self, cmd):
+            self.cmd = cmd
+            self.stdout = []
+            self.returncode = (
+                1
+                if any(isinstance(x, str) and "git+https://github.com" in x for x in cmd)
+                else 0
+            )
+
+        def wait(self):
+            return None
+
+    monkeypatch.setattr(ue.subprocess, "Popen", lambda cmd, **kwargs: _Proc(cmd))
+
+    ue._do_install()
+
+    assert st.state == "error"
+    assert "pip install failed" in (st.error_message or "")
+
+
+def test_do_install_wrapper_success_then_restart_failure(isolated_state, monkeypatch):
+    st = isolated_state
+    st.channel = "main"
+    st.latest_version = "4.0.0"
+
+    monkeypatch.setattr(ue.os, "geteuid", lambda: 1000)
+    monkeypatch.setattr(ue, "is_buildroot", lambda: False)
+    monkeypatch.setattr(ue, "_cleanup_stale_dist_info", lambda *args, **kwargs: None)
+    monkeypatch.setattr(ue.time, "sleep", lambda _s: None)
+    monkeypatch.setattr(ue.os.path, "isfile", lambda p: p == "/usr/local/bin/pymc-do-upgrade")
+    monkeypatch.setattr("repeater.service_utils.restart_service", lambda: (False, "systemctl failed"))
+
+    class _Proc:
+        def __init__(self, cmd):
+            self.cmd = cmd
+            self.stdout = ["ok\n"]
+            self.returncode = 0
+
+        def wait(self):
+            return None
+
+    monkeypatch.setattr(ue.subprocess, "Popen", lambda cmd, **kwargs: _Proc(cmd))
+
+    ue._do_install()
+
+    assert st.state == "error"
+    assert "restart failed" in (st.error_message or "")
+
+
+def test_do_install_wrapper_success_container_path(isolated_state, monkeypatch):
+    st = isolated_state
+    st.channel = "main"
+    st.latest_version = "5.0.0"
+
+    monkeypatch.setattr(ue.os, "geteuid", lambda: 1000)
+    monkeypatch.setattr(ue, "is_buildroot", lambda: False)
+    monkeypatch.setattr(ue, "is_container", lambda: True)
+    monkeypatch.setattr(ue, "get_container_restart_message", lambda: "container will restart")
+    monkeypatch.setattr(ue, "_cleanup_stale_dist_info", lambda *args, **kwargs: None)
+    monkeypatch.setattr(ue.time, "sleep", lambda _s: None)
+    monkeypatch.setattr(ue.os.path, "isfile", lambda p: p == "/usr/local/bin/pymc-do-upgrade")
+    monkeypatch.setattr("repeater.service_utils.restart_service", lambda: (True, "ok"))
+
+    class _Proc:
+        def __init__(self, cmd):
+            self.cmd = cmd
+            self.stdout = ["ok\n"]
+            self.returncode = 0
+
+        def wait(self):
+            return None
+
+    monkeypatch.setattr(ue.subprocess, "Popen", lambda cmd, **kwargs: _Proc(cmd))
+
+    ue._do_install()
+
+    assert st.state == "complete"
+    assert st.error_message is None
+    assert any("container will restart" in line for line in st.progress_lines)