1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
|
#!/usr/bin/env python3
"""Run every value-validation demo and emit one machine-readable results JSON.
Usage (from the tool root, ``open-iov/debug/risc-v-gsp-rm-vm``):
python3 harness/value_demos/run_all.py
python3 harness/value_demos/run_all.py --only s01,s03,s05,s09
VALUE_DEMO_BACKEND=qemu python3 harness/value_demos/run_all.py
Each demo runs in its own subprocess with a hard timeout, so one wedged demo
cannot take the suite down. The aggregate lands in
``harness/value_demos/artifacts/results.json``.
"""
from __future__ import annotations
import argparse
import json
import os
import subprocess
import sys
import time
HERE = os.path.dirname(os.path.abspath(__file__))
TOOL_ROOT = os.path.dirname(os.path.dirname(HERE))
sys.path.insert(0, HERE)
from _common import ARTIFACT_ROOT, SUITE_BUDGET_S, write_json # noqa: E402
DEMOS = [
("s01", "s01_reply_consumer.py"),
("s02", "s02_counter_publication.py"),
("s03", "s03_unknown_command.py"),
("s04", "s04_run_until_function.py"),
("s05", "s05_snapshot_diff.py"),
("s06", "s06_inject_service.py"),
("s07", "s07_version_diff.py"),
("s08", "s08_mutation_triage.py"),
("s09", "s09_grabbing_a_ghidra_address.py"),
("s10", "s10_full_session_evidence.py"),
]
def _run_one(key: str, script: str, timeout: float) -> dict:
path = os.path.join(HERE, script)
start = time.monotonic()
try:
proc = subprocess.run(
[sys.executable, path],
cwd=TOOL_ROOT,
capture_output=True,
text=True,
timeout=timeout,
)
stdout, stderr, returncode = proc.stdout, proc.stderr, proc.returncode
except subprocess.TimeoutExpired as exc:
stdout = (exc.stdout or b"").decode("utf-8", "replace") if isinstance(exc.stdout, bytes) else (exc.stdout or "")
stderr = (exc.stderr or b"").decode("utf-8", "replace") if isinstance(exc.stderr, bytes) else (exc.stderr or "")
returncode = -9
elapsed = time.monotonic() - start
result = None
for line in reversed(stdout.splitlines()):
if line.startswith("RESULT_JSON: "):
try:
result = json.loads(line[len("RESULT_JSON: ") :])
except json.JSONDecodeError:
result = None
break
if result is None:
result = {
"success": False,
"failure": f"no RESULT_JSON (exit {returncode})",
"stderr_tail": stderr[-800:],
}
result["demo_key"] = key
result["runner_elapsed_s"] = round(elapsed, 4)
result["exit_code"] = returncode
return result
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--only", default="", help="comma-separated demo keys (e.g. s01,s03)")
parser.add_argument("--timeout", type=float, default=45.0, help="per-demo timeout seconds")
parser.add_argument("--out", default=os.path.join(ARTIFACT_ROOT, "results.json"))
args = parser.parse_args()
selected = [item for item in DEMOS if not args.only or item[0] in args.only.split(",")]
if args.only and not selected:
print(f"no demos match --only {args.only!r}", file=sys.stderr)
return 2
started = time.monotonic()
results = []
for key, script in selected:
print(f"--- running {script} ---", flush=True)
result = _run_one(key, script, args.timeout)
results.append(result)
status = "PASS" if result.get("success") else "FAIL"
print(f"--- {key}: {status} ({result['runner_elapsed_s']:.2f}s) ---", flush=True)
if time.monotonic() - started > SUITE_BUDGET_S:
print(f"[run_all] suite budget {SUITE_BUDGET_S:.0f}s exceeded; continuing", flush=True)
total = time.monotonic() - started
passed = [r for r in results if r.get("success")]
payload = {
"format": "greeneye-value-demos-results/1",
"tool": "harness/value_demos/run_all.py",
"backend_env": os.environ.get("VALUE_DEMO_BACKEND", "native (default)"),
"demos_selected": [k for k, _ in selected],
"demos_passed": [r["demo_key"] for r in passed],
"demos_failed": [r["demo_key"] for r in results if not r.get("success")],
"total_elapsed_s": round(total, 3),
"measured_elapsed_s": round(sum(r["runner_elapsed_s"] for r in results), 3),
"results": results,
}
write_json(args.out, payload)
print(f"\n{len(passed)}/{len(results)} demos passed in {total:.1f}s; results: {args.out}")
print(f"RESULT_JSON: {json.dumps({k: payload[k] for k in ('format', 'demos_passed', 'demos_failed', 'total_elapsed_s')}, sort_keys=True)}")
return 0 if len(passed) == len(results) else 1
if __name__ == "__main__":
raise SystemExit(main())
|
