From 7b86716e06d1969b8c50186b0566d87c02660a83 Mon Sep 17 00:00:00 2001 From: Lloyd Date: Tue, 26 May 2026 14:59:31 +0100 Subject: [PATCH] Add unit tests for HTTP server, main daemon, service utilities, SQLite handler, and update endpoints --- repeater/web/update_endpoints.py | 14 +- tests/test_api_endpoints_core_coverage.py | 647 +++++++++++++++++++++ tests/test_auth_components.py | 127 ++++ tests/test_auth_endpoints.py | 363 ++++++++++++ tests/test_companion_bridge_frame_utils.py | 208 +++++++ tests/test_companion_ws_proxy.py | 172 ++++++ tests/test_handler_helpers_mesh_cli.py | 209 +++++++ tests/test_http_server_unit.py | 129 ++++ tests/test_main_py_more.py | 79 +++ tests/test_presets.py | 71 +++ tests/test_service_utils.py | 224 +++++++ tests/test_sqlite_handler_easy.py | 305 ++++++++++ tests/test_update_endpoints_unit.py | 454 +++++++++++++++ 13 files changed, 2995 insertions(+), 7 deletions(-) create mode 100644 tests/test_auth_components.py create mode 100644 tests/test_auth_endpoints.py create mode 100644 tests/test_companion_bridge_frame_utils.py create mode 100644 tests/test_companion_ws_proxy.py create mode 100644 tests/test_handler_helpers_mesh_cli.py create mode 100644 tests/test_http_server_unit.py create mode 100644 tests/test_main_py_more.py create mode 100644 tests/test_service_utils.py create mode 100644 tests/test_sqlite_handler_easy.py create mode 100644 tests/test_update_endpoints_unit.py diff --git a/repeater/web/update_endpoints.py b/repeater/web/update_endpoints.py index 58d9fa8..173592f 100644 --- a/repeater/web/update_endpoints.py +++ b/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}) diff --git a/tests/test_api_endpoints_core_coverage.py b/tests/test_api_endpoints_core_coverage.py index 557f56a..feba70a 100644 --- a/tests/test_api_endpoints_core_coverage.py +++ b/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"] diff --git a/tests/test_auth_components.py b/tests/test_auth_components.py new file mode 100644 index 0000000..b36cd54 --- /dev/null +++ b/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() diff --git a/tests/test_auth_endpoints.py b/tests/test_auth_endpoints.py new file mode 100644 index 0000000..9a49413 --- /dev/null +++ b/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/ 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() diff --git a/tests/test_companion_bridge_frame_utils.py b/tests/test_companion_bridge_frame_utils.py new file mode 100644 index 0000000..cc79738 --- /dev/null +++ b/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") diff --git a/tests/test_companion_ws_proxy.py b/tests/test_companion_ws_proxy.py new file mode 100644 index 0000000..f2f9d18 --- /dev/null +++ b/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() diff --git a/tests/test_handler_helpers_mesh_cli.py b/tests/test_handler_helpers_mesh_cli.py new file mode 100644 index 0000000..5fcd27c --- /dev/null +++ b/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" diff --git a/tests/test_http_server_unit.py b/tests/test_http_server_unit.py new file mode 100644 index 0000000..757e54a --- /dev/null +++ b/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("ok", 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() == "ok" + + 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") == "ok" + + +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 diff --git a/tests/test_main_py_more.py b/tests/test_main_py_more.py new file mode 100644 index 0000000..b807300 --- /dev/null +++ b/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() diff --git a/tests/test_presets.py b/tests/test_presets.py index 8f4c7e8..b89ba14 100644 --- a/tests/test_presets.py +++ b/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") diff --git a/tests/test_service_utils.py b/tests/test_service_utils.py new file mode 100644 index 0000000..379957f --- /dev/null +++ b/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 diff --git a/tests/test_sqlite_handler_easy.py b/tests/test_sqlite_handler_easy.py new file mode 100644 index 0000000..18bd699 --- /dev/null +++ b/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] diff --git a/tests/test_update_endpoints_unit.py b/tests/test_update_endpoints_unit.py new file mode 100644 index 0000000..be222b9 --- /dev/null +++ b/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)