146 lines
6.8 KiB
Python
146 lines
6.8 KiB
Python
#!/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 / "vendor/aster/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": ["lyra", case["mode"], "vendor/aster/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" / "lyra"
|
||
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("LYRA_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())
|