TP3/shell/tests/rt_exec.py

#!/usr/bin/env python3
"""rt_exec.py -- call every 8086 runtime entry with a known argument, on qemu,
and check what it does.

The runtime (shell/Runtime.mod) is hand-assembled 8086, so "it built" says
nothing about it.  This says it *runs*.  It used to run it on Unicorn, and
Unicorn 2.1.4 mis-decodes 16-bit ModRM memory operands, so it failed 33 of 33
and every one of those failures was the emulator's.  A wrong oracle is worse
than none: it cannot tell "my codegen is broken" from "the machine is broken".

So the machine is now qemu-system-i386, driven through the same boot sector the
.COM fixtures use (tests/exec/bootcom.s), and the expectations are unchanged
except where they were demonstrably wrong -- see EXPECTATION FIXES below.

    .pas-free:  RtProbe (base = RT_BASE)  -> blob + entry offsets
                rt_exec.py                 -> 16-40 byte case record
                exec/rtdrv.s (assembled)   -> a whole .COM image
                bootcom.s + qemu           -> one framed record per case
                rt_exec.py                 -> parsed, compared

    tests/rt_exec.py --probe   # the standalone runtime dump (base 0)
    tests/rt_exec.py --list    # the case names
    tests/rt_exec.py --show    # print what came out, assert nothing

EXPECTATION FIXES -- two expectations were wrong, and both were wrong in the
direction that made the HARNESS the suspect:

  * rdchar stores ONE byte (Runtime.EmitRdChar: `StDiDl' = MOV [DI], DL), so
    dumping a two-byte word and comparing it against ord(c) could never pass.
    The check was broken, the entry was right.  The case now reports two bytes
    and expects the character followed by the 0xEE poison that is still there.

  * rdint at end of input does NOT store anything.  TP3's xrdint returns to
    rnerr without touching the variable, and Runtime.EmitRdInt says so in its
    own comment ("^Z before any digit reaches rnend ... returns WITHOUT
    touching the variable").  The old expectation said the variable became 0.
    It stays at the poison, and that is the faithful answer.

Everything else is the original expectation verbatim.  The two `rdln' halves of
the six rdint cases are merged into the same case as the value they follow: a
readln that printed anything would change the same compared string, so the
assertion is unchanged - one qemu boot per check instead of two.
"""

import os
import re
import struct
import subprocess
import sys
import tempfile

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from run_com_exec import (GM2, QEMU, build_boot_sector, build_floppy,  # noqa: E402
                          run_qemu, visible_output)

HERE = os.path.dirname(os.path.abspath(__file__))
SHELL = os.path.dirname(HERE)

# --------------------------------------------------------------- the layout
# RT_BASE is restated from exec/rtdrv.s on purpose: the driver needs the blob at
# a known offset and Python needs to know where the driver thinks it is, so the
# two are written down and then CHECKED against each other (see build_image).
# A restated constant that is never compared is a constant waiting to lie.
RT_BASE = 0x0200
HDR = 0x0800
DLEN = 32
DATA = 0x0810
STORE = 0x0840
CASE = 0x0860
CASE_SZ = 40

# A .COM is loaded at CS:0100, and every address in the image is an image
# offset.  The load bias is therefore the difference between the two, and
# putting it in one place here stops it being applied -- or forgotten -- in each
# field.  exec/rtdrv.s has its own .set of it, for the same reason it has its
# own .set of the record offsets: two independent statements that are then
# checked against each other by the fact that any disagreement stops the cases
# from passing.
LOAD_BIAS = 0x0100

# The 8086 has a fixed 16-byte-case-record area in rtdrv.s, and the harness
# asserts this window is still zeros.  If the driver outgrows its space it
# starts eating the runtime it is testing, which would be silent; this makes it
# loud.
DRIVER_TOP = 0x0180

# The four framing bytes exec/rtdrv.s wraps each case's report in.
SOH, STX, ETX, EOT = 0x01, 0x02, 0x03, 0x04

