fengqun/examples/run_standard_scenario_acceptance.py

from pathlib import Path
import json
import subprocess
import sys


ROOT = Path(__file__).resolve().parents[1]


SCENARIOS = [
    {
        "id": "S01",
        "name": "syntax_import_sanity",
        "layer": "static",
        "given": "all swarm_minimal, examples, and tests Python files",
        "when": "compile every module",
        "then": "no syntax or import-time compile errors",
        "command": [sys.executable, "-B", "-m", "py_compile", *sorted(str(path.relative_to(ROOT)) for path in ROOT.glob("swarm_minimal/*.py")), *sorted(str(path.relative_to(ROOT)) for path in ROOT.glob("examples/*.py")), *sorted(str(path.relative_to(ROOT)) for path in ROOT.glob("tests/*.py"))],
    },
    {
        "id": "S02",
        "name": "unit_regression",
        "layer": "unit",
        "given": "mock NewAPI clients and in-memory swarm store",
        "when": "run the full unittest suite",
        "then": "all unit tests pass",
        "command": [sys.executable, "-B", "-m", "unittest", "discover", "-s", "tests"],
    },
    {
        "id": "S03-S06",
        "name": "deterministic_standard_scenarios",
        "layer": "scenario",
        "given": "no-network deterministic cases for continuity, dependency policy, scoring, and failure injection",
        "when": "run tests.test_standard_scenarios",
        "then": "all scenario assertions pass",
        "command": [sys.executable, "-B", "-m", "unittest", "tests.test_standard_scenarios"],
    },
    {
        "id": "S07",
        "name": "live_azure_newapi_continuous_reasoning",
        "layer": "live-integration",
        "given": "local .env with Azure PostgreSQL, Redis, Blob, and NewAPI credentials",
        "when": "run seven-step continuous reasoning acceptance",
        "then": "model discovery, PostgreSQL, Redis, Blob artifact, chain cursor, and convergence all pass",
        "command": [sys.executable, "-u", "-B", "examples/run_continuous_reasoning_acceptance.py"],
        "parse_json": True,
    },
    {
        "id": "S08",
        "name": "model_io_report_audit",
        "layer": "report-audit",
        "given": "generated docs/MODEL_AGNET_IO_REPORT.zh-CN.md",
        "when": "audit scenario coverage, task input/output sections, handoff evidence, and obvious secret patterns",
        "then": "the report is human-auditable and does not contain obvious secret values",
        "command": [sys.executable, "-B", "-m", "unittest", "tests.test_model_io_report_audit"],
    },
]


def main() -> None:
    results = []
    for scenario in SCENARIOS:
        completed = subprocess.run(
            scenario["command"],
            cwd=ROOT,
            text=True,
            capture_output=True,
            timeout=900,
        )
        evidence = summarize_process_output(completed.stdout, completed.stderr)
        parsed = None
        if scenario.get("parse_json") and completed.stdout.strip():
            parsed = parse_last_json(completed.stdout)
            if parsed:
                evidence = summarize_live_report(parsed)
        result = {
            "id": scenario["id"],
            "name": scenario["name"],
            "layer": scenario["layer"],
            "given": scenario["given"],
            "when": scenario["when"],
            "then": scenario["then"],
            "command": " ".join(scenario["command"]),
            "passed": completed.returncode == 0,
            "evidence": evidence,
        }
        if parsed:
            result["live_summary"] = parsed.get("summary", {})
        results.append(result)
        if completed.returncode != 0:
            break

    report = {
        "standard": "scenario-matrix-v1",
        "status": "PASS" if all(item["passed"] for item in results) and len(results) == len(SCENARIOS) else "FAIL",
        "scenarios": results,
    }
    print(json.dumps(report, ensure_ascii=False, indent=2))
    if report["status"] != "PASS":
        raise SystemExit(1)


def summarize_process_output(stdout: str, stderr: str) -> dict[str, object]:
    combined = "\n".join(part.strip() for part in [stdout, stderr] if part.strip())
    tail = combined[-1200:] if combined else "<no output>"
    return {"tail": tail}


def parse_last_json(text: str) -> dict[str, object] | None:
    stripped = text.strip()
    if not stripped:
        return None
    decoder = json.JSONDecoder()
    last = None
    index = 0
    while index < len(stripped):
        brace = stripped.find("{", index)
        if brace == -1:
            break
        try:
            value, end = decoder.raw_decode(stripped[brace:])
        except json.JSONDecodeError:
            index = brace + 1
            continue
        if isinstance(value, dict):
            last = value
        index = brace + end
    return last


def summarize_live_report(report: dict[str, object]) -> dict[str, object]:
    summary = report.get("summary", {})
    checks = report.get("checks", [])
    failed_checks = [check["name"] for check in checks if isinstance(check, dict) and not check.get("passed")]
    return {
        "summary": summary,
        "selected_models": report.get("selected_models", []),
        "failed_checks": failed_checks,
        "check_count": len(checks) if isinstance(checks, list) else 0,
    }


if __name__ == "__main__":
    main()