LithosAnanake/v4/tools/mkpost.py

#!/usr/bin/env python3
"""mkpost.py -- write capsules/v4/post79.4th from v3's POST cases.

    v4/tools/mkpost.py [--v3 PATH] [--report FILE]

A development tool, run when the cases change; its output is committed and
reviewed like any source.  docs/v4.0.0/NUCLEUS.md section 6.

What it does:

  1. Reads v3's case tables, v3/src/test_runner/modules/*.c, in the order
     v3's own POST runs them.  A case is a word, a name, a line of FORTH and
     whether an error is expected.
  2. Keeps the cases for words of the FORTH-79 Required Word Set.
  3. Runs every kept case, in order, in one session of the hosted v3 binary,
     and records what v3 did: an error or not, the data stack, and what was
     printed.  v3 is the reference.
  4. Writes the harness and one case after another as a .4th capsule:

         T{ DUP.basic            name the case, start clean
         T| 5 DUP . .            a line of the case, its output captured
         -> 5 5 }T 0 0 T=        v3's stack; v3's output: length, checksum

     or, for a case v3 ended with an error,  ->ERR  in place of the last
     line.

A case is left out, and listed in the report, when: its word reads the
keyboard; it forgets a word of the system or damages it on purpose; v3 could
not finish it in one
session; or one of its lines cannot be cut to fit a 64-character block
line.
"""
import argparse
import os
import re
import subprocess
import sys

ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", ".."))
MODULE_DIR = os.path.join(ROOT, "v3", "src", "test_runner", "modules")
OUTPUT = os.path.join(ROOT, "capsules", "v4", "post79.4th")
FIRST_BLOCK = 7000
LINE_MAX = 64

# v3's POST order (v3/src/test_runner/test_runner.c, test_modules[]), the
# modules that test FORTH-79 words.
MODULES = [
    "stack_words_test.c", "return_stack_words_test.c", "memory_words_test.c",
    "arithmetic_words_test.c", "logical_words_test.c",
    "mixed_arithmetic_words_test.c", "double_words_test.c",
    "format_words_test.c", "string_words_test.c", "io_words_test.c",
    "block_words_test.c", "dictionary_words_test.c",
    "dictionary_manipulation_words_test.c", "vocabulary_words_test.c",
    "system_words_test.c", "defining_words_tests.c", "control_words_test.c",
]

# The FORTH-79 Standard's Required Word Set, by the standard's own groups.
# Written from the standard's word list; check it against the document
# before relying on it for a claim of compliance.
REQUIRED = """
DUP DROP SWAP OVER ROT PICK ROLL ?DUP DEPTH >R R> R@
< = > 0< 0= 0> D< U< NOT
+ - * / MOD /MOD */ */MOD 1+ 1- 2+ 2- MAX MIN ABS NEGATE D+ DNEGATE
U* U/MOD AND OR XOR
@ ! C@ C! ? +! MOVE CMOVE FILL
DO LOOP +LOOP I J LEAVE IF ELSE THEN BEGIN UNTIL WHILE REPEAT EXIT EXECUTE
CR EMIT SPACE SPACES TYPE COUNT -TRAILING KEY EXPECT QUERY WORD
BASE DECIMAL . U. CONVERT <# # #S HOLD SIGN #>
LIST LOAD SCR BLOCK UPDATE BUFFER SAVE-BUFFERS EMPTY-BUFFERS
: ; VARIABLE CONSTANT VOCABULARY CREATE DOES>
CONTEXT CURRENT FORTH DEFINITIONS ' FIND FORGET
, ALLOT ." IMMEDIATE LITERAL STATE [ ] COMPILE [COMPILE]
( >IN ABORT QUIT HERE PAD BLK 79-STANDARD
""".split()

# Words whose cases wait for the keyboard, which a boot cannot type at.
KEYBOARD = {"KEY", "EXPECT", "QUERY"}

