open-standards/scripts/check-wai-core-freeze.py

#!/usr/bin/env python3
"""check-wai-core-freeze.py: WAI's core does not change except by an erratum or a version bump.

    check-wai-core-freeze.py [--root <repo>] [--against <git-ref>]
    check-wai-core-freeze.py [--root <repo>] --item <id>        the digest of one item, now
    check-wai-core-freeze.py [--root <repo>] --write-lock       rewrite the lock's digests

wai/CORE.md names what is core; wai/core/lock.json pins it, item by item, each with the
SHA-256 it had when the core version was set:

  unit         a span of the specification's text between the markers
               `<!-- wai-core:begin <id> -->` and `<!-- wai-core:end <id> -->`, in wai/SPEC.md
               or an extension (`file`); its digest is over the lines between them, leading
               and trailing blank lines left out, joined by line feeds, as UTF-8
  cap:<name>   a core capability's entry in wai/registry/capabilities.json: its kind,
               status, media, role, classes, sets, section, edition and date, as canonical
               JSON (keys sorted, no insignificant whitespace, UTF-8)
  wire:<name>  a core wire identifier's entries there, the same way
  corpus:<p>   a core conformance corpus's data (a file or a directory under wai/): the
               listing of its files in SPEC §3.1's `wai.prior.set` form, leaving out
               README.md, *.py, .gitattributes and __pycache__ (its prose and its verifiers)

A run fails when an item's digest is not what the lock gives it, or, when the item has
errata, what the last of them gives it (wai/CORE-ERRATA.md: rows of the lock's core version
alone). So a change to a core item passes only

  - as an erratum: a clarification that changes no meaning, recorded as a new row of
    CORE-ERRATA.md naming the item and its new digest (`--item <id>` prints it); or
  - with the core version bumped in lock.json and its digests rewritten (`--write-lock`).

It also fails when a marker is unpaired, nested, repeated or names a unit the lock does not
hold (adding to the core is a version bump), when a lock item is gone, when an erratum names
no item of the lock or is malformed, when wai/CORE.md does not state the lock's version and
status, and when an extension of wai/extensions/ is in no profile of wai/core/profiles.json,
or in two (a profile is free to change under its own version; every extension has one).

With --against <git-ref> (CI: a pull request's base, or the commit a push to main came
after) it also reads the history: the lock's items and digests at the ref and now are the
same unless the core version rose, and the errata at the ref are the first rows of the
errata now (they are appended, never edited). A `**Breaking:**` line in the changelog does
not acknowledge a change to the core: `wai breaking` reads those, this check does not.

Exit 0 when the core is as pinned, 1 when it is not, 2 on a wrong command line.
Standard library only; runs on Python 3.9.
"""
from __future__ import annotations

import hashlib
import json
import os
import re
import subprocess
import sys

LOCK = "wai/core/lock.json"
PROFILES = "wai/core/profiles.json"
ERRATA = "wai/CORE-ERRATA.md"
CORE_MD = "wai/CORE.md"
REGISTRY = "wai/registry/capabilities.json"
EXTENSIONS = "wai/extensions"
MARKER = re.compile(r"^<!-- wai-core:(begin|end) ([A-Za-z0-9._:-]+) -->$")
ROW = re.compile(r"^\|\s*`([^`]+)`\s*\|\s*`([^`]+)`\s*\|\s*`([^`]*)`\s*\|\s*([^|]*)\|\s*(.*?)\s*\|\s*$")
HEX64 = re.compile(r"^[0-9a-f]{64}$")
SKIP_IN_CORPUS = ("README.md", ".gitattributes")


class Bad(Exception):
    """Something the check needs and cannot read: it fails, saying what."""


def sha(b: bytes) -> str:
    return hashlib.sha256(b).hexdigest()


