TP3/shell/tests/run_com_exec.py

#!/usr/bin/env python3
"""run_com_exec.py -- execute a TP3-compiled .COM and assert its exact output.

This is the test the project could not write until now.  Everything before it
was a claim about the *image*; this is a claim about the *program*.  A .COM
that this compiler produced is booted on a real 8086 under qemu-system-i386,
and the exact bytes it writes are asserted against a file.

    .pas --Compiler.mod--> image --Linker.mod--> .COM
    .COM --bootcom.s--> floppy image --qemu--> serial bytes --assert-->

Nothing here trusts the compiler, the linker, or the boot sector:

  * the floppy image is built here, byte by byte;
  * the load address and the input descriptor's offset are *restated* from
    bootcom.s's own layout block, and this file refuses to run if they have
    drifted out of what the boot sector actually reads;
  * "the program exited" is not a marker in the output stream -- it is qemu's
    exit code, via the isa-debug-exit device.  So no sentinel byte can ever
    collide with real program output, and every byte on the serial port is
    program output with nothing to strip;
  * the expected output is a file per fixture, next to the fixture, so a
    wrong expectation shows up as a diff instead of an inline string that
    gets quietly edited to match.

Usage:

    tests/run_com_exec.py                 # every fixture with a .out file
    tests/run_com_exec.py t02_writeln     # one, by name
    tests/run_com_exec.py --list
    tests/run_com_exec.py --show t02      # print what came out, assert nothing

`--rebless` regenerates the .out files from what the machine actually did.
It exists because a project that cannot regenerate its expectations cannot
tell the difference between a bug and a stale file -- but it is never the way
to fix a red test, and the summary it prints says how many files it rewrote.
"""

import os
import subprocess
import sys
import tempfile

HERE = os.path.dirname(os.path.abspath(__file__))
SHELL = os.path.dirname(HERE)
GM2 = "/home/eric/bin/Modula2/Gm2/bin/gm2"
QEMU = "/usr/bin/qemu-system-i386"
TIMEOUT = 15

# --------------------------------------------------------------- the layout
# Restated from tests/exec/bootcom.s on purpose.  A test that asks the code
# under test where things are proves only that the code agrees with itself,
# so these are written out here and checked against each other; if the
# assembly's layout changes, the mismatch shows up as a loud failure rather
# than as a program loaded somewhere harmless.
LOAD_SEG = 0x0000        # the .COM's load segment
LOAD_OFF = 0x0100        # ...and offset; DOS puts a .COM at CS:0100
FLAT_LOAD = LOAD_SEG * 16 + LOAD_OFF
INLEN_OFF = 0x2000       # word: how many input bytes follow
INBUF_OFF = 0x2004       # the input bytes
INBUF_MAX = 0x00FC       # 2100h - 2004h
SECTORS = 16             # what bootcom.s reads off the disk
FLOPPY_SECTORS = 2880    # 1.44M, 18 sectors per track
FLOPPY_BYTES = FLOPPY_SECTORS * 512
COM_MAX = SECTORS * 512  # a .COM the boot sector could not load in full

# The descriptor has to be inside the region the boot sector reads, or the
# boot sector reads whatever the BIOS left there.  This is a check on the two
# constant blocks agreeing, and it is the reason they are both written down.
READ_END = FLAT_LOAD + SECTORS * 512
if INBUF_OFF + INBUF_MAX > READ_END:
    sys.exit("run_com_exec.py: the input descriptor at %04X..%04X runs past the "
             "%d sectors bootcom.s reads (which end at %04X)"
             % (INBUF_OFF, INBUF_OFF + INBUF_MAX, SECTORS, READ_END))
# ...and past the end of any .COM, or it would land inside the program.
if INBUF_OFF < COM_MAX:
    sys.exit("run_com_exec.py: the input descriptor at %04X is inside the %d "
             "bytes the boot sector reads, so it could overlap a .COM"
             % (INBUF_OFF, COM_MAX))

# isa-debug-exit at port 0501h.  qemu exits with (value<<1)|1, so a program
# that exits 0 gives rc=1 and a program that exits n gives (n<<1)|1.  Both
# mean "ran to completion"; anything else means it never got there.
EXIT_BASE = 1


def build_boot_sector():
    """Assemble tests/exec/bootcom.s and return its 512 bytes."""
    with tempfile.TemporaryDirectory() as td:
        obj, raw = os.path.join(td, "b.o"), os.path.join(td, "b.bin")
        r = subprocess.run(["as", "--32", "-o", obj,
                            os.path.join(HERE, "exec", "bootcom.s")],
                           capture_output=True)
        if r.returncode != 0:
            sys.exit("as failed:\n" + r.stderr.decode(errors="replace"))
        r = subprocess.run(["objcopy", "-O", "binary", obj, raw],
                           capture_output=True)
        if r.returncode != 0:
            sys.exit("objcopy failed:\n" + r.stderr.decode(errors="replace"))
        data = open(raw, "rb").read()
    if len(data) != 512:
        sys.exit("the boot sector is %d bytes, not 512" % len(data))
    if data[510:512] != b"\x55\xaa":
        sys.exit("the boot sector has no 55 AA signature")
    return data


