TP3/shell/tests/ptyharness.py

#!/usr/bin/env python3
"""Shared pty helpers for driving tpshell non-interactively.

The shell is a full-screen curses-style program: it switches the tty to raw
mode and reads one key byte at a time (Term.GetCh), so a test has to drive it
through a real pty, one key at a time, and wait for silence between keys.

Provides:
  drain/send          - read-until-quiet and "write keys, then settle"
  visible()           - strip ANSI escapes from captured output
  Screen              - minimal VT100 emulator (CSI 2J / H / r;cH / nD / m)
                        so assertions can run against the rendered screen
"""
import fcntl
import os
import pty
import select
import struct
import termios
import time

ROWS, COLS = 25, 80
SHELL_DIR = "/home/eric/Projets/Projets-Modula2/MyWork/TP3-comp/shell"
ESC = b"\x1b"
CTRL_K = b"\x0b"


def set_winsize(fd, rows=ROWS, cols=COLS):
    fcntl.ioctl(fd, termios.TIOCSWINSZ, struct.pack("HHHH", rows, cols, 0, 0))


def drain(fd, quiet=0.45, total=8.0):
    """Read until the child has been silent for `quiet` seconds."""
    out = b""
    deadline = time.time() + total
    while time.time() < deadline:
        r, _, _ = select.select([fd], [], [], 0.05)
        if r:
            try:
                d = os.read(fd, 65536)
            except OSError:
                break
            if not d:
                break
            out += d
            deadline = time.time() + quiet
    return out


def send(fd, data, quiet=0.45):
    """Write keys, then wait for the child to settle."""
    os.write(fd, data)
    return drain(fd, quiet)


def spawn(exe="./tpshell", cwd=SHELL_DIR):
    """Fork a pty running tpshell.  Returns (pid, fd)."""
    os.chdir(cwd)
    pid, fd = pty.fork()
    if pid == 0:
        os.environ["TERM"] = "xterm"
        os.execv(exe, [exe])
        os._exit(127)
    set_winsize(fd)
    return pid, fd


def reap(fd, pid):
    try:
        os.close(fd)
    except OSError:
        pass
    _, status = os.waitpid(pid, 0)
    return status


def status_str(status):
    if os.WIFSIGNALED(status):
        sig = os.WTERMSIG(status)
        names = {6: "SIGABRT", 11: "SIGSEGV", 15: "SIGTERM", 9: "SIGKILL"}
        return "SIGNAL %d (%s)" % (sig, names.get(sig, "?"))
    if os.WIFEXITED(status):
        return "EXIT %d" % os.WEXITSTATUS(status)
    return "raw %d" % status


def visible(b):
    """Strip ANSI escapes from captured output (latin-1 safe)."""
    out, i = [], 0
    while i < len(b):
        if b[i] == 0x1B:
            j = i + 1
            if j < len(b) and b[j] == ord("["):
                j += 1
                while j < len(b) and b[j] not in b"@ABCDEFGHJKSTfmnsulh":
                    j += 1
                i = j + 1
                continue
            i += 2
            continue
        out.append(b[i : i + 1])
        i += 1
    return b"".join(out).decode("latin-1")


class Screen:
    """Enough VT100 to render what Term.mod emits."""

    def __init__(self, rows=ROWS, cols=COLS):
        self.rows, self.cols = rows, cols
        self.r = self.c = 0
        self.clear()

    def clear(self):
        self.buf = [[" "] * self.cols for _ in range(self.rows)]

    def put(self, ch):
        if ch == "\n":
            self.r = min(self.r + 1, self.rows - 1)
            return
        if ch == "\r":
            self.c = 0
            return
        if 0 <= self.r < self.rows and 0 <= self.c < self.cols:
            self.buf[self.r][self.c] = ch
        self.c += 1
        if self.c >= self.cols:
            self.c = 0
            self.r = min(self.r + 1, self.rows - 1)

    def feed(self, data):
        if isinstance(data, bytes):
            data = data.decode("latin-1")
        i, n = 0, len(data)
        while i < n:
            ch = data[i]
            if ch == "\x1b":
                j = i + 1
                if j < n and data[j] == "[":
                    j += 1
                    start = j
                    while j < n and data[j] not in "@ABCDEFGHJKSTfmnsulh":
                        j += 1
                    if j >= n:
                        break
                    params, final = data[start:j], data[j]
                    if final == "H":
                        p = params.split(";")
                        self.r = (int(p[0]) if len(p) > 0 and p[0] else 1) - 1
                        self.c = (int(p[1]) if len(p) > 1 and p[1] else 1) - 1
                    elif final == "J":
                        if params == "2":
                            self.clear()
                    elif final == "D":
                        self.c = max(0, self.c - (int(params) if params else 1))
                    i = j + 1
                    continue
                i += 2
                continue
            self.put(ch)
            i += 1

    def text(self):
        return "\n".join("".join(row).rstrip() for row in self.buf)

    def row_with(self, needle):
        for idx, row in enumerate(self.buf):
            if needle in "".join(row):
                return idx
        return None