#!/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).

--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

# 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 main(argv):
    withDtrCheck = "--dtr-check" in argv
    argv = [arg for arg in argv if arg != "--dtr-check"]
    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"):
        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 or not withDtrCheck:
        return status
    return dtrCheck(conn.sendall, receive, wall, start, lines, got)


if __name__ == "__main__":
    sys.exit(main(sys.argv))