# Entry selector order, as Runtime.def declares it.  The names are used only to
# be looked up in RtProbe's output, so renaming an entry in Runtime.mod makes
# the lookup fail loudly instead of silently testing offset 0.
ENTS = ["initmem", "progend", "stackchk", "wrint", "wrchar", "wrbool",
        "wrreal", "wrln", "rdint", "rdchar", "rdbool", "rdln", "halt",
        "wrtinl"]

# A call slot's flags, as rtdrv.s reads them.
F_STACK = 1     # the argument goes on the stack, popped by the caller
F_AX = 2        # ...or in AX instead (initmem's convention)
F_INL = 4       # wrtinl: the argument is at the return address, not on the stack

# The byte patterns check_bp_contract looks for, and the rule they express:
# "an entry that borrows BP opens with PUSH BP; MOV BP,SP and pops it again
# before it returns" -- so an entry containing 8B EC anywhere must start with
# 55 8B EC and must contain a 5D.  See that function: the first half of that
# rule is exact and the second is a screen, and the difference matters.
PUSH_BP_MOV_BP_SP = b"\x55\x8b\xec"   # PUSH BP ; MOV BP,SP
MOV_BP_SP = b"\x8b\xec"                # MOV BP,SP
POP_BP = b"\x5d"                       # POP BP


