open-standards/scripts/check-wai-release-quickstart.py

#!/usr/bin/env python3
"""check-wai-release-quickstart.py: the release runs the quickstart from each archive before it
attaches it, and the Windows twin keeps the script's steps.

wai-release.yml builds the quickstart archive on each platform and attaches it to the release.
Before it is attached, each archive runs the quickstart as a new user runs it, from the archive
alone: wai/quickstart/test/clean-machine.sh on macOS and Linux, its PowerShell twin
clean-machine.ps1 on Windows. Each step of the twin is a step of the script, with the same
`wai` command. This holds the three files to that:

- each of the jobs macos, linux and windows runs its script in a step's command (a comment or
  a step's name that names it is not a run of it); that step has no `if:` and no
  `continue-on-error:`, and neither has its job, so it can neither be skipped nor fail without
  failing the job; and it comes before the job's first step that attaches anything to the
  release (one that runs wai/quickstart/publish.sh or uploads to a release's assets);
- clean-machine.sh's steps are unpack, contents, hello, verify, create, dev-check and two-tabs
  (the browser's), and clean-machine.ps1's are the same steps in the same order, the browser's
  aside, each running the same `wai` command (the command alone, up to the first `;`, `}`,
  `'`, `|`, `>` or `&&` after it);
- the unpack step of each checks the archive against the SHA-256 beside it ("$ARCHIVE.sha256",
  "$Archive.sha256") and fails when they differ, before it unpacks anything.

    check-wai-release-quickstart.py [--root DIR]

Exit 0 when all of it holds, 1 with one line a finding otherwise. Standard library only. The
selftest (check-wai-release-quickstart-selftest.sh) breaks each rule in a copy and requires a
finding.
"""
import os
import re
import sys

WORKFLOW = ".github/workflows/wai-release.yml"
SH = "wai/quickstart/test/clean-machine.sh"
PS = "wai/quickstart/test/clean-machine.ps1"
# The job, and the script it runs on its archive.
JOBS = [("macos", "clean-machine.sh"), ("linux", "clean-machine.sh"), ("windows", "clean-machine.ps1")]
SH_STEPS = ["unpack", "contents", "hello", "verify", "create", "dev-check", "two-tabs"]
BROWSER_STEP = "two-tabs"


def indent(line):
    return len(line) - len(line.lstrip(" "))


def is_comment(line):
    return line.strip().startswith("#")


def jobs(text):
    """{job name: its lines}, the jobs under the top-level `jobs:` key."""
    out = {}
    lines = text.splitlines()
    try:
        start = next(i for i, l in enumerate(lines) if l.rstrip() == "jobs:")
    except StopIteration:
        return out
    name = None
    for l in lines[start + 1:]:
        if l.strip() and not is_comment(l) and indent(l) == 0:
            break
        m = re.match(r"^  ([A-Za-z0-9_-]+):\s*$", l)
        if m:
            name = m.group(1)
            out[name] = []
        elif name is not None:
            out[name].append(l)
    return out


def job_keys(lines):
    """The job's own keys (indented four spaces), comments aside."""
    return [re.match(r"^\s*([A-Za-z0-9_-]+):", l).group(1) for l in lines if indent(l) == 4 and not is_comment(l) and re.match(r"^\s*([A-Za-z0-9_-]+):", l)]


def steps(lines):
    """The job's steps, in order: each {"keys": [its keys], "name": …, "run": [its command's
    lines, comments aside]}. A step is a list item of `steps:`; its keys are the item's first
    key and those at the indent of the item's content."""
    out = []
    in_steps = False
    item_indent = None
    cur = None
    run_indent = None
    for l in lines:
        if not l.strip():
            continue
        if indent(l) == 4 and not is_comment(l):
            in_steps = l.strip() == "steps:"
            cur = None
            continue
        if not in_steps:
            continue
        s = l.lstrip(" ")
        if s.startswith("- ") and not is_comment(l) and (item_indent is None or indent(l) == item_indent):
            item_indent = indent(l)
            cur = {"keys": [], "name": "", "run": []}
            out.append(cur)
            run_indent = None
            l = " " * (item_indent + 2) + s[2:]
        if cur is None:
            continue
        key_indent = item_indent + 2
        if run_indent is not None:
            if indent(l) > key_indent:
                if not is_comment(l):
                    cur["run"].append(l.strip())
                continue
            run_indent = None
        if is_comment(l) or indent(l) != key_indent:
            continue
        m = re.match(r"^\s*([A-Za-z0-9_-]+):\s*(.*)$", l)
        if not m:
            continue
        key, value = m.group(1), m.group(2).strip()
        cur["keys"].append(key)
        if key == "name":
            cur["name"] = value.strip("'\"")
        elif key == "run":
            if value in ("|", ">", "|-", ">-", "|+", ">+"):
                run_indent = key_indent
            elif value and not value.startswith("#"):
                cur["run"].append(value)
    return out


def runs_script(step, script):
    """Whether the step's command runs `script` (wai/quickstart/test/<script>, either slash)."""
    want = (f"wai/quickstart/test/{script}", f"wai\\quickstart\\test\\{script}")
    return any(w in l.split(" #")[0] for l in step["run"] for w in want)


def attaches(step):
    """Whether the step attaches something to the release."""
    return any("wai/quickstart/publish.sh" in l or re.search(r"/assets\b", l) for l in step["run"])