# ---- the harness: blocks FIRST_BLOCK up, FORTH-79 and the two nucleus hooks ----
HARNESS = r"""
( post79.4th -- POST for the FORTH-79 Required Word Set on v4. )
( Generated by v4/tools/mkpost.py from v3's POST cases: do not )
( edit.  docs/v4.0.0/NUCLEUS.md section 6.                     )
VARIABLE T#     VARIABLE TPASS  VARIABLE TFAIL  VARIABLE TOPEN
VARIABLE TBAD   VARIABLE TWANT  VARIABLE TLEN   VARIABLE TSUM
VARIABLE TDEP   VARIABLE TBASE
CREATE TBUF 64 ALLOT  CREATE TSTK 16 ALLOT  CREATE TNAME 8 ALLOT
0 T# !  0 TPASS !  0 TFAIL !  0 TOPEN !  10 TBASE !
: TB ( -- baddr ) TBUF 4 * ;
: TN ( -- baddr ) TNAME 4 * ;
( T-EMIT is given every character a case prints: the first 255 )
( are kept, all are counted and summed. )
: T-EMIT ( c -- )
  DUP TSUM @ 31 * + 16777215 AND TSUM !
  TLEN @ 255 < IF TB TLEN @ + C! ELSE DROP THEN  1 TLEN +! ;
=====
: T-CLEAR ( i*x -- ) BEGIN DEPTH WHILE DROP REPEAT ;
: T. ( n -- ) 0 <# #S #> TYPE ;
( what a failing case printed, and the stack it left )
: T-SHOW ( -- )
  ."  out<" TB TLEN @ 255 MIN TYPE ." > stack<"
  TDEP @ BEGIN DUP WHILE 1- DUP TSTK + @ . REPEAT DROP ." >" ;
( the verdict on the case that is open, if one is )
: T-CLOSE ( -- )
  TOPEN @ IF
    (CATCH) @ -1 = TWANT @ = 0= IF 1 TBAD ! THEN
    TBAD @ IF 1 TFAIL +!
      ." POST FAIL: " TN COUNT TYPE T-SHOW CR
    ELSE 1 TPASS +! THEN
    0 TOPEN !  0 (CATCH) !  TBASE @ BASE !
  THEN ;
=====
( T{ name   begin a case: clean stack, errors caught )
: T{ ( i*x -- ) T-CLOSE T-CLEAR
  BL WORD COUNT 31 MIN DUP TN C!  TN 1+ SWAP CMOVE
  1 T# +!  1 TOPEN !  0 TBAD !  0 TWANT !  0 TLEN !  0 TSUM !
  0 TDEP !  1 (CATCH) ! ;
( T| ...    a line of the case, what it prints captured )
: T| ' T-EMIT (EMIT-HOOK) ! ; IMMEDIATE
: T-END ( -- ) 0 (EMIT-HOOK) !  BASE @ TBASE !  DECIMAL ;
( -> ...    the case is over: set its stack aside, top first )
: -> ( i*x -- ) T-END 0
  BEGIN DEPTH 1 > WHILE
    DUP 16 < IF SWAP OVER TSTK + ! ELSE SWAP DROP 1 TBAD ! THEN
    1+ REPEAT
  16 MIN TDEP ! ;
( ->ERR     the case is over, and must have ended in an error )
: ->ERR ( i*x -- ) T-END  -1 TWANT !  T-CLEAR ;
=====
( ... }T    the stack the case must have left, in order )
: }T ( i*x -- )
  DEPTH TDEP @ = 0= IF 1 TBAD ! T-CLEAR EXIT THEN
  0 BEGIN DEPTH 1 > WHILE
    SWAP OVER TSTK + @ = 0= IF 1 TBAD ! THEN 1+ REPEAT DROP ;
( len sum T=   what the case must have printed )
: T= ( len sum -- ) TSUM @ = 0= IF 1 TBAD ! THEN
  TLEN @ = 0= IF 1 TBAD ! THEN ;
( the tally; a failure is an error, which ends the boot )
: T-REPORT ( -- ) T-CLOSE  0 (CATCH) !  DECIMAL
  ." PARITY:V4_POST tests=" T# @ T. ."  pass=" TPASS @ T.
  ."  fail=" TFAIL @ T. CR
  TFAIL @ IF -1 NODE-ERROR ! THEN ;
"""


# ---- reading v3's case tables ----------------------------------------------

TOKEN = re.compile(r'''
    \s+ | /\*.*?\*/ | //[^\n]* |
    (?P<str>"(?:\\.|[^"\\])*") |
    (?P<punct>[{}(),;=\[\]&*]) |
    (?P<word>[A-Za-z0-9_.+\-]+)
''', re.S | re.X)

ESCAPES = {"n": "\n", "t": "\t", "\\": "\\", '"': '"', "r": "\r", "0": "\0"}


def unescape(lit):
    out, i, body = [], 0, lit[1:-1]
    while i < len(body):
        if body[i] == "\\" and i + 1 < len(body):
            out.append(ESCAPES.get(body[i + 1], body[i + 1]))
            i += 2
        else:
            out.append(body[i])
            i += 1
    return "".join(out)


def tokens(text):
    pos, out = 0, []
    while pos < len(text):
        m = TOKEN.match(text, pos)
        if not m:
            pos += 1
            continue
        pos = m.end()
        if m.group("str"):
            value = unescape(m.group("str"))
            if out and out[-1][0] == "str":         # adjacent literals join
                out[-1] = ("str", out[-1][1] + value)
            else:
                out.append(("str", value))
        elif m.group("punct"):
            out.append(("punct", m.group("punct")))
        elif m.group("word"):
            out.append(("word", m.group("word")))
    return out


