open-standards/wai/quickstart/test/archive_doctor_test.py

#!/usr/bin/env python3
"""archive_doctor_test.py: archive_doctor.py passes an archive whose install and stack checks pass,
and fails each way one can not.

Each case makes a directory shaped like an unpacked archive, its bin/wai a stand-in that prints a
given report (and notes the PATH it was given), and states the exit status the check must give and
a line its output must hold. Unix only (the stand-in is a shell script); the check itself also runs
on Windows. Standard library only.
"""
from __future__ import annotations

import json
import os
import shutil
import stat
import subprocess
import sys
import tempfile

HERE = os.path.dirname(os.path.abspath(__file__))
CHECK = os.path.join(HERE, "archive_doctor.py")
REQUIRED = ["install.version", "install.path", "stack.wai-relay", "stack.wai-room", "stack.web", "stack.studio"]


# Where each check that names a path finds its piece, under the archive.
WHERE = {"install.version": "bin/wai", "stack.wai-relay": "bin/wai-relay", "stack.wai-room": "bin/wai-room", "stack.web": "share/wai/web", "stack.studio": "share/wai/studio"}


def envelope(states, root, paths=None):
    paths = dict({i: os.path.join(root, w) for i, w in WHERE.items()}, **(paths or {}))
    checks = [{"id": i, "state": s, "message": f"{i} said {s}", "fix": None, "facts": {"path": paths.get(i)} if i in paths else None} for i, s in states.items()]
    data = {"result": "failed", "counts": {}, "checks": checks}
    v = {"wai": "envelope/1", "op": "toolchain.doctor", "ok": False, "data": data}
    return json.dumps(v)


def healthy(**changes):
    states = {i: "pass" for i in REQUIRED}
    states.update({"port.8080": "fail", "browser.studio": "not-checked", "keystore.directory": "fail", "keychain.unlocked": "not-checked"})
    states.update(changes)
    return states


def archive(tmp, body):
    """A directory shaped like an unpacked archive whose bin/wai runs `body` (sh), or `body(root)`."""
    root = tempfile.mkdtemp(dir=tmp)
    os.makedirs(os.path.join(root, "bin"))
    if callable(body):
        body = body(root)
    wai = os.path.join(root, "bin", "wai")
    with open(wai, "w") as f:
        f.write("#!/bin/sh\nprintf '%s\\n' \"$PATH\" > \"$(dirname \"$0\")/../path-given\"\n" + body + "\n")
    os.chmod(wai, os.stat(wai).st_mode | stat.S_IXUSR)
    return root


def program(root, name, body):
    """bin/<name> in the archive at `root`, a stand-in that runs `body` (sh)."""
    p = os.path.join(root, "bin", name)
    with open(p, "w") as f:
        f.write("#!/bin/sh\n" + body + "\n")
    os.chmod(p, os.stat(p).st_mode | stat.S_IXUSR)


def printing(doc):
    return "cat <<'EOF'\n" + doc + "\nEOF"


def run(args):
    """The check's exit status and output; a check that does not end within 60 s is a status of its
    own, so that a hang fails its case rather than the whole test."""
    try:
        p = subprocess.run([sys.executable, "-I", CHECK] + args, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, timeout=60)
    except subprocess.TimeoutExpired:
        return "no end within 60 s", ""
    return p.returncode, p.stdout


