132 lines
3.9 KiB
Python
Executable file
132 lines
3.9 KiB
Python
Executable file
#!/usr/bin/env python3
|
|
"""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.
|
|
"""
|
|
|
|
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
|
|
|
|
|
|
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 loop(send, receive, wall, start, banner):
|
|
# send(bytes); receive() -> bytes, b"" for "nothing yet", None at EOF.
|
|
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
|
|
print("LOOP FAIL: stream=%r" % got[:SHOWN_BYTES + 40])
|
|
return 1
|
|
|
|
|
|
def main(argv):
|
|
if len(argv) < 4 or argv[1] not in ("tcp-server", "tcp-client", "pty"):
|
|
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""
|
|
|
|
return loop(lambda data: os.write(master, data), receive, wall, start, banner)
|
|
|
|
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)
|
|
|
|
def receive():
|
|
try:
|
|
data = conn.recv(READ_SIZE)
|
|
except socket.timeout:
|
|
return b""
|
|
except OSError:
|
|
return None
|
|
return data if data else None
|
|
|
|
return loop(conn.sendall, receive, wall, start, banner)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main(sys.argv))
|