def to_file(mem_off):
    """Where a SEGMENT-relative address ends up in the disk image.

    bootcom.s reads disk sector 2 (1-based, so file offset 512) into
    0000:0100.  Memory at segment offset M therefore comes from file offset

        512 + (M - 0100h)

    which is M + 100h for every M in the loaded region.  Getting this wrong
    is silent: the image is built, qemu boots it, and the descriptor simply
    sits at the wrong place.  t29_readln then read an INLEN of 0 - the
    descriptor was never where the shim looks - so AH=08h reported end of
    input on the very first call and readln waited forever for a line
    terminator that could not arrive.  So the conversion is in one named
    function rather than open-coded, and the two constants that define it are
    checked against each other above.
    """
    return mem_off - LOAD_OFF + 512


def build_floppy(boot, com, stdin=b""):
    """Assemble the 1.44M image: boot sector, then the .COM, then the input."""
    if len(com) > COM_MAX:
        sys.exit("the .COM is %d bytes and the boot sector reads only %d"
                 % (len(com), COM_MAX))
    if len(stdin) > INBUF_MAX:
        sys.exit("%d bytes of input, at most %d fit" % (len(stdin), INBUF_MAX))
    img = bytearray(FLOPPY_BYTES)
    img[0:512] = boot
    # The .COM goes at disk sector 2, which is where bootcom.s reads from.
    # to_file(LOAD_OFF) == 512, so this agrees with the mapping above.
    img[512:512 + len(com)] = com
    inlen_f, inbuf_f = to_file(INLEN_OFF), to_file(INBUF_OFF)
    img[inlen_f:inlen_f + 2] = len(stdin).to_bytes(2, "little")
    img[inbuf_f:inbuf_f + len(stdin)] = stdin
    return bytes(img)


def run_qemu(floppy):
    """Boot the image.  Return (output bytes, exit code or None, note)."""
    with tempfile.TemporaryDirectory() as td:
        ser = os.path.join(td, "serial.bin")
        cmd = [QEMU,
               "-drive", "file=%s,format=raw,if=floppy,index=0" % floppy,
               "-chardev", "file,id=s0,path=" + ser,
               "-serial", "chardev:s0",
               "-device", "isa-debug-exit,iobase=0x501,iosize=1",
               "-display", "none",
               "-no-reboot"]
        try:
            r = subprocess.run(cmd, capture_output=True, timeout=TIMEOUT)
        except subprocess.TimeoutExpired:
            # Whatever reached the serial port is still evidence, so keep it
            # and report the hang rather than throwing the output away.
            out = open(ser, "rb").read() if os.path.exists(ser) else b""
            return out, None, "the machine never halted (timeout %ds)" % TIMEOUT
        out = open(ser, "rb").read() if os.path.exists(ser) else b""
    # qemu's OWN code, which is the only record of whether the program reached
    # the isa-debug-exit port.  It used to be dropped and 0 returned, which
    # exit_code_of then had to read as "unknown" - so a program that ran to
    # completion and one that hung were indistinguishable here, and every
    # fixture reported exit=None.  Passing the real code through is what makes
    # "it terminated" an assertion rather than an assumption.
    return out, r.returncode, ""


def exit_code_of(rc):
    """The DOS exit code the program passed to INT 21h AH=4Ch, or None."""
    if rc is None or rc == 0 or (rc - EXIT_BASE) % 2 != 0:
        return None
    return (rc - EXIT_BASE) // 2


def ensure_comtest():
    """Return the path to comtest, rebuilding it if any source is newer.

    Timestamp-checked rather than assumed, because this project has already
    lost a session to a stale `comtest` disagreeing with a freshly built
    `compiletest`.  See the mtime comparison in tests/uitest.py's siblings.
    """
    exe = os.path.join(SHELL, "comtest")
    # The freshness list must be the files actually LINKED - the .o files - not
    # the .mod files they came from.  Watching the sources instead left a real
    # hole: `make` relinks Compiler.o from an edited Compiler.mod, and if that
    # happened after comtest was last built, comtest was older than every
    # .mod and so looked current while containing the old compiler.  The
    # failure mode is the worst kind: the .COM under test is generated by the
    # stale compiler, so a fix is invisible and a test that should have gone
    # red stays green.  The .mod files are kept in the list anyway, so editing a
    # source without running make also forces a relink.
    objs = ["TextBuf.o", "Posix.o", "Compiler.o", "Runtime.o", "Linker.o"]
    newer = objs + ["Compiler.mod", "Runtime.mod", "Linker.mod", "TextBuf.mod",
                    "Posix.c", os.path.join("tests", "ComTest.mod")]
    if os.path.exists(exe):
        t = os.path.getmtime(exe)
        if all(os.path.exists(os.path.join(SHELL, f))
               and os.path.getmtime(os.path.join(SHELL, f)) <= t
               for f in newer):
            return exe
    lst = os.path.join(SHELL, "tests", "ct.lst")
    link = ([os.path.join(SHELL, "tests", "ComTest.mod")]
            + [os.path.join(SHELL, o) for o in objs])
    subprocess.run([GM2, "-fiso", "-fgen-module-list=" + lst, "-o", "/dev/null"]
                   + link, capture_output=True, cwd=SHELL)
    r = subprocess.run([GM2, "-fiso", "-fuse-list=" + lst, "-o", exe] + link,
                       capture_output=True, cwd=SHELL)
    if r.returncode != 0:
        sys.exit("linking comtest failed:\n" + r.stderr.decode(errors="replace"))
    return exe