def main():
    tmp = tempfile.mkdtemp()
    fails = 0

    def expect(what, args, code, needle):
        nonlocal fails
        got, out = run(args)
        if got != code or needle not in out:
            fails += 1
            print(f"FAIL  {what}: exit {got} (wanted {code}); '{needle}' {'present' if needle in out else 'absent'}")
            print("\n".join("      " + l for l in out.splitlines()))
        else:
            print(f"ok    {what}")

    try:
        a = archive(tmp, lambda r: printing(envelope(healthy(), r)))
        expect("every install and stack check passes; ports, browsers and the keystore are printed, not required", [a], 0, "every required check passed (6)")
        given = open(os.path.join(a, "path-given")).read().split(os.pathsep)[0].strip()
        if given != os.path.join(os.path.abspath(a), "bin"):
            fails += 1
            print(f"FAIL  the archive's bin is not first on PATH: {given}")
        else:
            print("ok    the archive's bin is first on the PATH wai is given")
        for i in REQUIRED:
            a = archive(tmp, lambda r: printing(envelope(healthy(**{i: "fail"}), r)))
            expect(f"{i} fails", [a], 1, f"FAIL  {i} is fail")
        a = archive(tmp, lambda r: printing(envelope(healthy(**{"stack.wai-relay": "not-checked"}), r)))
        expect("a stack check not checked is no pass", [a], 1, "FAIL  stack.wai-relay is not-checked")
        a = archive(tmp, lambda r: printing(envelope(healthy(**{"stack.files": "not-checked"}), r)))
        expect("another stack check, not one of the six, is held too", [a], 1, "FAIL  stack.files is not-checked")
        states = healthy()
        del states["stack.web"]
        a = archive(tmp, lambda r: printing(envelope(states, r)))
        expect("a required check missing from the report", [a], 1, "FAIL  stack.web is not in the report")
        a = archive(tmp, "echo 'wai 0.4.0'")
        expect("something else than an envelope", [a], 1, "printed no envelope")
        a = archive(tmp, printing(json.dumps({"wai": "envelope/1", "op": "toolchain.show", "ok": True, "data": {}})))
        expect("another operation's envelope", [a], 1, "printed another document")
        a = archive(tmp, "exec sleep 300")
        expect("a wai that does not answer", [a, "--timeout", "2"], 1, "did not end within 2")
        a = archive(tmp, "exit 7")
        expect("a wai that prints nothing", [a], 1, "exited 7 and printed no envelope")
        # The relay and the room service, each started once before the doctor, as it asks them.
        a = archive(tmp, lambda r: printing(envelope(healthy(), r)))
        program(a, "wai-relay", "sleep 1; echo \"$@\" > \"$(dirname \"$0\")/../relay-args\"")
        program(a, "wai-room", "exit 3")
        expect("a slow first start, before the doctor, and a program that exits as the doctor will judge", [a], 0, "first start of bin/wai-relay: 1.")
        said = open(os.path.join(a, "relay-args")).read().strip() if os.path.exists(os.path.join(a, "relay-args")) else None
        if said != "--version --json":
            fails += 1
            print(f"FAIL  the relay's first start was not asked as the doctor asks it: {said}")
        else:
            print("ok    the relay's first start is asked as the doctor asks it")
        a = archive(tmp, lambda r: printing(envelope(healthy(), r)))
        program(a, "wai-room", "exec sleep 300")
        expect("a first start that does not end", [a, "--timeout", "2"], 1, "FAIL  the first start of bin/wai-room --version --json did not end within 2 s")
        # A piece found outside the archive: a relay elsewhere on PATH, Studio's files from WAI_STUDIO.
        elsewhere = tempfile.mkdtemp(dir=tmp)
        a = archive(tmp, lambda r: printing(envelope(healthy(), r, {"stack.wai-relay": os.path.join(elsewhere, "bin", "wai-relay")})))
        expect("a relay that passed from outside the archive", [a], 1, "FAIL  stack.wai-relay passed with")
        a = archive(tmp, lambda r: printing(envelope(healthy(), r, {"stack.studio": os.path.join(elsewhere, "studio")})))
        expect("Studio's files from outside the archive", [a], 1, "which is not in the archive's share/")
        a = archive(tmp, lambda r: printing(envelope(healthy(), r, {"stack.web": os.path.join(r, "bin", "web")})))
        expect("the room client's files in the archive, and not under share/", [a], 1, "FAIL  stack.web passed with")
        a = archive(tmp, lambda r: printing(envelope(healthy(), r, {"install.version": None})))
        expect("a passing check that names no path", [a], 1, "FAIL  install.version passed with None")
        empty = tempfile.mkdtemp(dir=tmp)
        expect("a directory with no bin/wai", [empty], 2, "no bin/wai in it")
        expect("no archive named", [], 2, "name one unpacked archive")
        a = archive(tmp, lambda r: printing(envelope(healthy(**{"browser.conformance": "pass"}), r)))
        expect("--wai with --require: the named check passes", ["--wai", os.path.join(a, "bin", "wai"), "--require", "browser.conformance"], 0, "every required check passed (1)")
        a = archive(tmp, lambda r: printing(envelope(healthy(**{"browser.conformance": "fail", "stack.web": "fail"}), r)))
        expect("--wai with --require: the named check fails, and only it is required", ["--wai", os.path.join(a, "bin", "wai"), "--require", "browser.conformance"], 1, "FAIL  browser.conformance is fail")
        got, out = run(["--wai", os.path.join(a, "bin", "wai"), "--require", "browser.conformance"])
        if "stack.web is fail" in out:
            fails += 1
            print("FAIL  --wai held a check it was not asked to")
        else:
            print("ok    --wai holds the checks it is asked to, and no others")
        expect("--wai with no --require", ["--wai", os.path.join(a, "bin", "wai")], 2, "--wai takes --require")
    finally:
        shutil.rmtree(tmp, ignore_errors=True)
    if fails:
        print(f"archive_doctor_test: {fails} case(s) failed")
        return 1
    print("archive_doctor_test: every case gave its stated status")
    return 0


if __name__ == "__main__":
    sys.exit(main())