def parse_braces(toks, i):
    """toks[i] is '{'; returns (nested list, index after the matching '}')."""
    items = []
    i += 1
    while toks[i] != ("punct", "}"):
        if toks[i] == ("punct", "{"):
            sub, i = parse_braces(toks, i)
            items.append(sub)
        elif toks[i] == ("punct", ","):
            i += 1
        else:
            items.append(toks[i][1] if toks[i][0] != "str" else ("s", toks[i][1]))
            i += 1
    return items, i + 1


def read_module(path):
    """Yields (word, case name, input, should_error) for every implemented case."""
    toks = tokens(open(path, encoding="utf-8", errors="replace").read())
    for i, t in enumerate(toks):
        if t != ("word", "WordTestSuite"):
            continue
        j = i
        while j < len(toks) and toks[j] != ("punct", "=") and toks[j] != ("punct", ";") and toks[j] != ("punct", "("):
            j += 1
        if j >= len(toks) or toks[j] != ("punct", "="):
            continue
        suites, _ = parse_braces(toks, j + 1)
        for suite in suites:
            if not isinstance(suite, list) or not suite or not isinstance(suite[0], tuple):
                continue
            word = suite[0][1]
            for case in suite[1]:
                if not isinstance(case, list) or len(case) < 6 or not isinstance(case[0], tuple):
                    continue
                name, text = case[0][1], case[1][1]
                should_error, implemented = case[4] != "0", case[5] != "0"
                if implemented:
                    yield word, name, text, should_error


# ---- cutting a case's line to block lines ------------------------------------

STRING_OPENERS = {'."', 'S"', 'ABORT"', '.('}


def atoms(text):
    """The line as pieces that must each stay on one line: a word, or a word
    that takes text together with its text.  None if the line has a new-line
    inside such a text."""
    out, i, n = [], 0, len(text)
    while i < n:
        if text[i].isspace():
            i += 1
            continue
        j = i
        while j < n and not text[j].isspace():
            j += 1
        word = text[i:j]
        upper = word.upper()
        if upper in STRING_OPENERS or upper == "(":
            close = ")" if upper in ("(", ".(") else '"'
            k = text.find(close, j + 1 if j < n else j)
            if k < 0:
                k = n - 1
            piece = text[i:k + 1]
            if "\n" in piece:
                return None
            out.append(piece)
            i = k + 1
        elif word == "\\":
            break                                   # the rest of the line is a comment
        else:
            out.append(word)
            i = j
    return out


def pack(prefix, pieces, width=LINE_MAX):
    """Pieces into lines no longer than `width`, each begun with `prefix`
    (which may be empty).  None if a piece cannot fit on a line."""
    head = [prefix] if prefix else []
    lines, current = [], []
    for piece in pieces:
        if len(" ".join(head + [piece])) > width:
            return None
        if current and len(" ".join(head + current + [piece])) > width:
            lines.append(current)
            current = []
        current.append(piece)
    if current:
        lines.append(current)
    return [" ".join(head + line) for line in lines]


# A store to the search order, which v3's C test runner survives and a
# running FORTH system does not.
DAMAGES = re.compile(r"\b(CONTEXT|CURRENT)\s+!")


def forgets_system_word(text):
    """True if the line FORGETs a word it did not itself define.  In a plain
    v3 session that takes the rest of the system with it, and v4 refuses."""
    words = text.split()
    for i, word in enumerate(words[:-1]):
        if word.upper() == "FORGET":
            target = words[i + 1]
            made = any(words[k] == target and k > 0 and words[k - 1].upper() in
                       (":", "CREATE", "VARIABLE", "CONSTANT", "VOCABULARY") for k in range(i))
            if not made:
                return True
    return False


# ---- what v3 does --------------------------------------------------------------

ANSI = re.compile(r"\x1b\[[0-9;]*m")


