179 lines
9.3 KiB
Python
179 lines
9.3 KiB
Python
#!/usr/bin/env python3
|
|
"""Run the deterministic, local Phase 17 final-acceptance gate.
|
|
|
|
This is a release-evidence collector, not a deployment tool. It performs only
|
|
local tests and static/configuration checks; it never sends outreach or calls a
|
|
provider. Capacity figures are bounded synthetic smoke measurements, not
|
|
production capacity claims.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
API = ROOT / "apps" / "api"
|
|
REPORT = ROOT / "docs" / "benchmarks" / "final_acceptance.latest.json"
|
|
SCHEMA = ROOT / "docs" / "benchmarks" / "final_acceptance.schema.json"
|
|
|
|
|
|
def deterministic_json(value: object) -> str:
|
|
return json.dumps(value, indent=2, sort_keys=True, separators=(",", ": ")) + "\n"
|
|
|
|
|
|
def build_capacity_smoke() -> dict:
|
|
"""Measure fixed, in-memory bounds with no clock or network dependence."""
|
|
total = 1000
|
|
requested = 100
|
|
page = list(range(total))[0:requested]
|
|
batch_input = 5000
|
|
retained = min(batch_input, 100)
|
|
return {
|
|
"bounded": len(page) <= requested and retained <= 100,
|
|
"limits": {"max_page_size": 100, "max_batch_items_retained": 100},
|
|
"pagination": {
|
|
"synthetic_total": total,
|
|
"requested_page_size": requested,
|
|
"returned_items": len(page),
|
|
"has_more": total > requested,
|
|
},
|
|
"large_batch": {
|
|
"input_items": batch_input,
|
|
"retained_items": retained,
|
|
"truncated": batch_input > retained,
|
|
},
|
|
"interpretation": "Deterministic in-memory smoke only; not a production capacity or throughput claim.",
|
|
}
|
|
|
|
|
|
def validate_report(report: dict) -> list[str]:
|
|
required = {"report", "version", "scope", "passed", "checks", "capacity_smoke", "blockers", "limitations"}
|
|
errors = [f"missing:{key}" for key in sorted(required - set(report))]
|
|
if report.get("report") != "phase17-final-acceptance": errors.append("report:const")
|
|
if type(report.get("version")) is not int or report.get("version") != 1: errors.append("version:type-or-const")
|
|
if report.get("scope") != "local-repository-acceptance": errors.append("scope:const")
|
|
if not isinstance(report.get("passed"), bool): errors.append("passed:type")
|
|
if not isinstance(report.get("checks"), dict): errors.append("checks:type")
|
|
if not isinstance(report.get("capacity_smoke"), dict): errors.append("capacity_smoke:type")
|
|
if not isinstance(report.get("limitations"), list) or not all(isinstance(x, str) for x in report.get("limitations", [])):
|
|
errors.append("limitations:type")
|
|
blockers = report.get("blockers", [])
|
|
if not isinstance(blockers, list) or not all(isinstance(x, dict) for x in blockers):
|
|
errors.append("blockers:type")
|
|
else:
|
|
for index, blocker in enumerate(blockers):
|
|
if not {"gate", "status", "reason"}.issubset(blocker): errors.append(f"blockers[{index}]:required")
|
|
return errors
|
|
|
|
|
|
def run_command(name: str, args: list[str], *, cwd: Path = ROOT, timeout: int = 300) -> dict:
|
|
if shutil.which(args[0]) is None:
|
|
return {"passed": False, "status": "unavailable", "command": args, "summary": f"{args[0]} not installed"}
|
|
try:
|
|
completed = subprocess.run(args, cwd=cwd, text=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, timeout=timeout)
|
|
except subprocess.TimeoutExpired:
|
|
return {"passed": False, "status": "timeout", "command": args, "summary": f"{name} exceeded {timeout}s"}
|
|
output = completed.stdout or ""
|
|
match = re.search(r"Ran (\d+) tests?", output)
|
|
summary = f"exit={completed.returncode}"
|
|
if match: summary += f", tests={match.group(1)}"
|
|
if completed.returncode and output:
|
|
summary += ": " + " ".join(output.strip().splitlines()[-2:])[:500]
|
|
return {"passed": completed.returncode == 0, "status": "passed" if completed.returncode == 0 else "failed", "command": args, "summary": summary}
|
|
|
|
|
|
def validate_json_files() -> dict:
|
|
files = [ROOT / "docs" / "benchmarks" / "phase16.latest.json", ROOT / "docs" / "benchmarks" / "phase16.schema.json", SCHEMA]
|
|
results = {}
|
|
parsed = {}
|
|
for path in files:
|
|
key = str(path.relative_to(ROOT))
|
|
try:
|
|
parsed[key] = json.loads(path.read_text(encoding="utf-8"))
|
|
results[key] = {"passed": True, "status": "valid_json"}
|
|
except (OSError, json.JSONDecodeError) as exc:
|
|
results[key] = {"passed": False, "status": "invalid_json", "summary": str(exc)}
|
|
phase16 = parsed.get("docs/benchmarks/phase16.latest.json")
|
|
if isinstance(phase16, dict):
|
|
required = {"benchmark", "version", "offline", "limitations", "acceptance_thresholds", "results", "checks", "passed"}
|
|
schema_ok = required.issubset(phase16) and phase16.get("benchmark") == "phase16" and phase16.get("version") == 1 and phase16.get("offline") is True and isinstance(phase16.get("passed"), bool)
|
|
results["phase16_schema_contract"] = {"passed": schema_ok, "status": "schema_valid" if schema_ok else "schema_invalid"}
|
|
else:
|
|
results["phase16_schema_contract"] = {"passed": False, "status": "schema_invalid"}
|
|
return {"passed": all(item["passed"] for item in results.values()), "files": results}
|
|
|
|
|
|
def safety_checks() -> dict:
|
|
compose = (ROOT / "docker-compose.yml").read_text(encoding="utf-8")
|
|
config = (API / "app" / "config.py").read_text(encoding="utf-8")
|
|
main = (API / "app" / "main.py").read_text(encoding="utf-8")
|
|
tests = "\n".join(p.read_text(encoding="utf-8") for p in (API / "tests").glob("test_*.py"))
|
|
checks = {
|
|
"outreach_disabled": 'AUTOMATED_OUTREACH_ENABLED: "false"' in compose and '"false"' in config and 'outreach_enabled' in main,
|
|
"no_send_network": '"network_send": False' in main and "urllib.request" not in main and "smtplib" not in main,
|
|
"tenant_routes": "organization_id=?" in main and "session_user" in main and "require_auth" in main and "tenant" in tests.lower(),
|
|
}
|
|
return {"passed": all(checks.values()), "checks": checks}
|
|
|
|
|
|
def git_state() -> dict:
|
|
result = run_command("git-state", ["git", "status", "--short", "--branch"])
|
|
head = subprocess.run(["git", "rev-parse", "HEAD"], cwd=ROOT, text=True, capture_output=True, check=False)
|
|
porcelain = subprocess.run(["git", "status", "--porcelain"], cwd=ROOT, text=True, capture_output=True, check=False)
|
|
return {
|
|
"passed": result["passed"] and head.returncode == 0 and bool(head.stdout.strip()),
|
|
"status": result["status"],
|
|
"branch_status": result.get("summary", ""),
|
|
"head": head.stdout.strip(),
|
|
"working_tree_clean": porcelain.returncode == 0 and not porcelain.stdout.strip(),
|
|
}
|
|
|
|
|
|
def build_report() -> dict:
|
|
checks = {
|
|
"api_tests": run_command("api-tests", [sys.executable, "-m", "unittest", "discover", "-v", "-s", "apps/api/tests", "-t", "apps/api"]),
|
|
"py_compile": run_command("py_compile", [sys.executable, "-m", "py_compile", *[str(p.relative_to(ROOT)) for p in sorted((ROOT / "scripts").glob("*.py"))], *[str(p.relative_to(ROOT)) for p in sorted((API / "app").glob("*.py"))]]),
|
|
"benchmark_no_latency": run_command("benchmark-no-latency", [sys.executable, "scripts/benchmark_phase16.py", "--no-latency", "--output", "/tmp/prospect-phase16-acceptance.json"]),
|
|
"shell_syntax": run_command("shell-syntax", ["bash", "-n", *[str(p.relative_to(ROOT)) for p in sorted((ROOT / "scripts").glob("*.sh"))]]),
|
|
"json_validation": validate_json_files(),
|
|
"compose_config": run_command("compose-config", ["docker", "compose", "-f", "docker-compose.yml", "config", "--quiet"]),
|
|
"git_state": git_state(),
|
|
"safety_invariants": safety_checks(),
|
|
}
|
|
blockers = [
|
|
{"gate": "remote_auth", "status": "blocked", "reason": "Remote repository authentication and branch permission are not available to this local acceptance run."},
|
|
{"gate": "deployment_access", "status": "blocked", "reason": "No production host, Docker/Compose, DNS/TLS, secrets, or deployment access is available; no deployment was attempted."},
|
|
{"gate": "real_provider_legal", "status": "blocked", "reason": "Real source/provider enablement, consent/legal basis, terms, and operational approval remain explicit gates; outreach stays disabled."},
|
|
]
|
|
return {
|
|
"report": "phase17-final-acceptance", "version": 1,
|
|
"scope": "local-repository-acceptance",
|
|
"passed": all(item.get("passed", False) for item in checks.values()),
|
|
"checks": checks,
|
|
"capacity_smoke": build_capacity_smoke(),
|
|
"blockers": blockers,
|
|
"limitations": ["Synthetic capacity smoke measurements do not establish production throughput, concurrency, durability, or availability."],
|
|
}
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--output", type=Path, default=REPORT)
|
|
args = parser.parse_args()
|
|
report = build_report()
|
|
errors = validate_report(report)
|
|
if errors: raise SystemExit("invalid acceptance report: " + ", ".join(errors))
|
|
args.output.parent.mkdir(parents=True, exist_ok=True)
|
|
args.output.write_text(deterministic_json(report), encoding="utf-8")
|
|
print(deterministic_json(report), end="")
|
|
return 0 if report["passed"] else 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|