diff options
Diffstat (limited to 'stacks/daemon/test_bans.py')
| -rw-r--r-- | stacks/daemon/test_bans.py | 199 |
1 files changed, 199 insertions, 0 deletions
diff --git a/stacks/daemon/test_bans.py b/stacks/daemon/test_bans.py new file mode 100644 index 0000000..80176ae --- /dev/null +++ b/stacks/daemon/test_bans.py @@ -0,0 +1,199 @@ +#!/usr/bin/env python3 +"""Unit and protocol tests for max25d AX.25 source ban list.""" +from __future__ import annotations + +import importlib.util +import socket +import subprocess +import sys +import tempfile +import time +from pathlib import Path +from unittest.mock import patch + +ROOT = Path(__file__).resolve().parents[2] +LAUNCHER = ROOT / "stacks/daemon/max25d" +DAEMON = ROOT / "stacks/daemon/max25d.py" + +sys.path.insert(0, str(DAEMON.parent)) +from banlist import BanList, callsign_banned, extract_ax25_source # noqa: E402 + + +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 + + +def test_extract_ax25_source() -> None: + assert extract_ax25_source("[AX25 UI DG1ABC>QST] 73") == "DG1ABC" + assert extract_ax25_source("[AX25 UI DK0WC-7>CB-0] hi") == "DK0WC-7" + assert extract_ax25_source("plain text") is None + assert extract_ax25_source("[CRDOP RX soft-crdop] hello") is None + + +def test_callsign_banned_matching() -> None: + assert callsign_banned("DG1ABC", "DG1ABC") + assert callsign_banned("DG1ABC", "DG1ABC-7") + assert not callsign_banned("DG1ABC", "DG1ABD") + assert callsign_banned("DK0WC-7", "DK0WC-7") + assert not callsign_banned("DK0WC-7", "DK0WC") + assert not callsign_banned("DK0WC-7", "DK0WC-8") + + +def test_banlist_file_roundtrip() -> None: + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "bans.txt" + path.write_text("# ignored comment\nDG1ABC\n\nDK0WC-7\n", encoding="utf-8") + bans = BanList(path) + assert bans.list() == ["DG1ABC", "DK0WC-7"] + assert bans.is_banned("DG1ABC-3") + assert bans.is_banned("DK0WC-7") + assert not bans.is_banned("DK0WC") + + bans.add("DL1TEST") + bans.remove("DG1ABC") + assert bans.list() == ["DK0WC-7", "DL1TEST"] + reloaded = BanList(path) + assert reloaded.list() == ["DK0WC-7", "DL1TEST"] + + +def test_on_backend_rx_silent_drop() -> None: + mod = load_max25d_module() + with tempfile.TemporaryDirectory() as tmp: + bans_path = Path(tmp) / "bans.txt" + bans_path.write_text("DG1ABC\n", encoding="utf-8") + cfg = mod.DaemonConfig(bans_file=str(bans_path)) + state = mod.DaemonState(cfg=cfg, bans=BanList(bans_path)) + mod.init_device_runtimes(state) + + sent: list[str] = [] + + def capture(_state, line, skip=None): + sent.append(line) + + with patch.object(mod, "broadcast", side_effect=capture), patch.object(mod, "log"): + mod.on_backend_rx(state, "tnc2c", "[AX25 UI DG1ABC>QST] blocked") + assert sent == [] + + mod.on_backend_rx(state, "tnc2c", "[AX25 UI DK0WC>QST] allowed") + assert sent == ["RX device=tnc2c [AX25 UI DK0WC>QST] allowed"] + + +def test_ban_commands_protocol() -> None: + port = free_port() + with tempfile.TemporaryDirectory() as tmp: + bans_path = Path(tmp) / "bans.txt" + ini_path = Path(tmp) / "max25d.ini" + ini_path.write_text( + f""" +[daemon] +device = tnc2c +hardware = tncs +[network] +tcp_host = 127.0.0.1 +tcp_port = {port} +[modem] +bans_file = {bans_path} +[stack] +auto_start = no +""", + encoding="utf-8", + ) + + 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) + raise RuntimeError(f"connect failed: {exc}") from exc + + reader = LineReader(sock) + try: + assert reader.read() == "OK" + reader.read() # STATUS + + sock.sendall(b"BAN DG1ABC\n") + assert reader.read() == "OK" + assert bans_path.read_text(encoding="utf-8").strip() == "DG1ABC" + + sock.sendall(b"BANS\n") + assert reader.read() == "BAN DG1ABC" + assert reader.read() == "OK" + + sock.sendall(b"UNBAN DG1ABC\n") + assert reader.read() == "OK" + assert bans_path.read_text(encoding="utf-8") == "" + + sock.sendall(b"UNBAN DG1ABC\n") + assert reader.read() == "ERR not banned" + + sock.sendall(b"BAN BAD!\n") + assert reader.read() == "ERR invalid callsign" + finally: + sock.close() + proc.terminate() + proc.wait(timeout=5) + + +def main() -> int: + tests = [ + test_extract_ax25_source, + test_callsign_banned_matching, + test_banlist_file_roundtrip, + test_on_backend_rx_silent_drop, + test_ban_commands_protocol, + ] + for test in tests: + test() + print(f"OK: {test.__name__}") + print("OK: banlist tests") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) |