diff options
Diffstat (limited to 'stacks/daemon/test_serial_watch.py')
| -rw-r--r-- | stacks/daemon/test_serial_watch.py | 241 |
1 files changed, 241 insertions, 0 deletions
diff --git a/stacks/daemon/test_serial_watch.py b/stacks/daemon/test_serial_watch.py new file mode 100644 index 0000000..7b4d8ad --- /dev/null +++ b/stacks/daemon/test_serial_watch.py @@ -0,0 +1,241 @@ +#!/usr/bin/env python3 +"""Offline tests for max25d TNC serial watch / stabilize.""" +from __future__ import annotations + +import configparser +import importlib.util +import sys +import tempfile +from pathlib import Path +from unittest.mock import MagicMock, patch + +ROOT = Path(__file__).resolve().parents[2] +DAEMON = ROOT / "stacks/daemon/max25d.py" + + +def load_max25d(): + from importlib.machinery import SourceFileLoader + + loader = SourceFileLoader("max25d", str(DAEMON)) + spec = importlib.util.spec_from_loader("max25d", loader) + assert spec and spec.loader + mod = importlib.util.module_from_spec(spec) + sys.modules["max25d"] = mod + loader.exec_module(mod) + return mod + + +def test_stack_serial_watch_config() -> None: + mod = load_max25d() + cp = configparser.ConfigParser() + cp.read_string( + """ +[stack] +auto_start = yes +serial_watch = no +serial_watch_interval = 45 +serial_repair_cooldown = 10 +stack_recover_only = no +stack_retry_interval = 90 +""" + ) + cfg = mod.DaemonConfig() + cfg.auto_start = mod._truthy(cp.get("stack", "auto_start")) + cfg.serial_watch = mod._truthy(cp.get("stack", "serial_watch")) + cfg.serial_watch_interval = cp.getint("stack", "serial_watch_interval") + cfg.serial_repair_cooldown = cp.getint("stack", "serial_repair_cooldown") + cfg.stack_recover_only = mod._truthy(cp.get("stack", "stack_recover_only")) + cfg.stack_retry_interval = cp.getint("stack", "stack_retry_interval") + assert cfg.serial_watch is False + assert cfg.serial_watch_interval == 45 + assert cfg.stack_recover_only is False + + +def test_serial_repair_statuses() -> None: + mod = load_max25d() + assert "error-io" in mod.BACKEND_RETRY_STATUSES + assert "error-host" in mod.SERIAL_REPAIR_STATUSES + assert "open" not in mod.SERIAL_REPAIR_STATUSES + + +def test_inline_tnc_prep() -> None: + mod = load_max25d() + cfg = mod.DaemonConfig(stack_recover_only=True) + state = mod.DaemonState(cfg=cfg) + from device_backends import DeviceBackendConfig # noqa: E402 + + state.devices["tnc2c"] = mod.DeviceRuntime( + cfg=DeviceBackendConfig(device_id="tnc2c", backend_type="kiss-serial") + ) + state.devices["baycom-ser12"] = mod.DeviceRuntime( + cfg=DeviceBackendConfig(device_id="baycom-ser12", backend_type="baycom-kiss") + ) + assert mod.uses_inline_tnc_prep(state, "tnc2c") + assert not mod.uses_inline_tnc_prep(state, "baycom-ser12") + + +def test_stabilize_ready_kiss_skips_probe() -> None: + from kiss_bridge import KissBridge, SerialProfile # noqa: E402 + + bridge = KissBridge(SerialProfile(), lambda _l: None, log=lambda _m: None) + bridge._fd = 99 + bridge._kiss_active = True + bridge.status = "ready" + + fake = MagicMock() + + with patch.object(bridge, "_stop_rx_thread") as stop_rx, patch.object( + bridge, "_load_recovery_mod", return_value=fake + ), patch.object(bridge, "_start_rx_thread") as start_rx: + ok = bridge.stabilize_session("CB-0", force=False) + + assert ok + assert bridge.status == "ready" + fake.probe_info.assert_not_called() + fake.recover_terminal.assert_not_called() + stop_rx.assert_not_called() + start_rx.assert_not_called() + + +def test_poll_serial_stability_skips_ready() -> None: + mod = load_max25d() + cfg = mod.DaemonConfig(serial_watch=True, stack_recover_only=True) + state = mod.DaemonState(cfg=cfg) + state.started_at = 0 + from device_backends import DeviceBackendConfig # noqa: E402 + + state.devices["tnc2c"] = mod.DeviceRuntime( + cfg=DeviceBackendConfig(device_id="tnc2c", backend_type="kiss-serial") + ) + rt = state.devices["tnc2c"] + rt.stack_status = "ready" + rt.last_watch = 0 + backend = MagicMock() + backend.status = "ready" + rt.backend = backend + + mod.poll_serial_stability(state) + + backend.stabilize_session.assert_not_called() + + +def test_stabilize_session_probe_path() -> None: + sys.path.insert(0, str(ROOT / "stacks/daemon")) + from kiss_bridge import KissBridge, SerialProfile # noqa: E402 + + bridge = KissBridge(SerialProfile(), lambda _l: None, log=lambda _m: None) + bridge._fd = 99 + bridge._kiss_active = True + + fake = MagicMock() + fake.probe_info.return_value = (True, b"TheFirmware cmd:", False) + fake.recover_terminal.return_value = (True, b"") + + with patch.object(bridge, "_write_unlocked"), patch.object( + bridge, "_drain_unlocked", return_value=b"" + ), patch.object(bridge, "_set_mycall_unlocked", return_value=True), patch.object( + bridge, "_enter_kiss_unlocked", return_value=True + ), patch.object(bridge, "_load_recovery_mod", return_value=fake), patch.object( + bridge, "_start_rx_thread" + ): + ok = bridge.stabilize_session("CB-0") + assert ok + assert bridge.status == "ready" + fake.recover_terminal.assert_not_called() + + +def test_rx_thread_deferred_until_ready() -> None: + from kiss_bridge import KissBridge, SerialProfile # noqa: E402 + + bridge = KissBridge(SerialProfile(), lambda _l: None, log=lambda _m: None) + bridge._fd = 99 + bridge._kiss_active = True + assert bridge._thread is None + bridge._start_rx_thread() + assert bridge._thread is not None + bridge._stop_rx_thread() + assert bridge._thread is None + + +def test_stabilize_stops_rx_during_recovery() -> None: + from kiss_bridge import KissBridge, SerialProfile # noqa: E402 + + bridge = KissBridge(SerialProfile(), lambda _l: None, log=lambda _m: None) + bridge._fd = 99 + bridge._kiss_active = True + bridge._start_rx_thread() + assert bridge._thread is not None + + fake = MagicMock() + fake.probe_info.return_value = (True, b"TheFirmware cmd:", False) + fake.recover_terminal.return_value = (True, b"") + + with patch.object(bridge, "_write_unlocked"), patch.object( + bridge, "_drain_unlocked", return_value=b"" + ), patch.object(bridge, "_set_mycall_unlocked", return_value=True), patch.object( + bridge, "_enter_kiss_unlocked", return_value=True + ), patch.object(bridge, "_load_recovery_mod", return_value=fake), patch.object( + bridge, "_start_rx_thread" + ) as start_rx: + ok = bridge.stabilize_session("CB-0") + assert ok + start_rx.assert_called_once() + + +def test_bootwait_escalate_config() -> None: + mod = load_max25d() + cfg = mod.DaemonConfig() + assert cfg.serial_bootwait_escalate is True + assert cfg.serial_bootwait_escalate_after == 3 + assert cfg.serial_bootwait_escalate_cooldown == 300 + + +def test_prep_escalates_on_error_host() -> None: + mod = load_max25d() + cfg = mod.DaemonConfig(stack_recover_only=True, serial_bootwait_escalate=True) + state = mod.DaemonState(cfg=cfg) + from device_backends import DeviceBackendConfig # noqa: E402 + + state.devices["tnc2c"] = mod.DeviceRuntime( + cfg=DeviceBackendConfig(device_id="tnc2c", backend_type="kiss-serial") + ) + backend = MagicMock() + backend.status = "error-host" + backend.stabilize_session.return_value = False + rt = state.devices["tnc2c"] + rt.backend = backend + + with patch.object(mod, "open_backend", return_value=True), patch.object( + mod, "escalate_to_bootwait_stack" + ) as escalate: + mod.prep_inline_serial_device(state, "tnc2c") + escalate.assert_called_once_with(state, "tnc2c") + + +def main() -> int: + test_stack_serial_watch_config() + print("OK test_stack_serial_watch_config") + test_serial_repair_statuses() + print("OK test_serial_repair_statuses") + test_inline_tnc_prep() + print("OK test_inline_tnc_prep") + test_stabilize_ready_kiss_skips_probe() + print("OK test_stabilize_ready_kiss_skips_probe") + test_poll_serial_stability_skips_ready() + print("OK test_poll_serial_stability_skips_ready") + test_stabilize_session_probe_path() + print("OK test_stabilize_session_probe_path") + test_rx_thread_deferred_until_ready() + print("OK test_rx_thread_deferred_until_ready") + test_stabilize_stops_rx_during_recovery() + print("OK test_stabilize_stops_rx_during_recovery") + test_bootwait_escalate_config() + print("OK test_bootwait_escalate_config") + test_prep_escalates_on_error_host() + print("OK test_prep_escalates_on_error_host") + print("All serial watch tests passed") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) |