def emit_com(exe, pas, workdir):
    """Compile one .pas to workdir/NAME.COM.  Return (path or None, output)."""
    # ComTest reads paths from stdin and writes the .COM into its own cwd,
    # named after the fixture's base name.  So cwd is the workdir.
    r = subprocess.run([exe], input=pas.encode() + b"\n",
                       capture_output=True, cwd=workdir)
    com = os.path.join(workdir,
                       os.path.basename(pas)[:-4] + ".COM")
    if not os.path.exists(com):
        return None, r.stdout.decode(errors="replace").strip()
    return com, r.stdout.decode(errors="replace").strip()


def visible_output(raw):
    """Turn a repr-able byte string into something a terminal shows."""
    return raw.decode("latin-1").replace("\r", "\\r").replace("\n", "\\n")


def cases():
    """Every fixture that has a .out file, with its optional input."""
    fx = os.path.join(HERE, "fixtures")
    for name in sorted(os.listdir(fx)):
        if not name.endswith(".out"):
            continue
        stem = name[:-4]
        stdin = b""
        inpath = os.path.join(fx, stem + ".in")
        if os.path.exists(inpath):
            stdin = open(inpath, "rb").read()
        yield {"name": stem,
               "out": os.path.join(fx, name),
               "in": inpath if os.path.exists(inpath) else None,
               "stdin": stdin}


def main(argv):
    if "--list" in argv:
        for c in cases():
            print(c["name"])
        return 0
    show = "--show" in argv
    rebless = "--rebless" in argv
    names = [a for a in argv if not a.startswith("-")]
    all_cases = list(cases())
    if names:
        sel = [c for c in all_cases if c["name"] in names]
        missing = set(names) - {c["name"] for c in all_cases}
        if missing:
            sys.exit("no such fixture: " + ", ".join(sorted(missing)))
    else:
        sel = all_cases

    boot = build_boot_sector()
    exe = ensure_comtest()
    workdir = tempfile.mkdtemp(prefix="tpexec-")
    passed = failed = blessed = 0
    for c in sel:
        pas = os.path.join(HERE, "fixtures", c["name"] + ".pas")
        com, log = emit_com(exe, pas, workdir)
        if com is None:
            print("  %-22s FAIL  comtest wrote no .COM" % c["name"])
            if log:
                print("       %s" % log)
            failed += 1
            continue
        data = open(com, "rb").read()
        img = os.path.join(workdir, c["name"] + ".img")
        open(img, "wb").write(build_floppy(boot, data, c["stdin"]))
        got, rc, note = run_qemu(img)
        want = open(c["out"], "rb").read()

        if show:
            print("  %-22s exit=%s  %s" % (c["name"], exit_code_of(rc),
                                           visible_output(got)))
            continue
        if rc is None:
            print("  %-22s FAIL  %s" % (c["name"], note))
            print("       output so far: %s" % visible_output(got))
            failed += 1
            continue
        code = exit_code_of(rc)
        if code is None:
            print("  %-22s FAIL  qemu rc=%d, which no .COM exit can produce "
                  "(the program never called INT 21h AH=4Ch)" % (c["name"], rc))
            failed += 1
            continue
        if got != want:
            if rebless:
                open(c["out"], "wb").write(got)
                print("  %-22s REBLESSED  %s" % (c["name"], visible_output(got)))
                blessed += 1
                continue
            print("  %-22s FAIL" % c["name"])
            print("       want  %s" % visible_output(want))
            print("       got   %s" % visible_output(got))
            failed += 1
            continue
        print("  %-22s PASS  %d bytes, exit %d"
              % (c["name"], len(got), code))
        passed += 1
    if show:
        return 0
    print()
    if rebless and blessed:
        print("REBLESSED %d expectation file(s).  Check the diff: a .out that "
              "changed is a claim that was wrong, or a program that is."
              % blessed)
    print("execution: %d passed, %d failed (of %d)"
          % (passed, failed, len(sel)))
    return 1 if failed else 0


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