Initial push

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())