Initial push

1 files changed, 194 insertions, 0 deletions

diff --git a/stacks/daemon/test_bcpr_backend.py b/stacks/daemon/test_bcpr_backend.py
new file mode 100644
index 0000000..4d196ad
--- /dev/null
+++ b/stacks/daemon/test_bcpr_backend.py
@@ -0,0 +1,194 @@
+#!/usr/bin/env python3
+"""Offline unit tests for bcpr device parse + BcprKissBackend (no UART)."""
+from __future__ import annotations
+
+import configparser
+import importlib.util
+import sys
+from pathlib import Path
+from unittest.mock import patch
+
+ROOT = Path(__file__).resolve().parents[2]
+sys.path.insert(0, str(ROOT / "stacks" / "daemon"))
+
+from device_backends import (  # noqa: E402
+    BcprKissBackend,
+    DeviceBackendConfig,
+    parse_device_spec,
+)
+
+
+def load_max25d():
+    from importlib.machinery import SourceFileLoader
+
+    loader = SourceFileLoader("max25d", str(ROOT / "stacks" / "daemon" / "max25d.py"))
+    spec = importlib.util.spec_from_loader("max25d", loader)
+    mod = importlib.util.module_from_spec(spec)
+    sys.modules["max25d"] = mod
+    loader.exec_module(mod)
+    return mod
+
+
+def test_bcpr_kiss_default_link() -> None:
+    cfg = DeviceBackendConfig(
+        device_id="max25e0",
+        backend_type="max25-bcpr-kiss",
+        bcpr_device="bc0",
+    )
+    backend = BcprKissBackend(cfg, on_rx=lambda _l: None)
+    assert backend._path == "/tmp/max25-bcpr/kiss-bc0"
+    assert backend.backend_type == "max25-bcpr-kiss"
+    assert backend._is_pty is True
+
+
+def test_max25e0_default_addrs() -> None:
+    from device_backends import MAX25E0_DEFAULT_IPV4, MAX25E0_DEFAULT_IPV6
+    text = """
+[devices]
+max25e0 = max25-bcpr:bc0
+"""
+    cp = configparser.ConfigParser()
+    cp.read_string(text)
+    from device_backends import parse_device_spec
+    dev = parse_device_spec("max25e0", "max25-bcpr:bc0", cp, {"hardware": "modems"})
+    assert dev.ipv4 == MAX25E0_DEFAULT_IPV4
+    assert dev.ipv6 == MAX25E0_DEFAULT_IPV6
+    # override
+    text2 = """
+[devices]
+max25e0 = max25-bcpr:bc0
+[device.max25e0]
+ipv4 = 10.0.0.25/24
+ipv6 = fd00::25/128
+"""
+    cp2 = configparser.ConfigParser()
+    cp2.read_string(text2)
+    dev2 = parse_device_spec("max25e0", "max25-bcpr:bc0", cp2, {"hardware": "modems"})
+    assert dev2.ipv4 == "10.0.0.25/24"
+    assert dev2.ipv6 == "fd00::25/128"
+    # fork inherits from [device.max25e0] (INI colon keys need care — parse by id)
+    text3 = """
+[device.max25e0]
+ipv4 = 10.0.0.25/24
+ipv6 = fd00::25/128
+"""
+    cp3 = configparser.ConfigParser()
+    cp3.read_string(text3)
+    fork = parse_device_spec("max25e0:bc1", "max25-bcpr:bc1", cp3, {"hardware": "modems"})
+    assert fork.ipv4 == "10.0.0.25/24"
+    assert fork.ipv6 == "fd00::25/128"
+
+
+def test_parse_bcpr_spec() -> None:
+    cp = configparser.ConfigParser()
+    cp.read_string(
+        """
+[device.max25e0]
+kiss_link = /var/run/max25-bcpr/kiss-bc0
+max25_bcpr_ini = /etc/max25/max25-bcpr.ini
+"""
+    )
+    dev = parse_device_spec("max25e0", "max25-bcpr:bc0", cp, {"hardware": "tncs"})
+    assert dev.backend_type == "max25-bcpr-kiss"
+    assert dev.max25_bcpr_device == "bc0"
+    assert dev.kiss_link == "/var/run/max25-bcpr/kiss-bc0"
+    assert dev.max25_bcpr_ini == "/etc/max25/max25-bcpr.ini"
+
+
+def test_feature_gate_bcpr() -> None:
+    mod = load_max25d()
+    cp = configparser.ConfigParser()
+    cp.read_string(
+        """
+[features]
+max25_bcpr = no
+[devices]
+default = max25e0
+max25e0 = max25-bcpr:bc0
+"""
+    )
+    cfg = mod.DaemonConfig()
+    cfg.feature_max25_bcpr = False
+    devices = mod.parse_devices(cp, cfg)
+    # parse_devices itself may include; load_config filters by features
+    assert any(d.device_id == "max25e0" for d in devices)
+    filtered = [d for d in devices if mod._device_allowed_by_features(d, cfg)]
+    assert filtered == []
+
+    cfg.feature_max25_bcpr = True
+    filtered = [d for d in devices if mod._device_allowed_by_features(d, cfg)]
+    assert len(filtered) == 1
+    assert filtered[0].backend_type == "max25-bcpr-kiss"
+
+
+def test_bcpr_kiss_pty_skips_tcdrain() -> None:
+    cfg = DeviceBackendConfig(
+        device_id="max25e0",
+        backend_type="max25-bcpr-kiss",
+        kiss_link="/tmp/max25-bcpr/kiss-bc0",
+        bcpr_device="bc0",
+    )
+    backend = BcprKissBackend(cfg, on_rx=lambda _l: None)
+    backend._fd = 99
+    backend._kiss_active = True
+    written: list[bytes] = []
+    drained = {"n": 0}
+
+    def fake_drain(_fd: int) -> None:
+        drained["n"] += 1
+
+    with patch("os.write", side_effect=lambda _fd, data: written.append(data) or len(data)):
+        with patch("termios.tcdrain", side_effect=fake_drain):
+            ok, display = backend.transmit("CB-0", "QST", "hello", ax25_ui=True)
+    assert ok, display
+    assert written and written[0].startswith(b"\xc0")
+    assert drained["n"] == 0
+
+
+def test_bcpr_kiss_stabilize_reopens_on_inode_mismatch() -> None:
+    cfg = DeviceBackendConfig(
+        device_id="max25e0",
+        backend_type="max25-bcpr-kiss",
+        kiss_link="/tmp/max25-bcpr/kiss-bc0",
+        bcpr_device="bc0",
+    )
+    backend = BcprKissBackend(cfg, on_rx=lambda _l: None)
+    backend._fd = 7
+    backend._kiss_active = True
+    backend.status = "ready"
+    backend._mycall = "CB-0"
+    calls: list[str] = []
+
+    class _St:
+        def __init__(self, ino: int) -> None:
+            self.st_ino = ino
+            self.st_dev = 1
+
+    with patch("os.path.exists", return_value=True):
+        with patch("os.stat", return_value=_St(200)):
+            with patch("os.fstat", return_value=_St(100)):
+                with patch.object(backend, "close", side_effect=lambda: calls.append("close")):
+                    with patch.object(backend, "open", side_effect=lambda: calls.append("open") or True):
+                        with patch.object(
+                            backend,
+                            "attach_session",
+                            side_effect=lambda c: calls.append(f"attach:{c}") or True,
+                        ):
+                            ok = backend.stabilize_session("CB-0", force=False)
+    assert ok
+    assert calls == ["close", "open", "attach:CB-0"]
+
+
+def main() -> int:
+    test_bcpr_kiss_default_link()
+    test_max25e0_default_addrs()
+    test_parse_bcpr_spec()
+    test_feature_gate_bcpr()
+    test_bcpr_kiss_pty_skips_tcdrain()
+    test_bcpr_kiss_stabilize_reopens_on_inode_mismatch()
+    print("OK: bcpr backend tests")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())