crosstyan
30cd956c5c
Track libavformat RTMP session initialization state so teardown only writes a trailer after the muxer header succeeds. This avoids calling av_write_trailer() on partially initialized sessions when avio_open2() fails with Connection refused. Add a fault-suite regression for libavformat RTMP connection refusal and update the summary helper to compute all_pass from the actual manifest size instead of a hardcoded seven-row expectation. Verified by rebuilding cvmmap_streamer and rtmp_output_tester, reproducing the refused-connection path without a crash, and running ./scripts/fault_suite.sh successfully (8/8).
120 lines
3.5 KiB
Python
Executable File
120 lines
3.5 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import csv
|
|
import json
|
|
import sys
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import cast
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class CliArgs:
|
|
manifest: str
|
|
output: str
|
|
run_id: str
|
|
run_dir: str
|
|
started_at: str
|
|
finished_at: str
|
|
mode: str
|
|
|
|
|
|
def parse_args() -> CliArgs:
|
|
parser = argparse.ArgumentParser(description="Build fault suite summary")
|
|
_ = parser.add_argument("--manifest", required=True)
|
|
_ = parser.add_argument("--output", required=True)
|
|
_ = parser.add_argument("--run-id", required=True)
|
|
_ = parser.add_argument("--run-dir", required=True)
|
|
_ = parser.add_argument("--started-at", required=True)
|
|
_ = parser.add_argument("--finished-at", required=True)
|
|
_ = parser.add_argument("--mode", required=True, choices=("baseline", "degraded"))
|
|
parsed = parser.parse_args(sys.argv[1:])
|
|
return CliArgs(
|
|
manifest=cast(str, parsed.manifest),
|
|
output=cast(str, parsed.output),
|
|
run_id=cast(str, parsed.run_id),
|
|
run_dir=cast(str, parsed.run_dir),
|
|
started_at=cast(str, parsed.started_at),
|
|
finished_at=cast(str, parsed.finished_at),
|
|
mode=cast(str, parsed.mode),
|
|
)
|
|
|
|
|
|
def parse_manifest(path: str) -> list[dict[str, str]]:
|
|
rows: list[dict[str, str]] = []
|
|
with open(path, "r", encoding="utf-8", newline="") as handle:
|
|
reader = csv.DictReader(handle, delimiter="\t")
|
|
for raw in reader:
|
|
row = {key: "" if value is None else str(value) for key, value in raw.items()}
|
|
rows.append(row)
|
|
return rows
|
|
|
|
|
|
def parse_int(value: str) -> int:
|
|
try:
|
|
return int(value)
|
|
except (TypeError, ValueError):
|
|
return -1
|
|
|
|
|
|
def parse_duration_ms(value: str) -> int:
|
|
try:
|
|
return int(value)
|
|
except (TypeError, ValueError):
|
|
return 0
|
|
|
|
|
|
def build_summary(args: CliArgs) -> dict[str, object]:
|
|
manifest_rows = parse_manifest(args.manifest)
|
|
rows = [
|
|
{
|
|
"order": parse_int(row["order"]),
|
|
"id": row["scenario_id"],
|
|
"name": row["name"],
|
|
"status": row["status"],
|
|
"reason": row["reason"],
|
|
"duration_ms": parse_duration_ms(row["duration_ms"]),
|
|
"exit_codes": {"command": parse_int(row["command_rc"])},
|
|
"evidence": {"log_path": row["log_path"]},
|
|
}
|
|
for row in sorted(manifest_rows, key=lambda item: parse_int(item["order"]))
|
|
]
|
|
|
|
pass_count = sum(1 for row in rows if row["status"] == "PASS")
|
|
fail_count = sum(1 for row in rows if row["status"] == "FAIL")
|
|
skip_count = sum(1 for row in rows if row["status"] == "SKIP")
|
|
all_pass = len(rows) > 0 and pass_count == len(rows) and fail_count == 0 and skip_count == 0
|
|
|
|
return {
|
|
"run_id": args.run_id,
|
|
"run_dir": args.run_dir,
|
|
"started_at": args.started_at,
|
|
"finished_at": args.finished_at,
|
|
"mode": args.mode,
|
|
"counts": {
|
|
"total": len(rows),
|
|
"pass": pass_count,
|
|
"fail": fail_count,
|
|
"skip": skip_count,
|
|
},
|
|
"all_pass": all_pass,
|
|
"recommended_exit_code": 0 if all_pass else 1,
|
|
"rows": rows,
|
|
}
|
|
|
|
|
|
def main() -> int:
|
|
args = parse_args()
|
|
output_path = Path(args.output)
|
|
output_path.parent.mkdir(parents=True, exist_ok=True)
|
|
summary = build_summary(args)
|
|
output_path.write_text(json.dumps(summary, indent=2) + "\n", encoding="utf-8")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|