endorphin-aster-vm/scripts/endor_examples.py

#!/usr/bin/env python3
"""Run Endor example files and retain actual output on the designated CI host.

Never run locally: this builds the CLI and executes language code.
"""

import argparse
from datetime import datetime, timezone
import json
import os
from pathlib import Path
import re
import shlex
import subprocess


ROOT = Path(__file__).resolve().parents[1]
EXAMPLES = ROOT / "crates/compiler/examples"


def decoded(value):
    return value.decode("utf-8", "replace") if isinstance(value, bytes) else value or ""


def invoke(command, timeout):
    """Keep stdout, stderr and failure state even for launch errors/timeouts."""
    record = {"command": command, "exit_code": None, "stdout": "", "stderr": "", "error": None}
    try:
        result = subprocess.run(command, cwd=ROOT, capture_output=True, text=True,
                                encoding="utf-8", errors="replace", timeout=timeout)
        record.update(exit_code=result.returncode, stdout=result.stdout, stderr=result.stderr)
    except subprocess.TimeoutExpired as error:
        record.update(stdout=decoded(error.stdout), stderr=decoded(error.stderr),
                      error=f"timed out after {timeout} seconds")
    except OSError as error:
        record["error"] = str(error)
    return record


def load_cases():
    manifest = json.loads((EXAMPLES / "cases.json").read_text(encoding="utf-8"))
    if manifest.get("format_version") != 1:
        raise ValueError("unsupported example manifest version")
    cases = manifest["cases"]
    names, sources = set(), set()
    for case in cases:
        if not re.fullmatch(r"[a-z][a-z0-9_-]*", case["name"]) or case["name"] in names:
            raise ValueError("invalid or duplicate example name")
        source = EXAMPLES / case["source"]
        if source.resolve().parent != EXAMPLES.resolve() or source.suffix != ".endor":
            raise ValueError("example source must be a direct .endor file")
        if case["mode"] not in ("run", "eval") or not isinstance(case["expected_stdout"], str):
            raise ValueError("invalid example mode or expected output")
        if case["source"] in sources:
            raise ValueError("duplicate example source")
        case["source_text"] = source.read_text(encoding="utf-8")
        names.add(case["name"])
        sources.add(case["source"])
    if not cases or sources != {path.name for path in EXAMPLES.glob("*.endor")}:
        raise ValueError("manifest must cover every .endor example exactly once")
    return cases


def fence(text, language="text"):
    # Keep source/output literal even if future examples print Markdown fences.
    longest = max((len(match[0]) for match in re.finditer(r"`+", text)), default=0)
    delimiter = "`" * max(3, longest + 1)
    return f"{delimiter}{language}\n{text}" + ("" if text.endswith("\n") else "\n") + delimiter


def markdown(report):
    lines = ["# Endor 示例实际运行结果", "", f"提交：`{report['commit']}`",
             f"状态：**{report['status']}**", f"生成时间（UTC）：{report['generated_at']}",
             "", "本报告仅覆盖 Endor 示例；整个 CI/CD 的状态以工作流结果为准。"]
    if report.get("error"):
        lines += ["", "报告阶段错误：", fence(report["error"])]
    for stage in report["preparation"]:
        lines += ["", "## 准备：" + shlex.join(stage["command"]),
                  f"退出码：{stage['exit_code']}；错误：{stage['error'] or '无'}"]
        if stage["exit_code"] != 0 or stage["error"]:
            lines += ["", "stdout：", fence(stage["stdout"]), "", "stderr：", fence(stage["stderr"])]
    for case in report["cases"]:
        lines += ["", f"## {case['name']} — {case['status']}", "",
                  "命令：", fence(shlex.join(case["command"]), "sh"), "", "源码：",
                  fence(case["source_text"], "endor"), "", "预期 stdout：",
                  fence(case["expected_stdout"]), "", "实际 stdout：", fence(case["stdout"]),
                  "", "实际 stderr：", fence(case["stderr"]), "",
                  f"预期退出码：0；实际退出码：{case['exit_code']}",
                  f"错误：{case['error'] or '无'}"]
    return "\n".join(lines) + "\n"


def collect(report):
    for case in load_cases():
        report["cases"].append({**case, "status": "NOT_RUN", "exit_code": None,
                                "stdout": "", "stderr": "", "error": None,
                                "expected_stderr": "", "expected_exit_code": 0,
                                "command": ["cli", case["mode"], "crates/compiler/examples/" + case["source"]]})
    build = invoke(["cargo", "build", "-p", "cli", "--locked"], timeout=600)
    report["preparation"].append(build)
    if build["exit_code"] != 0 or build["error"]:
        report.update(status="NOT_RUN", error="CLI build failed; examples were not executed.")
        return
    metadata = invoke(["cargo", "metadata", "--locked", "--no-deps", "--format-version", "1"], timeout=60)
    report["preparation"].append(metadata)
    if metadata["exit_code"] != 0 or metadata["error"]:
        report.update(status="NOT_RUN", error="Cannot locate the compiled CLI.")
        return
    binary = Path(json.loads(metadata["stdout"])["target_directory"]) / "debug" / "cli"
    for case in report["cases"]:
        command = [str(binary), *case["command"][1:]]
        actual = invoke(command, timeout=30)
        case.update({key: actual[key] for key in ["exit_code", "stdout", "stderr", "error"]})
        case["binary"] = str(binary)
        case["status"] = "PASS" if (
            case["exit_code"] == case["expected_exit_code"] and not case["error"]
            and case["stdout"] == case["expected_stdout"]
            and case["stderr"] == case["expected_stderr"]
        ) else "FAIL"
    report["status"] = "PASS" if all(case["status"] == "PASS" for case in report["cases"]) else "FAIL"


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--output-dir", type=Path, required=True)
    args = parser.parse_args()
    args.output_dir.mkdir(parents=True, exist_ok=True)
    report = {"format_version": 1, "commit": os.environ.get("ASTER_RELEASE_SHA", "unknown"),
              "generated_at": datetime.now(timezone.utc).isoformat(), "status": "NOT_RUN",
              "error": None, "preparation": [], "cases": []}
    try:
        collect(report)
    except Exception as error:
        report.update(status="FAIL", error=str(error))
    # Always persist diagnostics before returning a failing CI exit code.
    args.output_dir.joinpath("endor-examples.json").write_text(
        json.dumps(report, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
    rendered = markdown(report)
    args.output_dir.joinpath("endor-examples.md").write_text(rendered, encoding="utf-8")
    print(rendered)
    return 0 if report["status"] == "PASS" else 1


if __name__ == "__main__":
    raise SystemExit(main())