370 lines
14 KiB
Python
Executable file
370 lines
14 KiB
Python
Executable file
#!/usr/bin/env python3
|
|
# Part of JoeyLib, Copyright (c) 2026 Scott Duensing, under the MIT License.
|
|
# See LICENSE.md for license information.
|
|
# SPDX-License-Identifier: MIT
|
|
|
|
"""serialLoopPeer.py - the host end of the serial loopback gates.
|
|
|
|
The guest runs SERTEST (examples/sertest): it sends a banner, then echoes
|
|
every byte it receives. This peer waits for the banner (the guest's
|
|
transmit works), then sends a probe line and expects it back (its receive
|
|
works too).
|
|
|
|
serialLoopPeer.py tcp-server PORT WALL [BANNER] listen; the emulator connects
|
|
serialLoopPeer.py tcp-client PORT WALL [BANNER] connect to the emulator
|
|
serialLoopPeer.py pty PATHFILE WALL [BANNER] make a raw pty pair, write the
|
|
guest end's path to PATHFILE
|
|
|
|
WALL is the wall-clock budget in seconds. Prints one LOOP PASS / LOOP FAIL
|
|
line; exit 0 on pass, 1 on fail.
|
|
|
|
--close-hold FILE (pty) then has SERTEST close the port: its DTR check, run
|
|
for the closes alone (a pty has no modem lines), every marker line
|
|
acknowledged, after which the port stays closed. Prints one CLOSE PASS /
|
|
CLOSE FAIL line more, and keeps the pty pair open until FILE exists or WALL
|
|
runs out, so the gate can read the guest end's settings after the close
|
|
(verify-sdl3-serial.sh).
|
|
|
|
--stream N --rate RATE (TCP, nullmodem transparent:1) then streams N bytes of printable
|
|
ASCII through SERTEST's echo, at most STREAM_WINDOW bytes ahead of what has
|
|
come back, and checks every byte returns once, in order. Prints one STREAM
|
|
PASS / STREAM FAIL line more, with the round trip's bytes a second; PASS
|
|
needs that to be at least STREAM_MIN_PERCENT of what the line rate RATE
|
|
(bits a second) carries.
|
|
|
|
--dtr-check (TCP, with DOSBox's nullmodem in transparent:0 mode, which puts
|
|
the guest's modem lines in the byte stream as they change: 0xFF 0xFF is a
|
|
data 0xFF, 0xFF <b> the lines, bit 0 RTS, bit 1 DTR) then runs SERTEST's
|
|
DTR check: it sends DTR_CHECK, acknowledges each of SERTEST's marker lines
|
|
(so SERTEST changes a line only once the marker before has gone out whole),
|
|
and reads the line states against the markers. jlSerialSetDtr(false) and
|
|
(true) must reach the line, closing and opening the port again must keep
|
|
DTR up, and after a hang-up (DTR dropped, then the port closed) DTR must
|
|
stay down for HANGUP_WATCH_SECONDS. Prints one DTR PASS / DTR FAIL line
|
|
more.
|
|
"""
|
|
|
|
import os
|
|
import select
|
|
import socket
|
|
import sys
|
|
import time
|
|
import tty
|
|
|
|
DEFAULT_BANNER = b"JLSERTEST-READY"
|
|
PROBE = b"PING12345"
|
|
RESEND_SECONDS = 3.0 # the probe is re-sent until it comes back
|
|
POLL_SECONDS = 0.5
|
|
CONNECT_RETRY_SECONDS = 0.5
|
|
READ_SIZE = 256
|
|
SHOWN_BYTES = 80
|
|
|
|
# --stream: what the window, the chunks, and the pass mark are, and the bytes
|
|
# it sends (printable ASCII, so never SERTEST's DTR_CHECK).
|
|
STREAM_WINDOW = 1024
|
|
STREAM_CHUNK = 64
|
|
STREAM_POLL_SECONDS = 0.01
|
|
STREAM_SETTLE_SECONDS = 1.0
|
|
STREAM_MIN_PERCENT = 75
|
|
STREAM_FIRST_BYTE = 0x20
|
|
STREAM_BYTE_VALUES = 95
|
|
LINE_BITS_PER_BYTE = 10 # 8N1: a start bit, 8 data bits, a stop bit
|
|
|
|
# SERTEST's DTR check (examples/sertest/sertest.c).
|
|
DTR_CHECK = b"\x04"
|
|
DTR_ACK = b"\x06"
|
|
MARKS = (("off", b"JLDTR-OFF\r\n"), ("on", b"JLDTR-ON\r\n"), ("reopened", b"JLDTR-REOPENED\r\n"), ("hangup", b"JLDTR-HANGUP\r\n"))
|
|
HANGUP_WATCH_SECONDS = 3.0
|
|
|
|
# DOSBox nullmodem transparent:0 framing.
|
|
LINE_ESCAPE = 0xFF
|
|
LINE_DTR = 0x02
|
|
|
|
|
|
class LineDecoder:
|
|
"""Splits a transparent:0 stream into data and the guest's DTR states,
|
|
each state kept with the count of data bytes that came before it."""
|
|
|
|
def __init__(self):
|
|
self.pending = b""
|
|
self.dataCount = 0
|
|
self.dtr = []
|
|
|
|
def feed(self, raw):
|
|
buf = self.pending + raw
|
|
data = bytearray()
|
|
i = 0
|
|
while i < len(buf):
|
|
if buf[i] != LINE_ESCAPE:
|
|
data.append(buf[i])
|
|
i += 1
|
|
continue
|
|
if i + 1 >= len(buf):
|
|
break
|
|
if buf[i + 1] == LINE_ESCAPE:
|
|
data.append(LINE_ESCAPE)
|
|
else:
|
|
self.dtr.append((self.dataCount + len(data), (buf[i + 1] & LINE_DTR) != 0))
|
|
i += 2
|
|
self.pending = buf[i:]
|
|
self.dataCount += len(data)
|
|
return bytes(data)
|
|
|
|
def dtrBefore(self, position):
|
|
"""DTR as it was when the data byte at position arrived (None: never reported)."""
|
|
state = None
|
|
for at, dtr in self.dtr:
|
|
if at > position:
|
|
break
|
|
state = dtr
|
|
return state
|
|
|
|
def dtrChanges(self, first, last, value):
|
|
"""Whether DTR was reported as value after data byte first, up to last."""
|
|
return any(first < at <= last and dtr == value for at, dtr in self.dtr)
|
|
|
|
|
|
def acceptOne(port, wall):
|
|
server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
|
server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
|
|
server.bind(("127.0.0.1", port))
|
|
server.listen(1)
|
|
server.settimeout(wall)
|
|
try:
|
|
conn, _ = server.accept()
|
|
except socket.timeout:
|
|
return None
|
|
return conn
|
|
|
|
|
|
def connectOne(port, wall, start):
|
|
while time.time() - start < wall:
|
|
try:
|
|
return socket.create_connection(("127.0.0.1", port), timeout=2)
|
|
except OSError:
|
|
time.sleep(CONNECT_RETRY_SECONDS)
|
|
return None
|
|
|
|
|
|
def openPty(pathFile):
|
|
master, slave = os.openpty()
|
|
tty.setraw(slave)
|
|
tty.setraw(master)
|
|
# Publish atomically: the gate polls for the file.
|
|
with open(pathFile + ".tmp", "w") as f:
|
|
f.write(os.ttyname(slave))
|
|
os.rename(pathFile + ".tmp", pathFile)
|
|
return master
|
|
|
|
|
|
def ackMarks(send, got, base, marks):
|
|
# Acknowledge each of SERTEST's marker lines (after got[base:]) as it
|
|
# completes, recording where it is in marks.
|
|
while len(marks) < len(MARKS):
|
|
name, mark = MARKS[len(marks)]
|
|
at = got.find(mark, base)
|
|
if at < 0:
|
|
return
|
|
marks[name] = at
|
|
send(DTR_ACK)
|
|
|
|
|
|
def closeHold(send, receive, wall, start, got, holdFile):
|
|
# After the loopback: SERTEST's DTR check for its closes alone, then the
|
|
# pty held open until holdFile exists.
|
|
base = len(got)
|
|
send(DTR_CHECK)
|
|
marks = {}
|
|
while len(marks) < len(MARKS) and time.time() - start < wall:
|
|
data = receive()
|
|
if data is None:
|
|
break
|
|
got += data
|
|
ackMarks(send, got, base, marks)
|
|
if len(marks) == len(MARKS):
|
|
print("CLOSE PASS: every marker acknowledged; SERTEST has closed the port")
|
|
status = 0
|
|
else:
|
|
print("CLOSE FAIL: no %r from SERTEST; stream=%r" % (MARKS[len(marks)][1], got[base:base + SHOWN_BYTES]))
|
|
status = 1
|
|
sys.stdout.flush()
|
|
while not os.path.exists(holdFile) and time.time() - start < wall:
|
|
receive()
|
|
return status
|
|
|
|
|
|
def dtrCheck(send, receive, wall, start, lines, got):
|
|
# After the loopback: the stream so far is got; lines has decoded it.
|
|
base = len(got)
|
|
send(DTR_CHECK)
|
|
marks = {}
|
|
hangupAt = None
|
|
while time.time() - start < wall:
|
|
if hangupAt is not None and time.time() - hangupAt >= HANGUP_WATCH_SECONDS:
|
|
break
|
|
data = receive()
|
|
if data is None:
|
|
break
|
|
got += data
|
|
# The last marker starts the watch.
|
|
ackMarks(send, got, base, marks)
|
|
if hangupAt is None and len(marks) == len(MARKS):
|
|
hangupAt = time.time()
|
|
if len(marks) < len(MARKS):
|
|
print("DTR FAIL: no %r from SERTEST; stream=%r" % (MARKS[len(marks)][1], got[base:base + SHOWN_BYTES]))
|
|
return 1
|
|
if time.time() - hangupAt < HANGUP_WATCH_SECONDS:
|
|
print("DTR FAIL: out of time before the watch after the hang-up ended")
|
|
return 1
|
|
fails = []
|
|
if lines.dtrBefore(marks["off"]) is not False:
|
|
fails.append("jlSerialSetDtr(false) did not drop DTR")
|
|
if lines.dtrBefore(marks["on"]) is not True:
|
|
fails.append("jlSerialSetDtr(true) did not raise DTR")
|
|
if lines.dtrChanges(marks["on"], marks["reopened"], False):
|
|
fails.append("closing and opening the port dropped DTR")
|
|
if lines.dtrBefore(marks["hangup"]) is not False:
|
|
fails.append("jlSerialSetDtr(false) after the reopen did not drop DTR")
|
|
if lines.dtrChanges(marks["hangup"], lines.dataCount, True):
|
|
fails.append("closing the port after the hang-up raised DTR")
|
|
if fails:
|
|
print("DTR FAIL: %s; DTR states (data byte, DTR) since the check: %r" % ("; ".join(fails), [x for x in lines.dtr if x[0] >= base]))
|
|
return 1
|
|
print("DTR PASS: SetDtr(false/true) reached the line, close+open kept DTR up, a hang-up stayed down %.0f s after the close" % HANGUP_WATCH_SECONDS)
|
|
return 0
|
|
|
|
|
|
def loop(send, receive, wall, start, banner):
|
|
# send(bytes); receive() -> bytes, b"" for "nothing yet", None at EOF.
|
|
# Returns the exit status and the stream.
|
|
got = b""
|
|
sentAt = None
|
|
lastSend = 0.0
|
|
while time.time() - start < wall:
|
|
if banner in got and time.time() - lastSend > RESEND_SECONDS:
|
|
send(PROBE + b"\r\n")
|
|
lastSend = time.time()
|
|
if sentAt is None:
|
|
sentAt = len(got)
|
|
data = receive()
|
|
if data is None:
|
|
break
|
|
got += data
|
|
if sentAt is not None and PROBE in got[sentAt:]:
|
|
print("LOOP PASS: banner %r seen, probe echoed; stream=%r" % (banner, got[:SHOWN_BYTES]))
|
|
return 0, got
|
|
print("LOOP FAIL: stream=%r" % got[:SHOWN_BYTES + 40])
|
|
return 1, got
|
|
|
|
|
|
def stream(send, receive, wall, start, total, rate):
|
|
# The probe's last echoes go first, then the stream and its echo.
|
|
pattern = bytes(STREAM_FIRST_BYTE + (k % STREAM_BYTE_VALUES) for k in range(total))
|
|
settle = time.time() + STREAM_SETTLE_SECONDS
|
|
while time.time() < settle:
|
|
if receive() is None:
|
|
print("STREAM FAIL: the connection closed before the stream")
|
|
return 1
|
|
sent = 0
|
|
got = b""
|
|
began = time.time()
|
|
while len(got) < total and time.time() - start < wall:
|
|
if sent < total and sent - len(got) < STREAM_WINDOW:
|
|
chunk = pattern[sent:sent + STREAM_CHUNK]
|
|
send(chunk)
|
|
sent += len(chunk)
|
|
data = receive()
|
|
if data is None:
|
|
break
|
|
got += data
|
|
elapsed = time.time() - began
|
|
perSecond = len(got) / elapsed if elapsed > 0 else 0.0
|
|
floor = rate / LINE_BITS_PER_BYTE * STREAM_MIN_PERCENT / 100.0
|
|
if got != pattern:
|
|
at = next((k for k in range(min(len(got), total)) if got[k] != pattern[k]), min(len(got), total))
|
|
print("STREAM FAIL: %d of %d bytes back, the first wrong at %d; %r" % (len(got), total, at, got[at:at + 16]))
|
|
return 1
|
|
if perSecond < floor:
|
|
print("STREAM FAIL: %d bytes back in order, %.0f a second, under %.0f (%d%% of %d baud)" % (total, perSecond, floor, STREAM_MIN_PERCENT, rate))
|
|
return 1
|
|
print("STREAM PASS: %d bytes back in order, none lost, %.0f a second (%d baud carries %d)" % (total, perSecond, rate, rate // LINE_BITS_PER_BYTE))
|
|
return 0
|
|
|
|
|
|
def main(argv):
|
|
withDtrCheck = "--dtr-check" in argv
|
|
argv = [arg for arg in argv if arg != "--dtr-check"]
|
|
streamBytes = 0
|
|
rate = 0
|
|
if "--stream" in argv and argv.index("--stream") + 1 < len(argv):
|
|
at = argv.index("--stream")
|
|
streamBytes = int(argv[at + 1])
|
|
argv = argv[:at] + argv[at + 2:]
|
|
if "--rate" in argv and argv.index("--rate") + 1 < len(argv):
|
|
at = argv.index("--rate")
|
|
rate = int(argv[at + 1])
|
|
argv = argv[:at] + argv[at + 2:]
|
|
holdFile = None
|
|
if "--close-hold" in argv and argv.index("--close-hold") + 1 < len(argv):
|
|
at = argv.index("--close-hold")
|
|
holdFile = argv[at + 1]
|
|
argv = argv[:at] + argv[at + 2:]
|
|
if len(argv) < 4 or argv[1] not in ("tcp-server", "tcp-client", "pty") or (withDtrCheck and argv[1] == "pty") or (holdFile is not None and argv[1] != "pty") or (streamBytes > 0 and (argv[1] == "pty" or withDtrCheck or rate <= 0)):
|
|
sys.stderr.write(__doc__)
|
|
return 2
|
|
mode = argv[1]
|
|
wall = float(argv[3])
|
|
banner = argv[4].encode() if len(argv) > 4 else DEFAULT_BANNER
|
|
start = time.time()
|
|
|
|
if mode == "pty":
|
|
master = openPty(argv[2])
|
|
|
|
def receive():
|
|
ready, _, _ = select.select([master], [], [], POLL_SECONDS)
|
|
if not ready:
|
|
return b""
|
|
try:
|
|
return os.read(master, READ_SIZE)
|
|
except OSError:
|
|
return b""
|
|
|
|
send = lambda data: os.write(master, data)
|
|
status, got = loop(send, receive, wall, start, banner)
|
|
if status != 0 or holdFile is None:
|
|
return status
|
|
return closeHold(send, receive, wall, start, got, holdFile)
|
|
|
|
port = int(argv[2])
|
|
conn = acceptOne(port, wall) if mode == "tcp-server" else connectOne(port, wall, start)
|
|
if conn is None:
|
|
print("LOOP FAIL: no connection on port %d" % port)
|
|
return 1
|
|
conn.settimeout(POLL_SECONDS)
|
|
|
|
lines = LineDecoder()
|
|
|
|
def receive():
|
|
try:
|
|
data = conn.recv(READ_SIZE)
|
|
except socket.timeout:
|
|
return b""
|
|
except OSError:
|
|
return None
|
|
if not data:
|
|
return None
|
|
return lines.feed(data) if withDtrCheck else data
|
|
|
|
status, got = loop(conn.sendall, receive, wall, start, banner)
|
|
if status != 0:
|
|
return status
|
|
if streamBytes > 0:
|
|
conn.settimeout(STREAM_POLL_SECONDS)
|
|
return stream(conn.sendall, receive, wall, start, streamBytes, rate)
|
|
if not withDtrCheck:
|
|
return status
|
|
return dtrCheck(conn.sendall, receive, wall, start, lines, got)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main(sys.argv))
|