# --------------------------------------------------------------- RtProbe
def ensure_rtprobe():
    """Build tests/RtProbe.mod, return its path.

    Timestamp-checked, for the reason run_com_exec.ensure_comtest documents:
    a probe older than the runtime it reports on is worse than no probe, and
    the failure it produces -- a blob with the wrong addresses in it -- is
    silent at the byte level and catastrophic at the behaviour level.
    """
    exe = os.path.join(HERE, "rtprobe")
    newer = ["Runtime.mod", "Runtime.def", "Runtime.o", "Posix.c",
             os.path.join("tests", "RtProbe.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
    r = subprocess.run([GM2, "-fiso", "-o", exe, "tests/RtProbe.mod",
                        "Runtime.o", "Posix.o"],
                       capture_output=True, cwd=SHELL)
    if r.returncode != 0:
        sys.exit("building RtProbe failed:\n" + r.stderr.decode(errors="replace"))
    return exe


def load_runtime(base):
    """Run RtProbe at `base`.

    Return (blob, {name: image-absolute offset}, code_end).  `code_end` is the
    blob-relative offset where the code stops and the data area starts, and it
    is what bounds the last entry's bytes: without it the last entry's "does
    it contain MOV SP,BP" question would be answered partly by the data area,
    and a string or an entry-table word that happens to contain the pattern
    would turn a missing epilogue into a pass.
    """
    out = subprocess.run([ensure_rtprobe()], input=b"%d\n" % base,
                         capture_output=True, check=True).stdout.decode(
                             errors="replace")
    # RtProbe ends its lines CR LF, as a program of this vintage would.  Every
    # pattern below is `$'-anchored, and Python's `$' in MULTILINE stops before
    # "\n" only -- so without this the first regex fails and the report blames
    # the probe for it.
    out = out.replace("\r\n", "\n")
    m = re.search(r"^base=(\d+)$", out, re.M)
    if m is None or int(m.group(1)) != base:
        sys.exit("RtProbe did not confirm base=%d:\n%s" % (base, out[:400]))
    size = int(re.search(r"^(\d+) bytes", out, re.M).group(1))
    cm = re.search(r"^code ends at (\d+)$", out, re.M)
    if cm is None:
        sys.exit("RtProbe did not say where the code ends:\n%s" % out[:400])
    code_end = int(cm.group(1))
    if not 0 < code_end <= size:
        sys.exit("RtProbe says the code ends at %d, which is not inside a "
                 "%d byte blob" % (code_end, size))
    entries = {nm: int(v) for v, nm in
               re.findall(r"entry \d+ = (\d+)\s+\((\w+)\)", out)}
    missing = [e for e in ENTS if e not in entries]
    if missing:
        sys.exit("RtProbe did not report entries: %s" % ", ".join(missing))
    blob = bytearray()
    for line in out.splitlines():
        h = re.match(r"^[0-9A-F]{8}  ((?:[0-9A-F]{2} )+)$", line)
        if h:
            blob += bytes.fromhex(h.group(1).replace(" ", ""))
    if len(blob) != size:
        sys.exit("parsed %d bytes of runtime, RtProbe says %d"
                 % (len(blob), size))
    for nm, off in entries.items():
        if not base <= off < base + code_end:
            sys.exit("entry %s is at %d, outside the code area %d..%d -- the "
                     "entry table and the blob disagree"
                     % (nm, off, base, base + code_end))
    return bytes(blob), entries, code_end


def probe():
    """The standalone dump run_all.sh's `runtime probe' step wants: base 0."""
    raw = subprocess.run([ensure_rtprobe()], input=b"0\n", capture_output=True,
                         check=True).stdout
    sys.stdout.buffer.write(raw)
    return 0


def check_bp_contract(blob, entries, base, code_end):
    """Every entry must hand BP back.

    The driver keeps its cursor into the case record in BP, because an entry is
    allowed to destroy every register except BP and SP.  Two entries did not
    honour that: wrchar and wrbool borrowed BP to reach their argument -- [SP]
    is unencodable in 16-bit mode -- and returned without saving it.  One
    multi-call case caught it, and only because it made four calls in a row; a
    new entry that borrowed BP the same way would be caught by nothing.

    So the rule is checked here, over the built blob, before any case runs.

    The rule is the runtime's own convention, in two halves, and the halves are
    NOT equally strong -- which is worth writing down rather than discovering:

      * "an entry that borrows BP must START with 55 8B EC" is EXACT.  The
        prologue is the first thing the entry does, so a positional test
        cannot be fooled by anything.

      * "...and must contain a 5D somewhere" is a SCREEN, not a proof.  5D is
        POP BP, but it is also a perfectly ordinary displacement byte, and
        this code does not disassemble.  Requiring the POP to be contiguous
        with something else does not rescue it either: wrbool legitimately
        closes its frame with POP BP after the INT 21h, and a rule that
        insisted on 89 EC 5D would have failed a correct entry -- a false
        alarm, which is worse here than a miss because it teaches a reader to
        distrust the check.  So the honest position is: a push with no pop is
        CAUGHT unless some displacement in the entry happens to be 5Dh, and
        the argument displacements in this runtime are 2 and 4, so nothing can
        collide today.  tests/nonvacuity.sh removes the POP as well as the
        PUSH and requires each to turn this red, so the screen is at least
        witnessed rather than assumed.

    Counting 55s and 5Ds to check frame BALANCE would be worse than either:
    55 occurs as an immediate and 5D as a displacement, so the answer would
    come out wrong without the check ever looking wrong.  For the same reason
    each entry's bytes stop at the next entry, or at `code_end` for the last
    one: running on into the data area would let a string satisfy either test.

    The counts it returns are the point as much as the verdict.  A scan that
    matches nothing is indistinguishable from a scan that passes, and that has
    happened here before -- so zero borrowers is a failure, not a pass.

    Entries are counted by OFFSET, not by name: `progend' and `halt' are one
    entry with two names, and counting them twice would make the total here
    disagree with the number of things that can actually be wrong.
    """
    offs = sorted(set(entries.values()))
    scanned = borrowed = 0
    bad = []
    for i, a in enumerate(offs):
        lo = a - base
        hi = (offs[i + 1] - base) if i + 1 < len(offs) else code_end
        code = bytes(blob[lo:hi])
        scanned += 1
        if MOV_BP_SP not in code:
            continue
        borrowed += 1
        # The two halves are reported separately, because they are separate
        # faults with separate fixes and "it does not hand BP back" does not
        # say which of them happened.
        if not code.startswith(PUSH_BP_MOV_BP_SP):
            why = "borrows BP but does not open with 55 8B EC, so the "\
                  "caller's BP is destroyed"
        elif POP_BP not in code:
            why = "pushes BP but never pops it, so the caller gets a "\
                  "cursor two bytes off"
        else:
            continue
        names = sorted(n for n, v in entries.items() if v == a)
        bad.append("%s -- %s (image 0x%04X: %s)"
                   % (",".join(names), why, a, code[:12].hex(" ")))
    if bad:
        sys.exit("these entries do not hand BP back, so a caller that keeps a "
                 "cursor in BP -- which this driver does, and which is the whole "
                 "reason BP is the one register an entry may keep -- loses it:\n"
                 "  %s" % "\n  ".join(bad))
    if borrowed == 0:
        sys.exit("no entry borrows BP, so this check examined %d entries and "
                 "matched nothing: it is not looking at what it claims to "
                 "look at" % scanned)
    return scanned, borrowed


# ------------------------------------------------------------------ the case
def case_bytes(c, entries):
    """One 40 byte case record, exactly as exec/rtdrv.s reads it.

    The record is a second, independent statement of the same layout rtdrv.s
    carries in its .set block.  The cases only pass if the two agree on every
    offset, which is the whole reason for writing it twice: a driver that
    silently read the wrong field would still boot, still run, and still print.
    """
    if len(c["in"]) > 8:
        sys.exit("case %s supplies %d input bytes; the record holds 8"
                 % (c["name"], len(c["in"])))
    b = bytearray(CASE_SZ)
    struct.pack_into("<H", b, 0, c["dump"])             # OFF_DUMPLEN
    struct.pack_into("<H", b, 2, c["dumpadr"])          # OFF_DUMPADR
    b[4] = len(c["in"])                                 # OFF_INLEN
    b[5:5 + len(c["in"])] = c["in"]                     # OFF_INBUF
    b[13] = c["index"]                                  # OFF_CASE
    calls = c["calls"]
    if len(calls) > 4:
        sys.exit("case %s has %d calls; the record holds 4"
                 % (c["name"], len(calls)))
    struct.pack_into("<H", b, 14, len(calls))           # OFF_NCALLS
    for i, (ent, arg, mode) in enumerate(calls):
        o = 16 + 6 * i                                  # SLOT0
        struct.pack_into("<H", b, o, entries[ent])
        struct.pack_into("<H", b, o + 2, arg)
        b[o + 4] = mode
    return bytes(b)


def build_image(blob, rec):
    """Assemble exec/rtdrv.s into a complete .COM, and check it is the one
    we think it is.

    The three assertions are the point of this function.  An image that is
    subtly not the image we intended would still boot, still run, and still
    print something -- the failure would look like a runtime bug.
    """
    if RT_BASE + len(blob) > HDR:
        sys.exit("the runtime is %d bytes: at RT_BASE=%04X it runs into the "
                 "test scaffolding at %04X.  Move the .org's in rtdrv.s."
                 % (len(blob), RT_BASE, HDR))
    with tempfile.TemporaryDirectory() as td:
        open(os.path.join(td, "rtblob.bin"), "wb").write(blob)
        open(os.path.join(td, "rtcase.bin"), "wb").write(rec)
        src = os.path.join(HERE, "exec", "rtdrv.s")
        obj, raw = os.path.join(td, "r.o"), os.path.join(td, "r.bin")
        r = subprocess.run(["as", "--32", "-I", td, "-o", obj, src],
                           capture_output=True)
        if r.returncode != 0:
            sys.exit("as failed on rtdrv.s:\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"))
        img = open(raw, "rb").read()

    # E9 00 00: a near jump to offset 3 from offset 0, so the displacement is
    # 3 - (0 + 3) = 0.  The driver is the first thing after the jump, which is
    # what makes the rel16 zero and therefore useless as a check on the target
    # -- the check that matters is the one below, that the blob landed where the
    # driver thinks it did.
    if img[0:3] != b"\xe9\x00\x00":
        sys.exit("rtdrv.s does not open with a 3-byte near JMP over itself: %s"
                 % img[0:3].hex())
    if img[RT_BASE:RT_BASE + len(blob)] != blob:
        sys.exit("the blob is not at RT_BASE=%04X in the assembled image -- "
                 "rtdrv.s and rt_exec.py disagree about the layout"
                 % RT_BASE)
    gap = img[DRIVER_TOP:RT_BASE]
    if gap != b"\x00" * len(gap):
        sys.exit("the driver has outgrown its space: rtdrv.s no longer fits "
                 "under %04X, so it is overwriting the runtime" % DRIVER_TOP)
    return img


def parse_record(raw, want_index):
    """Split the serial output into (case index, printed, dumped) or raise.

    Nothing here is lenient.  A driver that printed nothing, a truncated
    record, or a record for the wrong case are all failures with a reason, not
    a comparison against an empty string that happens to pass.
    """
    if not raw or raw[0] != SOH:
        raise AssertionError("no record header: the driver printed %r" % raw[:64])
    if len(raw) < 5 or raw[3] != STX:
        raise AssertionError("malformed record header %r" % raw[:16])
    idx = int(raw[1:3], 16)
    if idx != want_index:
        raise AssertionError("record is for case %02X, expected %02X"
                             % (idx, want_index))
    if raw[-1] != EOT:
        raise AssertionError("record was never closed (EOT): the last call did "
                             "not return, or the machine halted early")
    end = raw.index(ETX, 4)
    if end < 0:
        raise AssertionError("record has no ETX: the calls printed nothing and "
                             "the driver never got past them")
    printed, dump = raw[4:end], raw[end + 1:-1]
    if len(dump) % 2 or not re.fullmatch(b"[0-9A-F]*", dump):
        raise AssertionError("the memory report %r is not hex" % dump)
    return idx, printed, bytes.fromhex(dump.decode())


# ----------------------------------------------------------------- the cases
def build_cases():
    """Every check, in the order the names are printed.

    Each case is a list of calls plus two expectations: the bytes the entry(s)
    printed, and the bytes that must be at `dumpadr' afterwards.  `dump=0' means
    the case asserts nothing about memory, which is itself an assertion: the
    report is empty, so a case cannot quietly acquire a memory check it did not
    ask for.

    The addresses handed to an entry as its argument are MEMORY addresses, not
    image offsets -- that is the whole of LoadBias, and it is the difference
    between a read entry storing where the harness looks and storing one page
    lower where nothing checks it.  V below is "the memory address of STORE",
    which is what every read case wants.
    """
    cs = []
    V = STORE + LOAD_BIAS

    def add(name, calls, out=b"", stdin=b"", dump=0, dumpbytes=b"",
            dumpadr=STORE, note=""):
        cs.append({"name": name, "calls": calls, "out": out, "in": stdin,
                   "dump": dump, "dumpbytes": dumpbytes, "dumpadr": dumpadr,
                   "note": note, "index": len(cs)})

    def s(ent, arg):
        return (ent, arg, F_STACK)

    def x(ent, arg):
        return (ent, arg, F_AX)

    def n(ent):
        return (ent, 0, 0)

    # initmem is handed the header in AX -- and the header's MEMORY address, not
    # its image offset.  Given the image offset it reads the two words it wants
    # out of blank space, finds both zero, and clears nothing at all: a silent
    # no-op, which is the reason this case poisons the data area to AAH first.
    add("initmem zeroes globals",
        [x("initmem", HDR + LOAD_BIAS)], dump=DLEN, dumpadr=DATA,
        dumpbytes=b"\x00" * DLEN,
        note="the data area is poisoned to AA first, so a zero is a result")

    for v, why in [(0, "the one value where a signed/unsigned slip is invisible"),
                   (1, ""), (-1, ""), (7, ""), (10, ""), (42, ""),
                   (100, ""), (12345, ""), (-32768, "the sign bit"), (32767, "")]:
        add("wrint(%d)" % v, [s("wrint", v & 0xFFFF)],
            out=str(v).encode(), note=why)

    add("wrchar('A')", [s("wrchar", ord("A"))], out=b"A")
    add("wrchar('!')", [s("wrchar", ord("!"))], out=b"!")

    add("wrbool(0)", [s("wrbool", 0)], out=b"FALSE")
    add("wrbool(1)", [s("wrbool", 1)], out=b"TRUE")
    add("wrbool(2)", [s("wrbool", 2)], out=b"TRUE",
        note="any non-zero is true, as TP3's xwrb does")

    add("wrln", [n("wrln")], out=b"\r\n")

    add("stackchk returns", [n("stackchk")], out=b"",
        note="no output: the assertion is that control came back at all")

    add("writeln(42) writeln TRUE in one program",
        [s("wrint", 42), s("wrchar", ord(" ")), s("wrbool", 1), n("wrln")],
        out=b"42 TRUE\r\n",
        note="four calls back to back, so no register leaks between them")

    # rdint: the value, and then the rdln that must eat the rest of the line.
    for text, want in [(b"  42abc", 42), (b"-17 x", -17), (b"+5", 5),
                       (b"0", 0), (b"  007", 7), (b"1234", 1234)]:
        add("rdint(%r) then rdln" % text,
            [s("rdint", V), n("rdln")], out=b"", stdin=text, dump=2,
            dumpbytes=struct.pack("<h", want),
            note="the delimiter must be left for rdln, which prints nothing")

    for c in "Q7":
        add("rdchar(%r)" % c, [s("rdchar", V)], out=b"", stdin=c.encode(),
            dump=2, dumpbytes=bytes([ord(c), 0xEE]),
            note="rdchar stores ONE byte, so the poison is still in byte 2")

    for c, want in [("T", 1), ("y", 1), ("1", 1), ("F", 0), ("n", 0), ("0", 0)]:
        add("rdbool(%r)" % c, [s("rdbool", V)], out=b"", stdin=c.encode(),
            dump=2, dumpbytes=struct.pack("<H", want))

    add("rdint at end of input leaves the variable alone",
        [s("rdint", V), n("rdln")], out=b"", stdin=b"", dump=2,
        dumpbytes=b"\xee\xee",
        note="TP3's xrdint returns to rnerr without storing; the poison stays")

    add("wrtinl reads its length and characters from the return address",
        [("wrtinl", 0, F_INL)], out=b"hello",
        note="the one entry whose argument is not a stack word")

    return cs


def run_case(c, blob, entries, boot):
    rec = case_bytes(c, entries)
    img = build_image(blob, rec)
    with tempfile.TemporaryDirectory() as td:
        floppy = os.path.join(td, "f.img")
        open(floppy, "wb").write(build_floppy(boot, img, c["in"]))
        return run_qemu(floppy)


def main(argv):
    if "--probe" in argv:
        return probe()
    cases = build_cases()
    if "--list" in argv:
        for c in cases:
            print(c["name"])
        return 0
    show = "--show" in argv
    only = [a for a in argv if not a.startswith("-")]

    blob, entries, code_end = load_runtime(RT_BASE)
    boot = build_boot_sector()

    # Before any machine is started: the contract the driver depends on.
    nscan, nborrow = check_bp_contract(blob, entries, RT_BASE, code_end)
    print("  %-52s PASS   (%d entries scanned, %d borrow BP and hand it "
          "back)" % ("every entry preserves BP", nscan, nborrow))
    npass = 1
    nfail = 0
    for c in cases:
        if only and not any(o in c["name"] for o in only):
            continue
        # A marker byte inside an expectation would make the framing
        # ambiguous, so it is refused rather than mis-parsed.
        for b in (SOH, STX, ETX, EOT):
            if bytes([b]) in c["out"]:
                sys.exit("case %s: its expected output contains the framing "
                         "byte %02X, which would be indistinguishable from the "
                         "driver's own markers" % (c["name"], b))
        raw, rc, note = run_case(c, blob, entries, boot)
        if show:
            print(c["name"], "->", repr(visible_output(raw)))
            continue
        try:
            _, printed, dumped = parse_record(raw, c["index"])
        except AssertionError as e:
            print("  %-52s FAIL  %s" % (c["name"], e))
            nfail += 1
            continue
        if printed != c["out"]:
            print("  %-52s FAIL  printed %s, want %s"
                  % (c["name"], visible_output(printed),
                     visible_output(c["out"])))
            nfail += 1
            continue
        if dumped != c["dumpbytes"]:
            print("  %-52s FAIL  +%04X = %s, want %s"
                  % (c["name"], c["dumpadr"], dumped.hex() or "-",
                     c["dumpbytes"].hex() or "-"))
            nfail += 1
            continue
        print("  %-52s PASS%s" % (c["name"],
                                  ("   (" + c["note"] + ")") if c["note"] else ""))
        npass += 1
    if show:
        return 0
    print("\nruntime entries: %d passed, %d failed (of %d)"
          % (npass, nfail, npass + nfail))
    return 1 if nfail else 0


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