def check_workflow(text, findings):
    js = jobs(text)
    for job, script in JOBS:
        if job not in js:
            findings.append(f"{WORKFLOW}: no job {job}, which runs {script} on the archive it attaches")
            continue
        where = f"{WORKFLOW}: the {job} job"
        for k in ("if", "continue-on-error"):
            if k in job_keys(js[job]):
                findings.append(f"{where} has `{k}:`, so the quickstart from its archive can be skipped or fail without failing it")
        st = steps(js[job])
        running = [i for i, s in enumerate(st) if runs_script(s, script)]
        if not running:
            findings.append(f"{where} runs {script} in no step's command: the archive it attaches is not run as a new user runs it")
            continue
        attach = next((i for i, s in enumerate(st) if attaches(s)), None)
        if attach is None:
            findings.append(f"{where} has no step that attaches its archive to the release (wai/quickstart/publish.sh, or an upload to the release's assets)")
        for i in running:
            s = st[i]
            what = f"{where}'s step {s['name']!r}, which runs {script},"
            for k in ("if", "continue-on-error"):
                if k in s["keys"]:
                    findings.append(f"{what} has `{k}:`: it can be skipped or fail without failing the job")
            if attach is not None and i > attach:
                findings.append(f"{what} comes after {st[attach]['name']!r}, which attaches the archive: the archive is attached before it is run")


def wai_command(rest):
    """The `wai` command in a step's text, alone: from `wai ` up to the first `;`, `}`, `'`,
    `|`, `>` or `&&` after it (what the step does with its output is not the command); None
    when the step runs no `wai` command."""
    i = rest.find("wai ")
    return re.split(r"[;}'|>]|&&", rest[i:])[0].strip() if i >= 0 else None


def sh_steps(text):
    """clean-machine.sh's steps: (name, the `wai` command it runs, or None)."""
    out = []
    for l in text.splitlines():
        t = l.strip()
        m = re.match(r"^(?:\[[^]]*\]\s*&&\s*)*step ([A-Za-z0-9-]+)(?: (.*))?$", t)
        if not m:
            continue
        out.append((m.group(1), wai_command(m.group(2) or "")))
    return out


def ps_steps(text):
    """clean-machine.ps1's steps: (name, the `wai` command its block runs, or None)."""
    out = []
    for l in text.splitlines():
        m = re.search(r"\bStep '([A-Za-z0-9-]+)' \{(.*)$", l.strip())
        if not m:
            continue
        out.append((m.group(1), wai_command(m.group(2))))
    return out


def block(text, start_re):
    """The lines of the block that starts on the line matching `start_re` and ends at the next
    line that is `}` alone at the start of a line."""
    lines = text.splitlines()
    for i, l in enumerate(lines):
        if re.search(start_re, l):
            for j in range(i + 1, len(lines)):
                if lines[j].rstrip() == "}":
                    return lines[i + 1:j]
            return lines[i + 1:]
    return None


def check_unpack(name, body, sums, compare, unpack, findings):
    """The unpack step's body reads the SHA-256 from `sums`, fails when it differs (`compare`),
    and does both before it unpacks (`unpack`)."""
    if body is None:
        findings.append(f"{name}: no unpack step")
        return
    code = [l.strip() for l in body if not l.strip().startswith("#")]
    reads = next((i for i, l in enumerate(code) if sums in l), None)
    fails = next((i for i, l in enumerate(code) if re.search(compare, l)), None)
    opens = next((i for i, l in enumerate(code) if re.search(unpack, l)), None)
    if reads is None:
        findings.append(f"{name}: the unpack step does not read the SHA-256 beside the archive ({sums})")
    if fails is None:
        findings.append(f"{name}: the unpack step does not fail when the archive's SHA-256 is not the one beside it")
    if opens is None:
        findings.append(f"{name}: the unpack step unpacks nothing")
    elif any(x is not None and x > opens for x in (reads, fails)):
        findings.append(f"{name}: the unpack step unpacks the archive before it checks its SHA-256")


def check_scripts(sh, ps, findings):
    s, p = sh_steps(sh), ps_steps(ps)
    names = [n for n, _ in s]
    if names != SH_STEPS:
        findings.append(f"{SH}: its steps are {names}, not {SH_STEPS}")
    want = [x for x in s if x[0] != BROWSER_STEP]
    if p != want:
        findings.append(f"{PS}: its steps are {p}; the twin runs clean-machine.sh's, the browser's aside: {want}")
    check_unpack(SH, block(sh, r"^unpack\(\)\s*\{"), '"$ARCHIVE.sha256"', r'\[ "\$got" = "\$want" \] \|\| \{.*\breturn 1', r"\btar -x", findings)
    check_unpack(PS, block(ps, r"^Step 'unpack' \{"), '"$Archive.sha256"', r"if \(\$got -ne \$want\) \{\s*throw\b", r"\bExpand-Archive\b|\btar -x", findings)


def main(argv):
    root = "."
    if argv[:1] == ["--root"] and len(argv) == 2:
        root = argv[1]
    elif argv:
        print("usage: check-wai-release-quickstart.py [--root DIR]", file=sys.stderr)
        return 2
    read = lambda p: open(os.path.join(root, p), encoding="utf-8").read()
    findings = []
    check_workflow(read(WORKFLOW), findings)
    check_scripts(read(SH), read(PS), findings)
    for f in findings:
        print(f"FAIL  {f}")
    if not findings:
        print(f"ok    {WORKFLOW}: the macos, linux and windows jobs run the quickstart from the archive before attaching it")
        print(f"ok    {PS} keeps {SH}'s steps, and both check the archive's SHA-256 before unpacking it")
    print(f"check-wai-release-quickstart: {len(findings)} finding(s)")
    return 1 if findings else 0


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