1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
|
#!/usr/bin/env python3
"""Offline smoke test for max25d M25/1 protocol."""
import socket
import subprocess
import sys
import time
from pathlib import Path
ROOT = Path(__file__).resolve().parents[2]
DAEMON = ROOT / "stacks/daemon/max25d"
INI = ROOT / "share/max25/max25d.ini.example"
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 wait_for_port(port: int, proc: subprocess.Popen[str], timeout: float = 8.0) -> bool:
deadline = time.time() + timeout
while time.time() < deadline:
if proc.poll() is not None:
return False
try:
probe = socket.create_connection(("127.0.0.1", port), timeout=0.25)
probe.close()
return True
except OSError:
time.sleep(0.1)
return False
def main() -> int:
port = free_port()
proc = subprocess.Popen(
[str(DAEMON), "--no-stack", "--no-serial", "-c", str(INI), "--tcp-port", str(port)],
stdout=subprocess.DEVNULL,
stderr=subprocess.PIPE,
text=True,
)
if not wait_for_port(port, proc):
err = proc.stderr.read() if proc.stderr else ""
proc.terminate()
proc.wait(timeout=3)
print(f"FAIL: max25d not listening on {port}: {err.strip()}", file=sys.stderr)
return 1
try:
sock = socket.create_connection(("127.0.0.1", port), timeout=3)
except OSError as exc:
proc.terminate()
proc.wait(timeout=3)
print(f"FAIL: connect: {exc}", file=sys.stderr)
return 1
reader = LineReader(sock)
try:
assert reader.read() == "OK"
status = reader.read()
assert status.startswith("STATUS ")
sock.sendall(b"SET CALLERID DG1ABC\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"GET STATUS\n")
st = reader.read()
assert "callerid=DG1ABC" in st
assert reader.read() == "OK"
print("OK: max25d protocol smoke")
return 0
finally:
sock.close()
proc.terminate()
proc.wait(timeout=5)
if __name__ == "__main__":
sys.exit(main())
|
