66 lines
3.4 KiB
Python
66 lines
3.4 KiB
Python
#!/usr/bin/env python3
|
|
"""Capture this rollout's controls-suite evidence; never issue a verifier verdict."""
|
|
import argparse
|
|
import ast
|
|
import json
|
|
from pathlib import Path
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
sys.path.insert(0, str(ROOT / "tools"))
|
|
from campaign import Campaign, ControlError, atomic_json
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--integrated", action="store_true")
|
|
parser.add_argument("--reuse", help="Existing canonical-RE-relative suite artifact, revalidated against source")
|
|
args = parser.parse_args()
|
|
campaign = Campaign(str(ROOT))
|
|
contract = campaign.load("controls-bootstrap")
|
|
source = campaign.source_binding(contract)
|
|
if args.reuse:
|
|
relative = args.reuse
|
|
result = json.loads(campaign.artifact_path(relative).read_text())
|
|
if result["source_binding"] != source or result["status"] != "passed":
|
|
raise ControlError("existing test evidence is not this passed source snapshot")
|
|
else:
|
|
test_source = ROOT / "verify/campaign/test_controls.py"
|
|
expected = sorted(node.name for node in ast.walk(ast.parse(test_source.read_text()))
|
|
if isinstance(node, ast.FunctionDef) and node.name.startswith("test_"))
|
|
command = [sys.executable, "-m", "unittest", "discover", "-s", "verify/campaign", "-v"]
|
|
process = subprocess.run(command, cwd=ROOT, text=True, capture_output=True)
|
|
actual = sorted(re.findall(r"^(test_\w+) .* \.\.\. ok$", process.stderr, re.M))
|
|
ok = bool(expected) and expected == actual and process.returncode == 0
|
|
result = {"schema": "sots-controls-test/1", "status": "passed" if ok else "failed",
|
|
"source_binding": source, "command": command, "expected": expected, "passed": actual,
|
|
"returncode": process.returncode, "stdout": process.stdout, "stderr": process.stderr}
|
|
if campaign.source_binding(contract) != source:
|
|
result["status"] = "failed"
|
|
result["error"] = "source changed during control tests"
|
|
relative = f"campaign/evidence/controls-suite-{time.time_ns()}.json"
|
|
atomic_json(ROOT / relative, result)
|
|
if result["status"] != "passed":
|
|
raise ControlError(f"controls suite failed; see {relative}")
|
|
artifact = campaign.artifact(relative)
|
|
record = {"id": "controls-suite", "axis": "control-tests", **artifact,
|
|
"source": contract["baseline"], "source_binding": source, "integrated": args.integrated,
|
|
"binaries": [campaign.artifact("tools/campaign.py"), campaign.artifact("tools/run_agent.py")],
|
|
"inputs": [campaign.artifact("verify/campaign/test_controls.py"),
|
|
campaign.artifact("campaign/contract.schema.json"),
|
|
campaign.artifact("campaign/rollout/controls_evidence.py")],
|
|
"outcomes": [{"criterion": "controls-negative-paths", "status": "pass", "artifact": artifact}]}
|
|
campaign.evidence(contract["id"], record, "rollout-lead", "lead", "openai/gpt-6-astra")
|
|
print(json.dumps({"artifact": relative, "executed": len(result["passed"]),
|
|
"integrated": args.integrated, "source_binding": source}, indent=2))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
main()
|
|
except (ControlError, OSError, ValueError) as exc:
|
|
print(str(exc), file=sys.stderr)
|
|
sys.exit(1)
|