Initial push

1 files changed, 518 insertions, 0 deletions

diff --git a/stacks/daemon/test_multi_device.py b/stacks/daemon/test_multi_device.py
new file mode 100644
index 0000000..a46a449
--- /dev/null
+++ b/stacks/daemon/test_multi_device.py
@@ -0,0 +1,518 @@
+#!/usr/bin/env python3
+"""Multi-device config and M25/1 protocol tests for max25d (no serial hardware)."""
+from __future__ import annotations
+
+import configparser
+import importlib.util
+import socket
+import subprocess
+import sys
+import tempfile
+import time
+from pathlib import Path
+from typing import Callable
+from unittest.mock import patch
+
+ROOT = Path(__file__).resolve().parents[2]
+LAUNCHER = ROOT / "stacks/daemon/max25d"
+DAEMON = ROOT / "stacks/daemon/max25d.py"
+
+
+def load_max25d_module():
+    from importlib.machinery import SourceFileLoader
+
+    loader = SourceFileLoader("max25d", str(DAEMON))
+    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
+
+
+class LineReader:
+    def __init__(self, sock: socket.socket) -> None:
+        self.sock = sock
+        self.buf = b""
+
+    def read(self, timeout: float = 5.0) -> str:
+        self.sock.settimeout(timeout)
+        while b"\n" not in self.buf:
+            chunk = self.sock.recv(4096)
+            if not chunk:
+                raise RuntimeError("connection closed")
+            self.buf += chunk
+        line, self.buf = self.buf.split(b"\n", 1)
+        return line.decode("utf-8")
+
+
+def free_port() -> int:
+    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    sock.bind(("127.0.0.1", 0))
+    port = sock.getsockname()[1]
+    sock.close()
+    return port
+
+
+class FakeBackend:
+    backend_type = "mock"
+
+    def __init__(self, cfg, root, on_rx, log=None):
+        self.device_id = cfg.device_id
+        self._on_rx = on_rx
+        self.status = "ready"
+
+    def open(self):
+        return True
+
+    def attach_session(self, mycall):
+        return True
+
+    def transmit(self, src, dst, text, ax25_ui):
+        return True, f"[AX25 UI {src}>{dst}] {text}"
+
+    def close(self):
+        self.status = "closed"
+
+    def detach_session(self):
+        pass
+
+
+def test_parse_legacy_single_device() -> None:
+    mod = load_max25d_module()
+    cp = configparser.ConfigParser()
+    cp.read_string(
+        """
+[daemon]
+device = tnc2c
+hardware = tncs
+[serial]
+device = /dev/ttyS4
+baud = 19200
+"""
+    )
+    cfg = mod.DaemonConfig(device="tnc2c", hardware="tncs")
+    devices = mod.parse_devices(cp, cfg)
+    assert len(devices) == 1
+    assert devices[0].device_id == "tnc2c"
+    assert devices[0].serial_device == "/dev/ttyS4"
+    assert devices[0].serial_baud == 19200
+    assert devices[0].backend_type == "kiss-serial"
+
+
+def test_parse_multi_devices() -> None:
+    mod = load_max25d_module()
+    cp = configparser.ConfigParser()
+    cp.read_string(
+        """
+[devices]
+default = tnc2c
+tnc2c = /dev/ttyS4
+pktnc2 = /dev/ttyS5
+dev3 = /dev/ttyUSB0
+dev4 = /dev/ttyUSB1
+dev5 = /dev/ttyUSB2
+[serial.dev3]
+baud = 9600
+"""
+    )
+    cfg = mod.DaemonConfig()
+    devices = mod.parse_devices(cp, cfg)
+    assert len(devices) == 5
+    ids = {d.device_id for d in devices}
+    assert ids == {"tnc2c", "pktnc2", "dev3", "dev4", "dev5"}
+    by_id = {d.device_id: d for d in devices}
+    assert by_id["tnc2c"].serial_device == "/dev/ttyS4"
+    assert by_id["dev3"].serial_baud == 9600
+    assert cfg.default_device == "tnc2c"
+
+
+def test_parse_pccom_kiss_device() -> None:
+    mod = load_max25d_module()
+    cp = configparser.ConfigParser()
+    cp.read_string(
+        """
+[features]
+baycom = yes
+pccom = yes
+[devices]
+default = pccom-kiss
+pccom-kiss = /dev/ttyUSB0
+"""
+    )
+    cfg = mod.DaemonConfig()
+    cfg.feature_baycom = True
+    cfg.feature_pccom = True
+    devices = mod.parse_devices(cp, cfg)
+    by_id = {d.device_id: d for d in devices}
+    assert by_id["pccom-kiss"].backend_type == "kiss-raw-serial"
+    assert by_id["pccom-kiss"].serial_device == "/dev/ttyUSB0"
+
+
+def test_parse_heterogeneous_devices() -> None:
+    mod = load_max25d_module()
+    cp = configparser.ConfigParser()
+    cp.read_string(
+        """
+[devices]
+default = tnc2c
+tnc2c = /dev/ttyS4
+baycom-ser12 = baycom:a
+baycom-kiss = /dev/ttyUSB0
+soft-crdop = crdop:default
+[device.baycom-ser12]
+kiss_link = /var/run/baycom-pr/kiss
+modem = a
+baycom_ini = /etc/baycom/baycom-pr.ini
+[device.soft-crdop]
+port = 8515
+"""
+    )
+    cfg = mod.DaemonConfig()
+    devices = mod.parse_devices(cp, cfg)
+    by_id = {d.device_id: d for d in devices}
+    assert by_id["tnc2c"].backend_type == "kiss-serial"
+    assert by_id["baycom-ser12"].backend_type == "baycom-kiss"
+    assert by_id["baycom-ser12"].kiss_link == "/var/run/baycom-pr/kiss"
+    assert by_id["baycom-ser12"].baycom_ini == "/etc/baycom/baycom-pr.ini"
+    assert by_id["baycom-kiss"].backend_type == "kiss-raw-serial"
+    assert by_id["soft-crdop"].backend_type == "crdop-tcp"
+    assert by_id["soft-crdop"].crdop_port == 8515
+
+
+def test_parse_enabled_filter() -> None:
+    mod = load_max25d_module()
+    cp = configparser.ConfigParser()
+    cp.read_string(
+        """
+[devices]
+enabled = tnc2c,pktnc2
+tnc2c = /dev/ttyS4
+pktnc2 = /dev/ttyS5
+extra = /dev/ttyUSB9
+"""
+    )
+    cfg = mod.DaemonConfig()
+    devices = mod.parse_devices(cp, cfg)
+    enabled = {d.device_id for d in devices if d.enabled}
+    disabled = {d.device_id for d in devices if not d.enabled}
+    assert enabled == {"tnc2c", "pktnc2"}
+    assert disabled == {"extra"}
+
+
+def test_backend_open_retry_on_error() -> None:
+    """CONNECT retries KISS open after a transient error-no-device."""
+    mod = load_max25d_module()
+    cfg = mod.DaemonConfig(hardware="modems", serial_enabled=True)
+    cfg.devices = [
+        mod.DeviceBackendConfig(
+            device_id="baycom-ser12",
+            backend_type="baycom-kiss",
+            kiss_link="/nonexistent/kiss",
+            enabled=True,
+        )
+    ]
+    cfg.default_device = "baycom-ser12"
+    state = mod.DaemonState(cfg=cfg)
+    mod.init_device_runtimes(state)
+
+    class ErrorThenReady(FakeBackend):
+        attempts = 0
+
+        def open(self):
+            ErrorThenReady.attempts += 1
+            if ErrorThenReady.attempts == 1:
+                self.status = "error-no-device"
+                return False
+            self.status = "ready"
+            return True
+
+    ErrorThenReady.attempts = 0
+    with patch.object(
+        mod,
+        "create_backend",
+        side_effect=lambda c, r, rx, log, prefix=None, on_invalid=None: ErrorThenReady(c, r, rx, log),
+    ):
+        assert not mod.attach_backend_session(state, "baycom-ser12")
+        assert mod.attach_backend_session(state, "baycom-ser12")
+        assert ErrorThenReady.attempts == 2
+
+
+def test_five_mock_backends() -> None:
+    """Five concurrent backend instances via mocked create_backend."""
+    mod = load_max25d_module()
+    opened: list[str] = []
+
+    class CountingBackend(FakeBackend):
+        def open(self):
+            opened.append(self.device_id)
+            return True
+
+    cfg = mod.DaemonConfig(hardware="tncs", serial_enabled=True)
+    cfg.devices = [
+        mod.DeviceBackendConfig(device_id=f"d{i}", serial_device=f"/dev/fake{i - 1}", enabled=True)
+        for i in range(1, 6)
+    ]
+    for d in cfg.devices:
+        d.backend_type = "kiss-serial"
+    cfg.default_device = "d1"
+    state = mod.DaemonState(cfg=cfg)
+    mod.init_device_runtimes(state)
+
+    with patch.object(
+        mod,
+        "create_backend",
+        side_effect=lambda c, r, rx, log, prefix=None, on_invalid=None: CountingBackend(c, r, rx, log),
+    ):
+        assert mod.attach_all_sessions(state)
+        assert len(opened) == 5
+        assert set(opened) == {f"d{i}" for i in range(1, 6)}
+
+        state.selected_device = "d3"
+        dev_id = mod.resolve_selected_device(state)
+        assert dev_id == "d3"
+        rt = state.devices["d3"]
+        ok, display = rt.backend.transmit("CB-0", "QST", "hi", True)
+        assert ok
+        assert "hi" in display
+
+
+def run_daemon_test(ini_text: str, fn: Callable[[LineReader, socket.socket], None]) -> None:
+    port = free_port()
+    with tempfile.NamedTemporaryFile("w", suffix=".ini", delete=False) as tmp:
+        tmp.write(ini_text)
+        ini_path = Path(tmp.name)
+
+    proc = subprocess.Popen(
+        [
+            str(LAUNCHER),
+            "--no-stack",
+            "--no-serial",
+            "-c",
+            str(ini_path),
+            "--tcp-port",
+            str(port),
+        ],
+        stdout=subprocess.DEVNULL,
+        stderr=subprocess.PIPE,
+        text=True,
+    )
+    time.sleep(0.6)
+    try:
+        sock = socket.create_connection(("127.0.0.1", port), timeout=3)
+    except OSError as exc:
+        proc.terminate()
+        proc.wait(timeout=3)
+        ini_path.unlink(missing_ok=True)
+        raise RuntimeError(f"connect failed: {exc}") from exc
+
+    reader = LineReader(sock)
+    try:
+        assert reader.read() == "OK"
+        fn(reader, sock)
+    finally:
+        sock.close()
+        proc.terminate()
+        proc.wait(timeout=5)
+        ini_path.unlink(missing_ok=True)
+
+
+def test_parse_dual_baycom_devices() -> None:
+    mod = load_max25d_module()
+    cp = configparser.ConfigParser()
+    cp.read_string(
+        """
+[devices]
+default = baycom-a
+baycom-a = baycom:a
+baycom-b = baycom:b
+[device.baycom-a]
+kiss_link = /var/run/baycom-pr/kiss-a
+modem = a
+baycom_ini = /etc/baycom/baycom-pr.ini
+[device.baycom-b]
+kiss_link = /var/run/baycom-pr/kiss-b
+modem = b
+baycom_ini = /etc/baycom/baycom-pr.ini
+"""
+    )
+    cfg = mod.DaemonConfig()
+    devices = mod.parse_devices(cp, cfg)
+    by_id = {d.device_id: d for d in devices}
+    assert set(by_id) == {"baycom-a", "baycom-b"}
+    assert by_id["baycom-a"].backend_type == "baycom-kiss"
+    assert by_id["baycom-b"].backend_type == "baycom-kiss"
+    assert by_id["baycom-a"].baycom_modem == "a"
+    assert by_id["baycom-b"].baycom_modem == "b"
+    assert by_id["baycom-a"].kiss_link == "/var/run/baycom-pr/kiss-a"
+    assert by_id["baycom-b"].kiss_link == "/var/run/baycom-pr/kiss-b"
+    assert cfg.default_device == "baycom-a"
+
+
+def test_protocol_dual_baycom() -> None:
+    # Legacy ids baycom-a/baycom-b are not on the Linux allowlist (product path: max25e0).
+    # Protocol dual-device coverage uses allowlisted modem-class ids instead.
+    ini = """
+[features]
+baycom = yes
+pccom = yes
+[daemon]
+mode = service
+hardware = modems
+device = baycom-kiss
+[devices]
+default = baycom-kiss
+baycom-kiss = /dev/null
+pccom-kiss = /dev/null
+[network]
+tcp_host = 127.0.0.1
+tcp_port = 7325
+[modem]
+callerid = N0CALL-0
+callid = QST
+[stack]
+auto_start = no
+"""
+
+    def exercise(reader: LineReader, sock: socket.socket) -> None:
+        status = reader.read()
+        assert "devices=baycom-kiss,pccom-kiss" in status
+        assert "device=baycom-kiss" in status
+
+        sock.sendall(b"GET DEVICES\n")
+        lines = []
+        while True:
+            line = reader.read()
+            lines.append(line)
+            if line == "OK":
+                break
+        device_lines = [l for l in lines if l.startswith("DEVICE ")]
+        assert len(device_lines) == 2
+        assert any("id=baycom-kiss" in l for l in device_lines)
+        assert any("id=pccom-kiss" in l for l in device_lines)
+
+        sock.sendall(b"SET DEVICE pccom-kiss\n")
+        assert reader.read() == "OK"
+
+        sock.sendall(b"GET STATUS\n")
+        st = reader.read()
+        assert "device=pccom-kiss" in st
+        assert reader.read() == "OK"
+
+    run_daemon_test(ini, exercise)
+
+
+def test_protocol_multi_device() -> None:
+    # Explicit two-device INI — share/max25d.ini.example is single-device by design.
+    ini = """
+[daemon]
+mode = standalone
+hardware = tncs
+device = tnc2c
+[devices]
+default = tnc2c
+tnc2c = /dev/ttyS4
+pktnc2 = /dev/ttyS5
+[network]
+tcp_host = 127.0.0.1
+tcp_port = 7325
+[modem]
+callerid = CB-0
+callid = QST
+[stack]
+auto_start = no
+"""
+
+    def exercise(reader: LineReader, sock: socket.socket) -> None:
+        status = reader.read()
+        assert status.startswith("STATUS ")
+        assert "devices=tnc2c,pktnc2" in status
+        assert "device=tnc2c" in status
+
+        sock.sendall(b"GET DEVICES\n")
+        lines = []
+        while True:
+            line = reader.read()
+            lines.append(line)
+            if line == "OK":
+                break
+        device_lines = [l for l in lines if l.startswith("DEVICE ")]
+        assert len(device_lines) == 2
+        assert any("id=tnc2c" in l for l in device_lines)
+        assert any("id=pktnc2" in l for l in device_lines)
+        assert any("backend=kiss-serial" in l for l in device_lines)
+
+        sock.sendall(b"SET DEVICE pktnc2\n")
+        assert reader.read() == "OK"
+
+        sock.sendall(b"GET STATUS\n")
+        st = reader.read()
+        assert "device=pktnc2" in st
+        assert reader.read() == "OK"
+
+        sock.sendall(b"SELECT DEVICE tnc2c\n")
+        assert reader.read() == "OK"
+
+        sock.sendall(b"CONNECT\n")
+        assert reader.read() == "EVENT connected"
+        assert reader.read() == "OK"
+
+        sock.sendall(b"SEND 73\n")
+        rx = reader.read()
+        assert rx.startswith("RX ")
+        assert reader.read() == "OK"
+
+        sock.sendall(b"DISCONNECT\n")
+        assert reader.read() == "EVENT disconnected"
+        assert reader.read() == "OK"
+
+    run_daemon_test(ini, exercise)
+
+
+def test_protocol_legacy_single() -> None:
+    ini = """
+[daemon]
+mode = standalone
+hardware = tncs
+device = tnc2c
+[network]
+tcp_host = 127.0.0.1
+tcp_port = 7325
+[modem]
+callerid = CB-0
+callid = QST
+[stack]
+auto_start = no
+"""
+
+    def exercise(reader: LineReader, sock: socket.socket) -> None:
+        status = reader.read()
+        assert "device=tnc2c" in status
+        assert "devices=tnc2c" in status
+
+        sock.sendall(b"SET DEVICE tnc2c\n")
+        assert reader.read() == "OK"
+
+    run_daemon_test(ini, exercise)
+
+
+def main() -> int:
+    tests = [
+        test_parse_legacy_single_device,
+        test_parse_multi_devices,
+        test_parse_heterogeneous_devices,
+        test_parse_dual_baycom_devices,
+        test_parse_enabled_filter,
+        test_backend_open_retry_on_error,
+        test_five_mock_backends,
+        test_protocol_multi_device,
+        test_protocol_dual_baycom,
+        test_protocol_legacy_single,
+    ]
+    for fn in tests:
+        fn()
+    print("OK: multi-device tests")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())