refactor: separate language engine from CLI and server hosts
Verify Aster engine / verify (push) Failing after 0s

This commit is contained in:
2026-10-20 01:08:51 +08:00
parent 51cd64e688
commit 55475a9d4b
72 changed files with 214 additions and 4466 deletions
+7 -15
View File
@@ -1,18 +1,10 @@
# 检查与发布脚本
# 引擎维护脚本
所有构建、测试、冒烟和部署脚本只在用户指定的远端环境运行。generate_ast.py 是源码生成/静态同步工具,可本地使用;test_generate_ast.py 仅远端执行。本目录 README 负责定位与副作用边界;全局流程见 [验证](../docs/operations/validation.md) 和 [部署](../docs/operations/deployment.md)。
| 文件 | 入口与副作用 |
| 文件 | 职责 |
| --- | --- |
| [generate_ast.py](generate_ast.py) | 从 Endor.asdl 生成并格式化 Rust/TypeScript/JSON Schema;--write 更新产物,--check 只比较,无依赖下载或语言执行 |
| [test_generate_ast.py](test_generate_ast.py) | 远端 unittest:schema 扩展同步、递归装箱/遍历、非法定义拒绝 |
| [endor_examples.py](endor_examples.py) | 仅远端:构建 CLI,按 examples/cases.json 执行每个真实 Endor 文件,严格比较 stdout/stderr/退出码;保存源码、命令、实际输出、PASS/FAIL/NOT_RUN 的 Markdown 和 JSON,失败也写报告 |
| [test_endor_examples.py](test_endor_examples.py) | 仅远端:报告清单覆盖、精确输出、不匹配、构建失败、超时部分输出与异常时仍保存双格式报告;模拟子进程,不启动 VM |
| [check.sh](check.sh) | 先检查 ASDL 产物同步、生成器与输出报告行为,再运行 Cargo fmt 与独立 lexer.peg/grammar.peg 的 rustfmt 检查、workspace 测试、严格 Clippy;会使用 Cargo 依赖和构建产物 |
| [smoke.py](smoke.py) | 标准库 Python 脚本;构建后用 Cargo metadata 定位 cli/server 的 debug 二进制,支持 CARGO_TARGET_DIR;文件 metadata 跨进程与日文值;失效数据库变量下独立启动、不创建 SQLite 文件、HTTP 与 SIGTERM,不改生产配置 |
| [ci-ces-control.sh](ci-ces-control.sh) | 需要 ASTER_RELEASE_SHA;构建 ci 镜像、创建临时网络/PostgreSQL/检查容器;检查后独立执行 Endor 示例,失败不掩盖原检查状态;清理容器前复制两份报告到宿主 ci-output,提前失败也保留 NOT_RUN 说明;退出清理临时容器和网络,保留专属远端 Cargo 缓存卷 |
| [deploy-ces-control.sh](deploy-ces-control.sh) | ces-control root 执行;校验已构建镜像、锁定发布、保留密码和数据、验证 HTTP 与无数据库 CLI 引导后切换 current,失败尝试恢复旧版;保留镜像;应用服务 aster 内调用 cli bootstrap,基础信息改写文件;一次性只读导入旧基础信息,不覆盖新文件或删除原库 |
| README.md | 本目录入口 |
| generate_ast.py | 从 Endor.asdl 生成 Rust/TypeScript/JSON Schema;--check 静态核对 |
| test_generate_ast.py | 生成器回归,仅远端运行 |
| check.sh | 远端格式、生成器、引擎测试和 Clippy |
CI/部署使用的镜像与路径仅属于本项目,禁止改动其他项目和共享 Runner。smoke.py 临时选择端口后再启动宿主,存在小的端口竞争窗口,指定环境遇到冲突可重试。
调整变量同步 [配置](../docs/operations/configuration.md),调整端点/命令同步 [运行指南](../docs/operations/README.md),调整脚本名同步 [工作流](../.gitea/workflows/README.md)。返回 [项目入口](../README.md)。
CLI 示例实际输出脚本已归 Lyra;数据库、HTTP 冒烟与部署脚本已归 Hestia。Aster 无生产部署。返回 [项目入口](../README.md)。
-1
View File
@@ -2,7 +2,6 @@
set -eu
python3 scripts/generate_ast.py --check
python3 scripts/test_generate_ast.py
python3 scripts/test_endor_examples.py
cargo fmt --all --check
rustfmt --edition 2024 --check crates/frontend/src/lexer.peg crates/frontend/src/grammar.peg
cargo test --workspace --locked
-75
View File
@@ -1,75 +0,0 @@
#!/bin/sh
# Run only on the designated remote CI host, never on the developer's machine.
set -eu
cd "$(CDPATH= cd -- "$(dirname -- "$0")/.." && pwd)"
release_sha=${ASTER_RELEASE_SHA:?ASTER_RELEASE_SHA is required}
case "$release_sha" in *[!0-9a-f]*|'') echo "Invalid release SHA" >&2; exit 1;; esac
test "${#release_sha}" -eq 40
ci_id="${GITHUB_RUN_ID:-manual}-$$"
network="endorphin-aster-vm-ci-$ci_id"
postgres="endorphin-aster-vm-ci-pg-$ci_id"
checks="endorphin-aster-vm-ci-checks-$ci_id"
image="endorphin-aster-vm-ci:$release_sha"
report_dir="$PWD/ci-output"
mkdir -p "$report_dir"
# Reset both outputs before any build, so even an early failure has an honest
# NOT_RUN artifact rather than a previous attempt's example results.
printf '# Endor 示例实际运行结果\n\n提交:%s\n\n状态:NOT_RUN\n\nCI 尚未完成示例执行;请查看验证步骤日志。\n' "$release_sha" > "$report_dir/endor-examples.md"
printf '{"format_version":1,"commit":"%s","status":"NOT_RUN","error":"CI did not finish example execution; inspect the verification log.","cases":[]}\n' "$release_sha" > "$report_dir/endor-examples.json"
cleanup() {
docker rm -f "$checks" "$postgres" >/dev/null 2>&1 || true
docker network rm "$network" >/dev/null 2>&1 || true
}
trap cleanup EXIT
trap 'exit 130' INT
trap 'exit 143' TERM
docker build --target ci -t "$image" .
docker network create "$network" >/dev/null
# Disposable test database: no production mounts and no published ports.
docker run -d --name "$postgres" --network "$network" --network-alias postgres \
--env POSTGRES_USER=aster --env POSTGRES_DB=aster \
--env POSTGRES_PASSWORD=aster-ci-only \
--tmpfs /var/lib/postgresql/data postgres:17-bookworm >/dev/null
ready=false
for attempt in $(seq 1 60); do
if docker exec "$postgres" pg_isready -h 127.0.0.1 -U aster -d aster >/dev/null 2>&1; then
ready=true
break
fi
sleep 1
done
if [ "$ready" != true ]; then
echo "CI PostgreSQL did not become ready" >&2
exit 1
fi
check_status=0
docker run --name "$checks" --init --network "$network" \
--mount type=volume,src=endorphin-aster-vm-ci-cargo,dst=/usr/local/cargo/registry \
--mount type=volume,src=endorphin-aster-vm-ci-target,dst=/build/target \
--env ASTER_TEST_POSTGRES_URL='postgres://aster:aster-ci-only@postgres:5432/aster?sslmode=disable' \
--env ASTER_RELEASE_SHA="$release_sha" \
"$image" sh -eu -c '
checks=0
sh scripts/check.sh || checks=$?
examples=0
python3 scripts/endor_examples.py --output-dir /build/ci-output || examples=$?
test "$checks" -eq 0
test "$examples" -eq 0
cargo test -p storage --test postgres --locked -- --ignored
python3 scripts/smoke.py
' || check_status=$?
# Copy reports before the EXIT trap removes the stopped checks container.
# Keep the original validation failure status if report extraction also fails.
copy_status=0
docker cp "$checks:/build/ci-output/endor-examples.md" "$report_dir/endor-examples.md" || copy_status=$?
docker cp "$checks:/build/ci-output/endor-examples.json" "$report_dir/endor-examples.json" || copy_status=$?
if [ "$check_status" -ne 0 ]; then
exit "$check_status"
fi
exit "$copy_status"
-119
View File
@@ -1,119 +0,0 @@
#!/bin/sh
# Deploy only the image produced after the remote database checks succeed.
set -eu
cd "$(CDPATH= cd -- "$(dirname -- "$0")/.." && pwd)"
test "$(id -u)" -eq 0 || { echo "Deployment requires root on ces-control" >&2; exit 1; }
release_sha=${ASTER_RELEASE_SHA:?ASTER_RELEASE_SHA is required}
case "$release_sha" in *[!0-9a-f]*|'') echo "Invalid release SHA" >&2; exit 1;; esac
test "${#release_sha}" -eq 40
image="endorphin-aster-vm:$release_sha"
docker image inspect "$image" >/dev/null
umask 077
root=/opt/endorphin-aster-vm
config=/etc/endorphin-aster-vm/production.env
release="$root/releases/$release_sha"
mkdir -p "$release" /etc/endorphin-aster-vm /var/lib/endorphin-aster-vm/postgres /var/lib/endorphin-aster-vm/host
chown 10001:10001 /var/lib/endorphin-aster-vm/host
chmod 700 /var/lib/endorphin-aster-vm/host
exec 9>/var/lock/endorphin-aster-vm-deploy.lock
flock 9
if [ ! -f "$config" ]; then
password=$(openssl rand -hex 24)
printf 'POSTGRES_PASSWORD=%s\nASTER_PORT=18081\n' "$password" > "$config"
unset password
fi
chmod 600 "$config"
cp docker-compose.ces-control.yml "$release/compose.yml"
printf 'ASTER_IMAGE=%s\n' "$image" > "$release/release.env"
previous=$(readlink -f "$root/current" 2>/dev/null || true)
legacy_export=
cleanup_export() {
if [ -n "$legacy_export" ]; then rm -f "$legacy_export"; fi
}
trap cleanup_export EXIT
compose_at() {
location=$1
shift
docker compose -p endorphin-aster-vm --env-file "$config" \
--env-file "$location/release.env" -f "$location/compose.yml" "$@"
}
rollback() {
echo "Deployment failed; application data has been retained" >&2
if [ -n "$previous" ] && [ "$previous" != "$release" ] && [ -f "$previous/compose.yml" ]; then
echo "Restoring the previously healthy release" >&2
compose_at "$previous" up -d --no-build --wait --wait-timeout 180 || true
fi
}
if ! compose_at "$release" up -d --no-build --wait --wait-timeout 180; then
rollback
exit 1
fi
if ! health=$(compose_at "$release" exec -T aster curl --fail --silent http://127.0.0.1:8080/health); then
rollback
exit 1
fi
printf '%s' "$health" | python3 -c 'import json,sys; d=json.load(sys.stdin); assert d == {"status":"ok"}' || {
rollback
exit 1
}
if ! bootstrap=$(compose_at "$release" exec -T aster cli bootstrap); then
rollback
exit 1
fi
printf '%s' "$bootstrap" | python3 -c 'import sys; s=sys.stdin.read(); assert "runtime=EndorObject type_system=disabled" in s; assert "released heap_objects=0" in s' || {
rollback
exit 1
}
# Import legacy basic information once, without changing database records.
# Files are authoritative; cli import only creates records that are missing.
if [ ! -f /var/lib/endorphin-aster-vm/host/.legacy-imported ]; then
if ! has_legacy=$(compose_at "$release" exec -T postgres env PGCLIENTENCODING=UTF8 psql -X -U aster -d aster -A -t -v ON_ERROR_STOP=1 \
-c "SELECT to_regclass('public.aster_metadata') IS NOT NULL"); then
rollback
exit 1
fi
if [ "$has_legacy" = t ]; then
legacy_export=$(mktemp)
if ! compose_at "$release" exec -T postgres env PGCLIENTENCODING=UTF8 psql -X -U aster -d aster -A -t -v ON_ERROR_STOP=1 \
-c "SELECT COALESCE(json_agg(json_build_object('key', key, 'value', value)), '[]'::json) FROM public.aster_metadata" > "$legacy_export"; then
rollback
exit 1
fi
if ! compose_at "$release" exec -T aster cli metadata import < "$legacy_export"; then
rollback
exit 1
fi
rm -f "$legacy_export"
legacy_export=
elif [ "$has_legacy" != f ]; then
echo "Cannot determine legacy information schema" >&2
rollback
exit 1
fi
if ! touch /var/lib/endorphin-aster-vm/host/.legacy-imported; then
rollback
exit 1
fi
fi
# File-backed host information; no SQL table or database write.
if ! compose_at "$release" exec -T aster cli metadata put deployment.release "$release_sha"; then
rollback
exit 1
fi
if ! actual=$(compose_at "$release" exec -T aster cli metadata get deployment.release); then
rollback
exit 1
fi
if [ "$actual" != "$release_sha" ]; then
rollback
exit 1
fi
ln -sfn "$release" "$root/current.next"
mv -Tf "$root/current.next" "$root/current"
compose_at "$release" ps
echo "Deployed release=$release_sha postgres=provisioned endpoint=127.0.0.1:18081"
echo "Persistent PostgreSQL data: /var/lib/endorphin-aster-vm/postgres"
-145
View File
@@ -1,145 +0,0 @@
#!/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())
-88
View File
@@ -1,88 +0,0 @@
#!/usr/bin/env python3
"""Remote CI end-to-end smoke test, no external database required."""
import json
import os
from pathlib import Path
import signal
import socket
import subprocess
import tempfile
import time
import urllib.request
ROOT = Path(__file__).resolve().parents[1]
def run(*args, env):
return subprocess.run(args, cwd=ROOT, env=env, check=True, text=True, capture_output=True).stdout.strip()
def main():
env = os.environ.copy()
run("cargo", "build", "--workspace", "--locked", env=env)
# Cargo may use a configured CARGO_TARGET_DIR; discover the actual binaries.
metadata = json.loads(run("cargo", "metadata", "--no-deps", "--format-version", "1", env=env))
binary_dir = Path(metadata["target_directory"]) / "debug"
cli = str(binary_dir / "cli")
server = str(binary_dir / "server")
with tempfile.TemporaryDirectory(prefix="aster-smoke-") as directory:
env["ASTER_STATE_DIR"] = str(Path(directory) / "host-info")
env["SQLITE_PATH"] = str(Path(directory) / "nested" / "aster.db")
# Neither executable may use database settings for basic VM information.
env["DATABASE_URL"] = "invalid-database-url"
bootstrap = run(cli, "bootstrap", env=env)
assert "runtime=EndorObject type_system=disabled" in bootstrap, bootstrap
assert "objects_distinct=true heap_objects=2" in bootstrap, bootstrap
assert "released heap_objects=0" in bootstrap, bootstrap
assert "stored (file)" in run(cli, "metadata", "put", "project.note", "日本語・中文・😀", env=env)
assert run(cli, "metadata", "get", "project.note", env=env) == "日本語・中文・😀"
legacy = json.dumps([{"key": "project.note", "value": "old"},
{"key": "legacy.note", "value": "preserved"}])
subprocess.run([cli, "metadata", "import"], input=legacy, cwd=ROOT, env=env,
check=True, text=True, capture_output=True)
assert run(cli, "metadata", "get", "project.note", env=env) == "日本語・中文・😀"
assert run(cli, "metadata", "get", "legacy.note", env=env) == "preserved"
assert not Path(env["SQLITE_PATH"]).exists(), "CLI must not open a database"
with socket.socket() as sock:
sock.bind(("127.0.0.1", 0))
port = sock.getsockname()[1]
env["ASTER_LISTEN"] = f"127.0.0.1:{port}"
with tempfile.TemporaryFile(mode="w+") as log:
process = subprocess.Popen([server], cwd=ROOT, env=env, stdout=log, stderr=log)
try:
deadline = time.monotonic() + 15
while True:
try:
with urllib.request.urlopen(f"http://127.0.0.1:{port}/health", timeout=1) as response:
health = json.load(response)
break
except (OSError, ValueError):
if process.poll() is not None or time.monotonic() >= deadline:
log.seek(0)
raise AssertionError(f"server did not become healthy: {log.read()}")
time.sleep(0.1)
assert health == {"status": "ok"}, health
for _ in range(2):
with urllib.request.urlopen(f"http://127.0.0.1:{port}/bootstrap", timeout=2) as response:
info = json.load(response)
assert info == {"object_model": "EndorObject", "type_system": False, "heap_objects": 0}, info
finally:
process.send_signal(signal.SIGTERM)
try:
process.wait(timeout=10)
except subprocess.TimeoutExpired:
process.kill()
process.wait()
raise AssertionError("server did not shut down gracefully")
assert process.returncode == 0, process.returncode
assert not Path(env["SQLITE_PATH"]).exists(), "Server must not open a database"
assert run(cli, "metadata", "get", "host.version", env=env) == "0.1.0"
record = json.loads((Path(env["ASTER_STATE_DIR"]) / "host.version.json").read_text(encoding="utf-8"))
assert record == {"format_version": 1, "key": "host.version", "value": "0.1.0"}
assert run(cli, "metadata", "get", "project.note", env=env) == "日本語・中文・😀"
print("PASS: no implicit database access; file metadata persistence; HTTP health/bootstrap; SIGTERM shutdown")
if __name__ == "__main__":
main()
-80
View File
@@ -1,80 +0,0 @@
"""Remote-only regression checks for example output capture and failure reports."""
import contextlib
import io
import json
from pathlib import Path
import subprocess
import tempfile
import unittest
from unittest.mock import patch
import endor_examples as examples
def result(stdout="", code=0):
return {"command": ["mock"], "exit_code": code, "stdout": stdout, "stderr": "", "error": None}
def report():
return {"format_version": 1, "commit": "test", "generated_at": "test", "status": "NOT_RUN",
"preparation": [], "cases": [], "error": None}
class ExampleReportTests(unittest.TestCase):
def test_manifest_runs_every_source_and_captures_exact_output(self):
cases = examples.load_cases()
responses = [result(), result(json.dumps({"target_directory": "/tmp/ci-target"}))]
responses += [result(case["expected_stdout"]) for case in cases]
collected = report()
with patch.object(examples, "invoke", side_effect=responses) as invoke:
examples.collect(collected)
self.assertEqual(collected["status"], "PASS")
self.assertEqual(invoke.call_count, len(cases) + 2)
for actual, expected in zip(collected["cases"], cases):
self.assertEqual(actual["stdout"], expected["expected_stdout"])
self.assertEqual(actual["status"], "PASS")
self.assertEqual(actual["binary"], "/tmp/ci-target/debug/cli")
def test_output_mismatch_fails_and_build_failure_never_claims_examples_ran(self):
cases = examples.load_cases()
responses = [result(), result(json.dumps({"target_directory": "/tmp/target"}))]
responses += [result("wrong\n")] + [result(case["expected_stdout"]) for case in cases[1:]]
collected = report()
with patch.object(examples, "invoke", side_effect=responses):
examples.collect(collected)
self.assertEqual(collected["status"], "FAIL")
self.assertEqual(collected["cases"][0]["stdout"], "wrong\n")
self.assertEqual(collected["cases"][0]["status"], "FAIL")
self.assertIn("wrong", examples.markdown(collected))
collected = report()
with patch.object(examples, "invoke", return_value=result(code=1)) as invoke:
examples.collect(collected)
self.assertEqual(invoke.call_count, 1)
self.assertEqual(collected["status"], "NOT_RUN")
self.assertTrue(all(case["status"] == "NOT_RUN" for case in collected["cases"]))
def test_timeout_keeps_partial_utf8_stdout_and_stderr(self):
failure = subprocess.TimeoutExpired(["cli"], 1, output="日".encode(), stderr=b"partial error")
with patch.object(examples.subprocess, "run", side_effect=failure):
captured = examples.invoke(["cli"], 1)
self.assertIsNone(captured["exit_code"])
self.assertEqual(captured["stdout"], "日")
self.assertEqual(captured["stderr"], "partial error")
self.assertIn("timed out", captured["error"])
def test_unexpected_failure_still_writes_both_artifacts(self):
with tempfile.TemporaryDirectory() as directory:
with patch("sys.argv", ["endor_examples.py", "--output-dir", directory]), \
patch.object(examples, "collect", side_effect=ValueError("broken manifest")), \
contextlib.redirect_stdout(io.StringIO()):
self.assertEqual(examples.main(), 1)
payload = json.loads(Path(directory, "endor-examples.json").read_text(encoding="utf-8"))
self.assertEqual(payload["status"], "FAIL")
self.assertEqual(payload["error"], "broken manifest")
self.assertIn("broken manifest", Path(directory, "endor-examples.md").read_text(encoding="utf-8"))
self.assertTrue(examples.fence("```\n日").startswith("````text\n"))
if __name__ == "__main__":
unittest.main()