open-standards/joule-code/joule-code-rs/crates/joule-moq-interop/scripts/public.py

#!/usr/bin/env python3
"""joule-moq-interop against the public relays an interop runner registry lists.

    python3 -I scripts/public.py <joule-moq-interop binary> <implementations.json> <out dir> [--receipts]

For each relay endpoint in the registry (`roles.relay.remote[]`, endpoints marked
inactive skipped) and each draft both it and joule-moq speak (22, 18, 16), it runs the
runner's test cases (TAP 14, one process per endpoint and draft) and, with
`--receipts`, the wai-receipts scenario; it writes each run's TAP to <out dir> and one
line per run to <out dir>/results.jsonl. It reaches the public internet: nothing runs it
in CI. Each run's debug log goes beside its TAP. Standard library only.
"""

import json
import os
import re
import subprocess
import sys
import time

OURS = {22, 18, 16}


def drafts_of(entry):
    out = set()
    for d in entry.get("draft_versions", []):
        m = re.fullmatch(r"draft-(\d+)", d)
        if m:
            out.add(int(m.group(1)))
    return out


def parse_tap(text):
    """(test name, ok, skip reason, YAML lines) for each test point."""
    points = []
    cur = None
    for line in text.splitlines():
        m = re.match(r"^(ok|not ok) \d+ - (\S+)(?: # SKIP (.*))?$", line)
        if m:
            cur = {"test": m.group(2), "ok": m.group(1) == "ok", "skip": m.group(3), "diag": {}}
            points.append(cur)
            continue
        m = re.match(r"^  ([a-z_0-9]+): (.*)$", line)
        if m and cur is not None:
            try:
                cur["diag"][m.group(1)] = json.loads(m.group(2))
            except ValueError:
                cur["diag"][m.group(1)] = m.group(2)
    return points


def main():
    if len(sys.argv) < 4:
        print(__doc__, file=sys.stderr)
        return 2
    binary, registry, out = sys.argv[1], sys.argv[2], sys.argv[3]
    receipts = "--receipts" in sys.argv[4:]
    os.makedirs(out, exist_ok=True)
    reg = json.load(open(registry))
    lines = []
    for name, entry in sorted(reg["implementations"].items()):
        relay = entry.get("roles", {}).get("relay", {})
        for ep in relay.get("remote", []):
            if ep.get("status", "active") == "inactive":
                continue
            url = ep["url"]
            transport = ep.get("transport", "quic" if url.startswith("moqt://") else "webtransport")
            if transport == "quic" and url.startswith("https://"):
                # The registry names some native QUIC endpoints with https://; the moqt
                # URI is the same authority and path.
                url = "moqt://" + url[len("https://"):]
            for d in sorted(drafts_of(entry) & OURS, reverse=True):
                # The runner's tests; then wai-receipts by namespace, and, where that does
                # not pass, by PUBLISH and with no per-object properties, to find which
                # part the relay does not carry.
                runs = [(None, [])]
                if receipts:
                    runs.append(("wai-receipts", []))
                for t, extra in runs + ([] if not receipts else [("wai-receipts", ["--flow", "publish"]), ("wai-receipts", ["--no-object-properties"])]):
                    if extra and lines and lines[-1].get("test") == "wai-receipts" and lines[-1]["exit"] == 0 and lines[-1]["relay"] == name and lines[-1]["draft"] == d and lines[-1]["transport"] == transport and not lines[-1].get("extra"):
                        continue
                    label = f"{name}_{transport}_d{d}" + (f"_{t}" if t else "") + "".join("_" + x.strip("-") for x in extra if x.startswith("--"))
                    cmd = [binary, "--verbose", "--relay", url, "--draft", str(d), "--timeout-scale", "3"] + extra
                    if ep.get("tls_disable_verify"):
                        cmd.append("--tls-disable-verify")
                    if t:
                        cmd += ["--test", t, "--json", os.path.join(out, label + ".json")]
                    start = time.time()
                    try:
                        p = subprocess.run(cmd, capture_output=True, text=True, timeout=180)
                        tap, log, code = p.stdout, p.stderr, p.returncode
                    except subprocess.TimeoutExpired as e:
                        text = lambda x: (x or b"").decode(errors="replace") if isinstance(x, bytes) else (x or "")
                        tap, log, code = text(e.stdout), text(e.stderr), -1
                    open(os.path.join(out, label + ".tap"), "w").write(tap)
                    # The session's messages and the bridge's notes, at debug level.
                    open(os.path.join(out, label + ".log"), "w").write(log)
                    rec = {
                        "test": t,
                        "extra": extra,
                        "relay": name,
                        "url": url,
                        "transport": transport,
                        "draft": d,
                        "exit": code,
                        "seconds": round(time.time() - start, 1),
                        "points": parse_tap(tap),
                        "when": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
                    }
                    lines.append(rec)
                    summary = " ".join(f"{x['test']}={'skip' if x['skip'] else ('ok' if x['ok'] else 'FAIL')}" for x in rec["points"])
                    print(f"{label}: exit {code} {summary}", flush=True)
    with open(os.path.join(out, "results.jsonl"), "w") as f:
        for r in lines:
            f.write(json.dumps(r) + "\n")
    return 0


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