| |
| 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"] |
| |
| 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()) |
|
|
