File size: 3,335 Bytes
25b7932 | 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 | #!/usr/bin/env python3
from __future__ import annotations
import json, os, shutil
from pathlib import Path
from fastapi.testclient import TestClient
from app import app
def main() -> int:
os.environ["SUBSTRATE_ACTION_KEY"] = "test-secret-key"
os.environ["SUBSTRATE_RECEIPT_ROOT"] = "runs/api_receipts_validation"
root = Path(os.environ["SUBSTRATE_RECEIPT_ROOT"])
if root.exists():
shutil.rmtree(root)
client = TestClient(app)
report = {
"version": "v3.3.1",
"health_ok": False,
"unauthorized_blocked": False,
"command_ok": False,
"receipt_fetch_ok": False,
"pressure_spine_hir_x_oam": False,
"source_mutation_blocked": False,
"workspace_local_receipt_written": False,
"human_review_required": False,
"overlay_proposal_present": False,
"canonical_boundary_present": False,
"status": "FAIL"
}
health = client.get("/v1/mesh/health")
report["health_ok"] = health.status_code == 200 and health.json().get("status") == "HELD"
unauth = client.post("/v1/mesh/command", json={"raw_user_text":"robotics humans DNA"})
report["unauthorized_blocked"] = unauth.status_code == 401
headers = {"Authorization": "Bearer test-secret-key"}
payload = {
"command_type":"DISCOVER",
"raw_user_text":"robotics humans DNA",
"workspace_id":"workspace_validation_001",
"session_id":"user_001",
"source_mutation_allowed": False
}
command = client.post("/v1/mesh/command", json=payload, headers=headers)
if command.status_code == 200:
data = command.json()
report["command_ok"] = (data.get("status") == "HELD" and bool(data.get("command_id")))
report["pressure_spine_hir_x_oam"] = data.get("receipt", {}).get("pressure_spine") == "HIR × OAM"
report["source_mutation_blocked"] = data.get("receipt", {}).get("source_mutation") == "NONE__OVERLAY_PROPOSAL_ONLY"
report["human_review_required"] = data.get("human_review_required") is True
report["overlay_proposal_present"] = bool(data.get("overlay_proposals"))
report["canonical_boundary_present"] = data.get("boundary", {}).get("canonical_sources") == "READ_ONLY" and data.get("boundary", {}).get("canonical_lexicon") == "READ_ONLY"
report["workspace_local_receipt_written"] = Path(data.get("api_receipt_paths", {}).get("response_path", "")).exists()
receipt = client.get(f"/v1/mesh/receipt/{data['command_id']}?workspace_id=workspace_validation_001", headers=headers)
report["receipt_fetch_ok"] = receipt.status_code == 200 and receipt.json().get("command_id") == data["command_id"]
# Save example response for packet readers.
ex = Path("examples/responses")
ex.mkdir(parents=True, exist_ok=True)
(ex / "robotics_humans_dna_api_response.json").write_text(json.dumps(data, indent=2, ensure_ascii=False), encoding="utf-8")
report["status"] = "PASS" if all(v is True for k,v in report.items() if k not in {"version","status"}) else "FAIL"
Path("V3_3_1_VALIDATION_RESULT.json").write_text(json.dumps(report, indent=2), encoding="utf-8")
print(json.dumps(report, indent=2))
return 0 if report["status"] == "PASS" else 1
if __name__ == "__main__":
raise SystemExit(main())
|