def canon(v) -> bytes:
    return json.dumps(v, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode("utf-8")


def read(root, rel):
    try:
        with open(os.path.join(root, rel), encoding="utf-8") as f:
            return f.read()
    except OSError as e:
        raise Bad(f"{rel}: {e.strerror}")


def load_json(root, rel):
    try:
        return json.loads(read(root, rel))
    except ValueError as e:
        raise Bad(f"{rel}: not JSON ({e})")


# ---------------------------------------------------------------- the items, as they are now


def spans(text, rel):
    """{unit id: digest} of every marked span of a file, and the marker faults found."""
    out, faults, open_id, buf = {}, [], None, []
    for n, line in enumerate(text.split("\n"), 1):
        m = MARKER.match(line.strip())
        if not m:
            if open_id is not None:
                buf.append(line)
            elif "wai-core:" in line:
                faults.append(f"{rel}:{n}: a line that names wai-core is not a marker: {line.strip()[:80]}")
            continue
        kind, uid = m.groups()
        if kind == "begin":
            if open_id is not None:
                faults.append(f"{rel}:{n}: unit {uid} begins inside unit {open_id}")
                continue
            if uid in out:
                faults.append(f"{rel}:{n}: unit {uid} is marked twice")
            open_id, buf = uid, []
        else:
            if open_id != uid:
                faults.append(f"{rel}:{n}: end of {uid} where {open_id or 'no unit'} is open")
                continue
            while buf and not buf[0].strip():
                buf.pop(0)
            while buf and not buf[-1].strip():
                buf.pop()
            out[uid] = sha("\n".join(buf).encode("utf-8"))
            open_id = None
    if open_id is not None:
        faults.append(f"{rel}: unit {open_id} has no end marker")
    return out, faults


def marked_files(root):
    files = ["wai/SPEC.md"]
    d = os.path.join(root, EXTENSIONS)
    if os.path.isdir(d):
        files += [f"{EXTENSIONS}/{n}" for n in sorted(os.listdir(d)) if n.endswith(".md")]
    return files


def listing(root, rel):
    """SPEC §3.1's listing of a corpus's data files (a file, or every file below a directory)."""
    p = os.path.join(root, "wai", rel)
    if os.path.isfile(p):
        files = [(os.path.basename(rel), p)]
    elif os.path.isdir(p):
        files = []
        for d, dirs, names in os.walk(p):
            dirs[:] = sorted(x for x in dirs if x != "__pycache__")
            for n in names:
                if n in SKIP_IN_CORPUS or n.endswith((".py", ".pyc")):
                    continue
                full = os.path.join(d, n)
                files.append((os.path.relpath(full, p).replace(os.sep, "/"), full))
    else:
        return None
    lines = []
    for path, full in sorted(files, key=lambda x: x[0].encode("utf-8")):
        with open(full, "rb") as f:
            lines.append(f"{sha(f.read())}  {path}\n")
    return "".join(lines).encode("utf-8")


def current(root, lock):
    """{item id: digest now, or None where the item is gone}, and the marker faults."""
    now, faults, found = {}, [], {}
    for rel in marked_files(root):
        got, fl = spans(read(root, rel), rel)
        faults += fl
        for uid, d in got.items():
            if uid in found:
                faults.append(f"unit {uid} is marked in {found[uid][0]} and in {rel}")
            found[uid] = (rel, d)
    lock_units = {u["id"]: u for u in lock.get("units", [])}
    for uid, u in lock_units.items():
        at = found.get(uid)
        if at is not None and at[0] != f"wai/{u['file']}":
            faults.append(f"unit {uid} is marked in {at[0]}, and the lock places it in wai/{u['file']}")
        now[uid] = at[1] if at else None
    for uid, (rel, _) in found.items():
        if uid not in lock_units:
            faults.append(f"{rel}: unit {uid} is marked and the lock holds no such unit: adding to the core is a "
                          f"version bump (a marker of a profile's text does not belong here)")
    reg = load_json(root, REGISTRY)
    caps = {c.get("capability"): c for c in reg.get("capabilities", [])}
    for c in lock.get("capabilities", []):
        e = caps.get(c["capability"])
        now["cap:" + c["capability"]] = sha(canon(e)) if e is not None else None
    for w in lock.get("wire_identifiers", []):
        es = sorted((x for x in reg.get("wire_identifiers", []) if x.get("name") == w["name"]), key=canon)
        now["wire:" + w["name"]] = sha(canon(es)) if es else None
    for c in lock.get("corpora", []):
        l = listing(root, c["path"])
        now["corpus:" + c["path"]] = sha(l) if l else None
    return now, faults


def pinned(lock):
    """{item id: digest} the lock gives."""
    out = {u["id"]: u["sha256"] for u in lock.get("units", [])}
    out.update({"cap:" + c["capability"]: c["sha256"] for c in lock.get("capabilities", [])})
    out.update({"wire:" + w["name"]: w["sha256"] for w in lock.get("wire_identifiers", [])})
    out.update({"corpus:" + c["path"]: c["sha256"] for c in lock.get("corpora", [])})
    return out


# ---------------------------------------------------------------- errata


def errata(text):
    """The rows of CORE-ERRATA.md's table, in order: (erratum, item, sha256, date, what)."""
    rows = []
    for n, line in enumerate(text.split("\n"), 1):
        if not line.startswith("| `"):
            continue
        m = ROW.match(line)
        if not m:
            raise Bad(f"{ERRATA}:{n}: a row that is not `| \\`<erratum>\\` | \\`<item>\\` | \\`<sha256>\\` | <date> | <what> |`")
        rows.append(tuple(x.strip() for x in m.groups()))
    return rows


def version_key(v):
    try:
        return tuple(int(x) for x in v.split("."))
    except ValueError:
        raise Bad(f"core version {v!r} is not dotted integers")


# ---------------------------------------------------------------- the checks


def check_tree(root):
    """Findings about the tree alone, and the lock and errata read."""
    findings = []
    lock = load_json(root, LOCK)
    version, status = lock.get("core"), lock.get("status")
    if not isinstance(version, str) or not isinstance(status, str):
        raise Bad(f"{LOCK}: no core version or status")
    version_key(version)
    rows = errata(read(root, ERRATA))
    want = pinned(lock)
    now, faults = current(root, lock)
    findings += faults
    seen = set()
    expect = dict(want)
    for e, item, d, date, what in rows:
        if e in seen:
            findings.append(f"{ERRATA}: erratum {e} is listed twice")
        seen.add(e)
        if not e.startswith(version + "-E"):
            continue
        if item not in want:
            findings.append(f"{ERRATA}: erratum {e} names {item}, which the lock of core {version} does not hold")
            continue
        if not HEX64.match(d):
            findings.append(f"{ERRATA}: erratum {e}'s digest is not 64 lowercase hexadecimal digits")
            continue
        if not what:
            findings.append(f"{ERRATA}: erratum {e} says nothing of what it clarifies")
        expect[item] = d
    for item in sorted(want):
        got = now.get(item)
        if got is None:
            findings.append(f"core item {item} is gone. Removing from the core changes its meaning: bump the core "
                            f"version in {LOCK} and rewrite its digests (--write-lock), or restore it")
        elif got != expect[item]:
            findings.append(f"core item {item} changed (its digest is now {got}). A clarification that changes no "
                            f"meaning: add a row to {ERRATA} naming `{item}` with that digest. A change of meaning: "
                            f"bump the core version in {LOCK} and rewrite its digests (--write-lock)")
    md = read(root, CORE_MD)
    if f"core {version} {status}" not in md:
        findings.append(f"{CORE_MD} does not state `core {version} {status}`, the lock's version and status")
    prof = load_json(root, PROFILES)
    exts = sorted(n[:-3] for n in os.listdir(os.path.join(root, EXTENSIONS)) if n.endswith(".md"))
    owner = {}
    for p in prof.get("profiles", []):
        for x in p.get("extensions", []):
            owner.setdefault(x, []).append(p["id"])
    for x in prof.get("wholly_core", []):
        owner.setdefault(x, []).append("the core")
    for x in exts:
        if x not in owner:
            findings.append(f"extensions/{x}.md is in no profile: name it in {PROFILES} (a profile changes freely)")
        elif len(owner[x]) > 1:
            findings.append(f"extensions/{x}.md is in {', '.join(owner[x])}: an extension is in one profile")
    for x in owner:
        if x not in exts:
            findings.append(f"{PROFILES} names extensions/{x}.md, which does not exist")
    for p in prof.get("profiles", []):
        if not isinstance(p.get("version"), str) or not p.get("id"):
            findings.append(f"{PROFILES}: a profile without an id or a version: {p.get('id')}")
        for b in p.get("builds_on", []):
            if b not in want:
                findings.append(f"{PROFILES}: profile {p.get('id')} builds on {b}, which is not a core item")
    return findings, lock, rows


def git_show(root, ref, rel):
    r = subprocess.run(["git", "-C", root, "show", f"{ref}:{rel}"], capture_output=True)
    if r.returncode == 0:
        return r.stdout.decode("utf-8")
    v = subprocess.run(["git", "-C", root, "rev-parse", "--verify", "--quiet", f"{ref}^{{commit}}"], capture_output=True)
    if v.returncode != 0:
        raise Bad(f"{ref} is not a commit of this clone")
    return None


def check_history(root, ref, lock, rows):
    findings, notes = [], []
    old_text = git_show(root, ref, LOCK)
    if old_text is None:
        notes.append(f"{ref} holds no {LOCK}: the freeze starts with this change")
        return findings, notes
    old = json.loads(old_text)
    if old["core"] == lock["core"]:
        if pinned(old) != pinned(lock):
            changed = sorted(k for k in set(pinned(old)) | set(pinned(lock)) if pinned(old).get(k) != pinned(lock).get(k))
            findings.append(f"{LOCK} changed its items or digests ({', '.join(changed[:6])}{' …' if len(changed) > 6 else ''}) "
                            f"and kept core {lock['core']}: the lock is rewritten only with a version bump; a "
                            f"clarification is an erratum")
    elif version_key(lock["core"]) < version_key(old["core"]):
        findings.append(f"{LOCK}: core {lock['core']} is below core {old['core']} at {ref}")
    else:
        notes.append(f"core {old['core']} → {lock['core']}: a version bump, the lock rewritten")
    old_rows_text = git_show(root, ref, ERRATA)
    old_rows = errata(old_rows_text) if old_rows_text is not None else []
    if rows[:len(old_rows)] != old_rows:
        findings.append(f"{ERRATA}: a row {ref} holds was edited or removed; errata are appended")
    return findings, notes


def write_lock(root):
    lock = load_json(root, LOCK)
    now, faults = current(root, lock)
    if faults:
        raise Bad("; ".join(faults))
    for u in lock["units"]:
        u["sha256"] = now[u["id"]]
    for c in lock["capabilities"]:
        c["sha256"] = now["cap:" + c["capability"]]
    for w in lock["wire_identifiers"]:
        w["sha256"] = now["wire:" + w["name"]]
    for c in lock["corpora"]:
        c["sha256"] = now["corpus:" + c["path"]]
    gone = [k for k, v in now.items() if v is None]
    if gone:
        raise Bad(f"items the lock names and the tree lacks: {', '.join(gone)}")
    with open(os.path.join(root, LOCK), "w", encoding="utf-8") as f:
        json.dump(lock, f, indent=1, ensure_ascii=False)
        f.write("\n")
    print(f"wrote {LOCK}: core {lock['core']} {lock['status']}, {len(now)} items")


def main(argv):
    args = list(argv)
    root, against, item, write = ".", None, None, False
    while args:
        a = args.pop(0)
        if a in ("--root", "--against", "--item") and args:
            v = args.pop(0)
            if a == "--root":
                root = v
            elif a == "--against":
                against = v
            else:
                item = v
        elif a == "--write-lock":
            write = True
        else:
            print(__doc__.split("\n\n")[1], file=sys.stderr)
            return 2
    try:
        if write:
            write_lock(root)
            return 0
        if item is not None:
            now, _ = current(root, load_json(root, LOCK))
            if item not in now or now[item] is None:
                print(f"no item {item} in the lock, or it is gone", file=sys.stderr)
                return 1
            print(now[item])
            return 0
        findings, lock, rows = check_tree(root)
        notes = []
        if against:
            f2, notes = check_history(root, against, lock, rows)
            findings += f2
    except Bad as e:
        print(f"check-wai-core-freeze: FAIL: {e}")
        return 1
    for n in notes:
        print(f"note: {n}")
    for f in findings:
        print(f"FAIL: {f}")
    if findings:
        print(f"check-wai-core-freeze: {len(findings)} finding(s) against core {lock['core']} {lock['status']}")
        return 1
    print(f"check-wai-core-freeze: core {lock['core']} {lock['status']} is as pinned: "
          f"{len(pinned(lock))} items, {sum(1 for r in rows if r[0].startswith(lock['core'] + '-E'))} errata")
    return 0


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