def run_v3(v3, cases):
    """Runs the cases in one session.  Returns a list, one entry per case:
    None if v3 did not get through it, else (error, stack, output)."""
    script = [": T| ; IMMEDIATE", ": TCLR BEGIN DEPTH WHILE DROP REPEAT ;"]
    for n, case in enumerate(cases):
        script.append('TCLR ." [[B%d]]"' % n)
        script.extend(case["lines"])
        script.append('." [[E%d]]" BASE @ DECIMAL .S BASE ! ." [[S%d]]"' % (n, n))
    script.append("BYE")
    done = subprocess.run([v3, "--log-none"], input="\n".join(script) + "\n", capture_output=True,
                          text=True, errors="replace", timeout=600)
    text = ANSI.sub("", done.stdout)
    results = []
    for n, case in enumerate(cases):
        begin, mid, end = "[[B%d]] ok\nok> " % n, "ok> [[E%d]]" % n, "[[S%d]]" % n
        b = text.find(begin)
        e = text.find(mid, b) if b >= 0 else -1
        s = text.find(end, e) if e >= 0 else -1
        if b < 0 or e < 0 or s < 0:
            results.append(None)
            continue
        body = text[b + len(begin):e]
        # one "ok> " and one " ok" or " ERROR" for every line of the case
        chunks = body.split("ok> ")
        output, error, ok = "", False, True
        for chunk in chunks:
            if chunk.endswith(" ERROR\n"):
                output += chunk[:-len(" ERROR\n")]
                error = True
            elif chunk.endswith(" ok\n"):
                output += chunk[:-len(" ok\n")]
            else:
                ok = False
        m = re.match(r"<(\d+)>((?: -?\d+)*) \n", text[e + len(mid):s])
        if not ok or not m:
            results.append(None)
            continue
        stack = [int(v) for v in m.group(2).split()]
        results.append((error, stack[:-1], output))   # the last value is BASE
    return results


def checksum(output):
    total = 0
    for ch in output.encode("latin-1", errors="replace"):
        total = (total * 31 + ch) & 0xFFFFFF
    return total


# ---- writing the capsule -------------------------------------------------------

def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--v3", default=os.path.join(ROOT, "lfs", "amd64", "starforth"))
    ap.add_argument("--report", default=None)
    args = ap.parse_args()

    required, left_out, cases, seen = set(REQUIRED), [], [], {}
    for module in MODULES:
        for word, name, text, should_error in read_module(os.path.join(MODULE_DIR, module)):
            if word not in required:
                continue
            label = ("%s.%s" % (word, name))[:31]
            seen[label] = seen.get(label, 0) + 1
            if seen[label] > 1:
                label = (label[:28] + "~%d" % seen[label])[:31]
            if word in KEYBOARD:
                left_out.append((label, "reads the keyboard"))
                continue
            if forgets_system_word(text):
                left_out.append((label, "forgets a word of the system"))
                continue
            if DAMAGES.search(text.upper()):
                left_out.append((label, "damages the system on purpose"))
                continue
            pieces = atoms(text)
            lines = pack("T|", pieces) if pieces is not None else None
            if not lines:
                left_out.append((label, "a line cannot be cut to fit a block line"))
                continue
            cases.append({"label": label, "word": word, "text": text, "lines": lines, "should_error": should_error})

    # v3 is run until every case left is one it gets through
    while True:
        results = run_v3(args.v3, cases)
        bad = [i for i, r in enumerate(results) if r is None]
        if not bad:
            break
        first = bad[0]
        left_out.append((cases[first]["label"], "v3 did not get through it in one session"))
        del cases[first]

    blocks, current = [], []

    def flush():
        if current:
            blocks.append(list(current))
            del current[:]

    for part in HARNESS.strip("\n").split("=====\n"):
        current.extend(part.strip("\n").split("\n"))
        flush()

    disagreements = []
    for case, (error, stack, output) in zip(cases, results):
        if error != case["should_error"]:
            disagreements.append((case["label"], "v3's table says %s, v3 %s"
                                  % ("error" if case["should_error"] else "no error",
                                     "raised one" if error else "raised none")))
        lines = ["T{ " + case["label"]] + case["lines"]
        if error:
            lines.append("->ERR")
        else:
            tail = ["->"] + [str(v) for v in stack] + ["}T", str(len(output)), str(checksum(output)), "T="]
            lines.extend(pack("", tail))
        if len(current) + len(lines) > 16:
            flush()
        current.extend(lines)
    flush()
    blocks.append(["T-REPORT", "FORGET T#  DECIMAL"])

    with open(OUTPUT, "w", encoding="ascii") as out:
        for k, block in enumerate(blocks):
            assert len(block) <= 16, block
            out.write("Block %d\n" % (FIRST_BLOCK + k))
            for line in block:
                assert len(line) <= LINE_MAX, line
                out.write(line + "\n")

    tested = sorted({c["word"] for c in cases})
    untested = [w for w in REQUIRED if w not in set(tested)]
    report = ["POST cases: %d, in %d blocks (%d to %d)" % (len(cases), len(blocks), FIRST_BLOCK, FIRST_BLOCK + len(blocks) - 1),
              "Required words with at least one case: %d of %d" % (len(tested), len(REQUIRED)),
              "Required words with no case: " + " ".join(untested), "",
              "Left out (%d):" % len(left_out)]
    report += ["  %-32s %s" % item for item in left_out]
    report += ["", "v3 did not do what its own table expects (%d):" % len(disagreements)]
    report += ["  %-32s %s" % item for item in disagreements]
    text = "\n".join(report) + "\n"
    if args.report:
        open(args.report, "w").write(text)
    sys.stdout.write(text)


if __name__ == "__main__":
    main()