From 200749ca5b3f079abd87a8a498ea4a230dfed329 Mon Sep 17 00:00:00 2001 From: twentyOne2x Date: Mon, 13 Jul 2026 07:13:18 +0200 Subject: [PATCH] Prevent GCP proof harness heredoc deadlocks --- ops/build_gcp_cloudsql_restore_proof.py | 100 +++++++++++++ ops/run_gcp_cloudsql_restore_drill.sh | 84 +++-------- scripts/gcp_iap_operator.sh | 164 +++------------------- scripts/gcp_iap_operator_result.py | 179 ++++++++++++++++++++++++ tests/test_gcp_iap_operator_access.py | 64 ++++----- 5 files changed, 348 insertions(+), 243 deletions(-) create mode 100755 ops/build_gcp_cloudsql_restore_proof.py create mode 100755 scripts/gcp_iap_operator_result.py diff --git a/ops/build_gcp_cloudsql_restore_proof.py b/ops/build_gcp_cloudsql_restore_proof.py new file mode 100755 index 0000000..8a371fe --- /dev/null +++ b/ops/build_gcp_cloudsql_restore_proof.py @@ -0,0 +1,100 @@ +#!/usr/bin/env python3 +"""Build the prepared proof manifest for a Cloud SQL restore drill.""" + +from __future__ import annotations + +import argparse +import hashlib +import json +from datetime import datetime, timezone +from pathlib import Path + + +def sha256(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as handle: + for chunk in iter(lambda: handle.read(1024 * 1024), b""): + digest.update(chunk) + return digest.hexdigest() + + +def build_proof(args: argparse.Namespace) -> dict[str, object]: + summary = json.loads(args.source_summary.read_text(encoding="utf-8")) + return { + "artifact": "gcp_cloudsql_restore_drill", + "generated_at_utc": datetime.now(timezone.utc).isoformat(), + "mode": "execute" if args.execute else "dry_run", + "status": "prepared_for_execute" if args.execute else "prepared", + "project_id": args.project_id, + "region": args.region, + "cloudsql_instance": args.instance, + "database": args.database, + "postgres_schema": summary["schema"], + "source_sqlite_backup": str(args.sqlite_backup), + "source_sqlite_backup_sha256": sha256(args.sqlite_backup), + "source_integrity_check": summary["source_integrity_check"], + "source_table_count": summary["table_count"], + "source_total_rows": summary["source_total_rows"], + "source_table_counts": {table["name"]: table["row_count"] for table in summary["tables"]}, + "gcs_import_uri": args.gcs_import_uri, + "local_private_artifacts": { + "postgres_import_sql": str(args.import_sql), + "source_summary_json": str(args.source_summary), + "target_counts_sql": str(args.target_counts_sql), + }, + "planned_commands": [ + f"gcloud storage cp {args.import_sql} {args.gcs_import_uri}", + f"gcloud sql import sql {args.instance} {args.gcs_import_uri} --project={args.project_id} " + f"--database={args.database} --quiet --format=json", + "run target-counts.sql from a trusted VPC/Cloud SQL connector path and compare " + "source_total_rows/source_table_count", + ], + "not_proven_by_this_artifact": [ + "GCS object generation until EXECUTE=1 succeeds", + "Cloud SQL import operation until EXECUTE=1 succeeds", + "Cloud SQL query readback until target-counts.sql is run from a trusted VPC/connector path", + ], + } + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--source-summary", type=Path, required=True) + parser.add_argument("--sqlite-backup", type=Path, required=True) + parser.add_argument("--import-sql", type=Path, required=True) + parser.add_argument("--target-counts-sql", type=Path, required=True) + parser.add_argument("--proof-json", type=Path, required=True) + parser.add_argument("--project-id", required=True) + parser.add_argument("--instance", required=True) + parser.add_argument("--database", required=True) + parser.add_argument("--region", required=True) + parser.add_argument("--gcs-import-uri", required=True) + parser.add_argument("--execute", action="store_true") + return parser.parse_args() + + +def main() -> int: + args = parse_args() + proof = build_proof(args) + args.proof_json.write_text(json.dumps(proof, indent=2, sort_keys=True) + "\n", encoding="utf-8") + print( + json.dumps( + { + "artifact": proof["artifact"], + "mode": proof["mode"], + "status": proof["status"], + "source_integrity_check": proof["source_integrity_check"], + "source_table_count": proof["source_table_count"], + "source_total_rows": proof["source_total_rows"], + "gcs_import_uri": proof["gcs_import_uri"], + "proof": str(args.proof_json), + }, + indent=2, + sort_keys=True, + ) + ) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/ops/run_gcp_cloudsql_restore_drill.sh b/ops/run_gcp_cloudsql_restore_drill.sh index db888d4..dc7ad9d 100755 --- a/ops/run_gcp_cloudsql_restore_drill.sh +++ b/ops/run_gcp_cloudsql_restore_drill.sh @@ -54,73 +54,25 @@ python3 ops/sqlite_to_postgres_dump.py \ --target-counts-sql "$TARGET_COUNTS_SQL" \ > "$CONVERT_LOG" -python3 - "$SOURCE_SUMMARY" "$SQLITE_BACKUP" "$IMPORT_SQL" "$TARGET_COUNTS_SQL" "$PROOF_JSON" "$PROJECT_ID" "$INSTANCE" "$DATABASE" "$REGION" "$GCS_IMPORT_URI" "$EXECUTE" <<'PY' -import hashlib -import json -import sys -from datetime import UTC, datetime -from pathlib import Path +execute_arg=() +if [[ "$EXECUTE" == "1" ]]; then + execute_arg=(--execute) +fi -summary_path = Path(sys.argv[1]) -sqlite_backup = Path(sys.argv[2]) -import_sql = Path(sys.argv[3]) -target_counts_sql = Path(sys.argv[4]) -proof_path = Path(sys.argv[5]) -project_id, instance, database, region, gcs_import_uri, execute = sys.argv[6:12] -summary = json.loads(summary_path.read_text()) - -def sha256(path: Path) -> str: - digest = hashlib.sha256() - with path.open("rb") as handle: - for chunk in iter(lambda: handle.read(1024 * 1024), b""): - digest.update(chunk) - return digest.hexdigest() - -proof = { - "artifact": "gcp_cloudsql_restore_drill", - "generated_at_utc": datetime.now(UTC).isoformat(), - "mode": "execute" if execute == "1" else "dry_run", - "status": "prepared" if execute != "1" else "prepared_for_execute", - "project_id": project_id, - "region": region, - "cloudsql_instance": instance, - "database": database, - "postgres_schema": summary["schema"], - "source_sqlite_backup": str(sqlite_backup), - "source_sqlite_backup_sha256": sha256(sqlite_backup), - "source_integrity_check": summary["source_integrity_check"], - "source_table_count": summary["table_count"], - "source_total_rows": summary["source_total_rows"], - "source_table_counts": {table["name"]: table["row_count"] for table in summary["tables"]}, - "gcs_import_uri": gcs_import_uri, - "local_private_artifacts": { - "postgres_import_sql": str(import_sql), - "source_summary_json": str(summary_path), - "target_counts_sql": str(target_counts_sql), - }, - "planned_commands": [ - f"gcloud storage cp {import_sql} {gcs_import_uri}", - f"gcloud sql import sql {instance} {gcs_import_uri} --project={project_id} --database={database} --quiet --format=json", - "run target-counts.sql from a trusted VPC/Cloud SQL connector path and compare source_total_rows/source_table_count", - ], - "not_proven_by_this_artifact": [ - "GCS object generation until EXECUTE=1 succeeds", - "Cloud SQL import operation until EXECUTE=1 succeeds", - "Cloud SQL query readback until target-counts.sql is run from a trusted VPC/connector path", - ], -} -proof_path.write_text(json.dumps(proof, indent=2, sort_keys=True) + "\n") -print(json.dumps({ - "artifact": proof["artifact"], - "mode": proof["mode"], - "status": proof["status"], - "source_integrity_check": proof["source_integrity_check"], - "source_table_count": proof["source_table_count"], - "source_total_rows": proof["source_total_rows"], - "gcs_import_uri": proof["gcs_import_uri"], - "proof": str(proof_path), -}, indent=2, sort_keys=True)) -PY +# The helper records source_table_counts in the proof manifest without a large +# stdin heredoc, which can deadlock on constrained macOS Bash pipe buffers. +python3 ops/build_gcp_cloudsql_restore_proof.py \ + --source-summary "$SOURCE_SUMMARY" \ + --sqlite-backup "$SQLITE_BACKUP" \ + --import-sql "$IMPORT_SQL" \ + --target-counts-sql "$TARGET_COUNTS_SQL" \ + --proof-json "$PROOF_JSON" \ + --project-id "$PROJECT_ID" \ + --instance "$INSTANCE" \ + --database "$DATABASE" \ + --region "$REGION" \ + --gcs-import-uri "$GCS_IMPORT_URI" \ + "${execute_arg[@]}" if [[ "$EXECUTE" != "1" ]]; then exit 0 diff --git a/scripts/gcp_iap_operator.sh b/scripts/gcp_iap_operator.sh index 7f0e816..137c3b4 100755 --- a/scripts/gcp_iap_operator.sh +++ b/scripts/gcp_iap_operator.sh @@ -21,6 +21,8 @@ readonly CORY_REPLAY_LEGACY_DIR="/home/teleo/gcp-cory-model-replay-20260712t1940 readonly DISABLED_ROLLBACK_DB="teleo_canonical_pre_20260712t1905z" readonly REQUEST_ID_RE='^iap-[a-z0-9]{12,32}$' readonly TARGET_DB_RE='^teleo_clone_[a-z0-9][a-z0-9_]{0,50}$' +readonly SCRIPT_DIR="$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" && pwd)" +readonly RESULT_HELPER="$SCRIPT_DIR/gcp_iap_operator_result.py" RUNNER_TEMP_DIR="" VALIDATED_BUNDLE_DIR="" @@ -600,69 +602,13 @@ write_sanitized_result() { local request_id="$4" local target_db="$5" local exit_code="$6" - RESULT_PATH="$output_path" RAW_STDOUT="$raw_stdout" OPERATION="$operation" \ - REQUEST_ID="$request_id" TARGET_DB="$target_db" EXIT_CODE="$exit_code" python3 - <<'PY' -import hashlib -import json -import os -from pathlib import Path - -raw = Path(os.environ["RAW_STDOUT"]).read_bytes() -exit_code = int(os.environ["EXIT_CODE"]) -remote = {} -if exit_code == 0: - try: - candidate = json.loads(raw.decode("utf-8").splitlines()[-1]) - allowed = { - "operation", - "request_id", - "target_db", - "status", - "instance", - "expected_internal_ip", - "active_state", - "sub_state", - "nrestarts", - "dispatcher_version", - "dispatcher_sha256", - "bundle_commit", - "bundle_sha256", - "result_sha256", - "result_bytes", - "full_result_export_ready", - "fixed_baseline_validation", - "no_message_send", - "live_service_unchanged", - "live_profile_unchanged", - "clone_database_remaining", - "run_directory_remaining", - "legacy_directory_remaining", - "export_file_remaining", - "leftover_temp_directory_count", - "rollback_database_datallowconn", - "rollback_database_connections", - } - remote = {key: value for key, value in candidate.items() if key in allowed} - except (UnicodeDecodeError, json.JSONDecodeError, IndexError): - exit_code = 90 - -payload = { - "schema": "livingip.gcpIapOperatorSanitizedResult.v1", - "operation": os.environ["OPERATION"], - "request_id": os.environ["REQUEST_ID"], - "target_db": os.environ["TARGET_DB"], - "status": "pass" if exit_code == 0 and remote.get("status") == "pass" else "fail", - "ssh_exit_code": exit_code, - "remote_result": remote, - "remote_stdout_sha256": hashlib.sha256(raw).hexdigest(), - "remote_stdout_bytes": len(raw), - "raw_stdout_retained": False, - "raw_stderr_retained": False, - "credential_values_logged": False, - "ssh_debug_enabled": False, -} -Path(os.environ["RESULT_PATH"]).write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8") -PY + python3 "$RESULT_HELPER" sanitize \ + --result-path "$output_path" \ + --raw-stdout "$raw_stdout" \ + --operation "$operation" \ + --request-id "$request_id" \ + --target-db "$target_db" \ + --exit-code "$exit_code" chmod 0600 "$output_path" } @@ -685,83 +631,22 @@ validate_downloaded_replay() { local result_path="$1" local full_result="$2" local target_db="$3" - RESULT_PATH="$result_path" FULL_RESULT="$full_result" TARGET_DB="$target_db" \ - EXPECTED_CORY_REPLAY_DB="$CORY_REPLAY_DB" \ - EXPECTED_CORY_REPLAY_FINGERPRINT="$CORY_REPLAY_FINGERPRINT" \ - python3 - <<'PY' -import hashlib -import json -import os -from pathlib import Path - -result_path = Path(os.environ["RESULT_PATH"]) -full_path = Path(os.environ["FULL_RESULT"]) -result = json.loads(result_path.read_text(encoding="utf-8")) -full_bytes = full_path.read_bytes() -full = json.loads(full_bytes) -remote = result.get("remote_result") or {} -checks = full.get("checks") or {} -required = { - "result_hash": hashlib.sha256(full_bytes).hexdigest() == remote.get("result_sha256"), - "result_size": len(full_bytes) == remote.get("result_bytes"), - "status": full.get("status") == "pass", - "target": full.get("target_database") == os.environ["TARGET_DB"], - "strict_six_of_six": (full.get("score") or {}).get("passes") == 6 - and (full.get("score") or {}).get("expected_prompt_count") == 6, - "all_checks": bool(checks) and all(checks.values()), -} -if os.environ["TARGET_DB"] == os.environ["EXPECTED_CORY_REPLAY_DB"]: - expected_counts = { - "claims": 1837, - "sources": 4145, - "claim_edges": 4916, - "claim_evidence": 4670, - "kb_proposals": 26, - } - before_counts = ((full.get("canonical_status_before") or {}).get("high_signal_rows") or {}) - after_counts = ((full.get("canonical_status_after") or {}).get("high_signal_rows") or {}) - required.update( - { - "fixed_fingerprint_before": (full.get("database_fingerprint_before") or {}).get("sha256") - == os.environ["EXPECTED_CORY_REPLAY_FINGERPRINT"], - "fixed_fingerprint_after": (full.get("database_fingerprint_after") or {}).get("sha256") - == os.environ["EXPECTED_CORY_REPLAY_FINGERPRINT"], - "fixed_counts_before": all(before_counts.get(key) == value for key, value in expected_counts.items()), - "fixed_counts_after": all(after_counts.get(key) == value for key, value in expected_counts.items()), - } - ) -failed = sorted(name for name, passed in required.items() if not passed) -if failed: - raise SystemExit("downloaded replay receipt failed: " + ", ".join(failed)) -result["full_result_exported_and_validated"] = True -result["full_result_retained"] = False -result["full_result_sha256"] = hashlib.sha256(full_bytes).hexdigest() -result["full_result_bytes"] = len(full_bytes) -result["full_result_fixed_checks"] = required -result_path.write_text(json.dumps(result, indent=2, sort_keys=True) + "\n", encoding="utf-8") -full_path.unlink() -PY + python3 "$RESULT_HELPER" validate-replay \ + --result-path "$result_path" \ + --full-result "$full_result" \ + --target-db "$target_db" \ + --expected-cory-replay-db "$CORY_REPLAY_DB" \ + --expected-cory-replay-fingerprint "$CORY_REPLAY_FINGERPRINT" } mark_result_export_failure() { local result_path="$1" local exit_code="$2" local cleanup_exit_code="$3" - RESULT_PATH="$result_path" EXIT_CODE="$exit_code" CLEANUP_EXIT_CODE="$cleanup_exit_code" python3 - <<'PY' -import json -import os -from pathlib import Path - -path = Path(os.environ["RESULT_PATH"]) -payload = json.loads(path.read_text(encoding="utf-8")) -payload["status"] = "fail" -payload["full_result_exported_and_validated"] = False -payload["full_result_retained"] = False -payload["full_result_export_failure"] = "receipt_download_validation_or_remote_export_cleanup_failed" -payload["ssh_exit_code"] = int(os.environ["EXIT_CODE"]) -payload["remote_full_result_export_removed"] = int(os.environ["CLEANUP_EXIT_CODE"]) == 0 -path.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8") -PY + python3 "$RESULT_HELPER" mark-export-failure \ + --result-path "$result_path" \ + --exit-code "$exit_code" \ + --cleanup-exit-code "$cleanup_exit_code" } runner_main() { @@ -892,16 +777,7 @@ runner_main() { if [[ "$exit_code" -ne 0 ]]; then mark_result_export_failure "$result_path" "$exit_code" "$export_cleanup_exit_code" else - RESULT_PATH="$result_path" python3 - <<'PY' -import json -import os -from pathlib import Path - -path = Path(os.environ["RESULT_PATH"]) -payload = json.loads(path.read_text(encoding="utf-8")) -payload["remote_full_result_export_removed"] = True -path.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8") -PY + python3 "$RESULT_HELPER" mark-export-removed --result-path "$result_path" fi fi printf 'sanitized_result=%s\n' "$result_path" diff --git a/scripts/gcp_iap_operator_result.py b/scripts/gcp_iap_operator_result.py new file mode 100755 index 0000000..d3995bd --- /dev/null +++ b/scripts/gcp_iap_operator_result.py @@ -0,0 +1,179 @@ +#!/usr/bin/env python3 +"""Sanitize and validate local GCP IAP operator result artifacts.""" + +from __future__ import annotations + +import argparse +import hashlib +import json +from pathlib import Path +from typing import Any + +ALLOWED_REMOTE_FIELDS = { + "operation", + "request_id", + "target_db", + "status", + "instance", + "expected_internal_ip", + "active_state", + "sub_state", + "nrestarts", + "dispatcher_version", + "dispatcher_sha256", + "bundle_commit", + "bundle_sha256", + "result_sha256", + "result_bytes", + "full_result_export_ready", + "fixed_baseline_validation", + "no_message_send", + "live_service_unchanged", + "live_profile_unchanged", + "clone_database_remaining", + "run_directory_remaining", + "legacy_directory_remaining", + "export_file_remaining", + "leftover_temp_directory_count", + "rollback_database_datallowconn", + "rollback_database_connections", +} + + +def write_sanitized_result(args: argparse.Namespace) -> int: + raw = args.raw_stdout.read_bytes() + exit_code = args.exit_code + remote: dict[str, Any] = {} + if exit_code == 0: + try: + candidate = json.loads(raw.decode("utf-8").splitlines()[-1]) + remote = {key: value for key, value in candidate.items() if key in ALLOWED_REMOTE_FIELDS} + except (AttributeError, UnicodeDecodeError, json.JSONDecodeError, IndexError): + exit_code = 90 + + payload = { + "schema": "livingip.gcpIapOperatorSanitizedResult.v1", + "operation": args.operation, + "request_id": args.request_id, + "target_db": args.target_db, + "status": "pass" if exit_code == 0 and remote.get("status") == "pass" else "fail", + "ssh_exit_code": exit_code, + "remote_result": remote, + "remote_stdout_sha256": hashlib.sha256(raw).hexdigest(), + "remote_stdout_bytes": len(raw), + "raw_stdout_retained": False, + "raw_stderr_retained": False, + "credential_values_logged": False, + "ssh_debug_enabled": False, + } + args.result_path.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8") + return 0 + + +def validate_downloaded_replay(args: argparse.Namespace) -> int: + result = json.loads(args.result_path.read_text(encoding="utf-8")) + full_bytes = args.full_result.read_bytes() + full = json.loads(full_bytes) + remote = result.get("remote_result") or {} + checks = full.get("checks") or {} + required = { + "result_hash": hashlib.sha256(full_bytes).hexdigest() == remote.get("result_sha256"), + "result_size": len(full_bytes) == remote.get("result_bytes"), + "status": full.get("status") == "pass", + "target": full.get("target_database") == args.target_db, + "strict_six_of_six": (full.get("score") or {}).get("passes") == 6 + and (full.get("score") or {}).get("expected_prompt_count") == 6, + "all_checks": bool(checks) and all(checks.values()), + } + if args.target_db == args.expected_cory_replay_db: + expected_counts = { + "claims": 1837, + "sources": 4145, + "claim_edges": 4916, + "claim_evidence": 4670, + "kb_proposals": 26, + } + before_counts = (full.get("canonical_status_before") or {}).get("high_signal_rows") or {} + after_counts = (full.get("canonical_status_after") or {}).get("high_signal_rows") or {} + required.update( + { + "fixed_fingerprint_before": (full.get("database_fingerprint_before") or {}).get("sha256") + == args.expected_cory_replay_fingerprint, + "fixed_fingerprint_after": (full.get("database_fingerprint_after") or {}).get("sha256") + == args.expected_cory_replay_fingerprint, + "fixed_counts_before": all(before_counts.get(key) == value for key, value in expected_counts.items()), + "fixed_counts_after": all(after_counts.get(key) == value for key, value in expected_counts.items()), + } + ) + failed = sorted(name for name, passed in required.items() if not passed) + if failed: + raise SystemExit("downloaded replay receipt failed: " + ", ".join(failed)) + result["full_result_exported_and_validated"] = True + result["full_result_retained"] = False + result["full_result_sha256"] = hashlib.sha256(full_bytes).hexdigest() + result["full_result_bytes"] = len(full_bytes) + result["full_result_fixed_checks"] = required + args.result_path.write_text(json.dumps(result, indent=2, sort_keys=True) + "\n", encoding="utf-8") + args.full_result.unlink() + return 0 + + +def mark_export_failure(args: argparse.Namespace) -> int: + payload = json.loads(args.result_path.read_text(encoding="utf-8")) + payload["status"] = "fail" + payload["full_result_exported_and_validated"] = False + payload["full_result_retained"] = False + payload["full_result_export_failure"] = "receipt_download_validation_or_remote_export_cleanup_failed" + payload["ssh_exit_code"] = args.exit_code + payload["remote_full_result_export_removed"] = args.cleanup_exit_code == 0 + args.result_path.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8") + return 0 + + +def mark_export_removed(args: argparse.Namespace) -> int: + payload = json.loads(args.result_path.read_text(encoding="utf-8")) + payload["remote_full_result_export_removed"] = True + args.result_path.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8") + return 0 + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description=__doc__) + subparsers = parser.add_subparsers(dest="command", required=True) + + sanitize = subparsers.add_parser("sanitize") + sanitize.add_argument("--result-path", type=Path, required=True) + sanitize.add_argument("--raw-stdout", type=Path, required=True) + sanitize.add_argument("--operation", required=True) + sanitize.add_argument("--request-id", required=True) + sanitize.add_argument("--target-db", required=True) + sanitize.add_argument("--exit-code", type=int, required=True) + sanitize.set_defaults(handler=write_sanitized_result) + + replay = subparsers.add_parser("validate-replay") + replay.add_argument("--result-path", type=Path, required=True) + replay.add_argument("--full-result", type=Path, required=True) + replay.add_argument("--target-db", required=True) + replay.add_argument("--expected-cory-replay-db", required=True) + replay.add_argument("--expected-cory-replay-fingerprint", required=True) + replay.set_defaults(handler=validate_downloaded_replay) + + export_failure = subparsers.add_parser("mark-export-failure") + export_failure.add_argument("--result-path", type=Path, required=True) + export_failure.add_argument("--exit-code", type=int, required=True) + export_failure.add_argument("--cleanup-exit-code", type=int, required=True) + export_failure.set_defaults(handler=mark_export_failure) + + export_removed = subparsers.add_parser("mark-export-removed") + export_removed.add_argument("--result-path", type=Path, required=True) + export_removed.set_defaults(handler=mark_export_removed) + return parser.parse_args() + + +def main() -> int: + args = parse_args() + return args.handler(args) + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_gcp_iap_operator_access.py b/tests/test_gcp_iap_operator_access.py index ef01fb4..46d58de 100644 --- a/tests/test_gcp_iap_operator_access.py +++ b/tests/test_gcp_iap_operator_access.py @@ -10,6 +10,7 @@ REPO_ROOT = Path(__file__).resolve().parents[1] PLAN = REPO_ROOT / "ops" / "plan_gcp_iap_operator_access.py" WORKFLOW = REPO_ROOT / ".github" / "workflows" / "gcp-iap-operator.yml" OPERATOR = REPO_ROOT / "scripts" / "gcp_iap_operator.sh" +RESULT_HELPER = REPO_ROOT / "scripts" / "gcp_iap_operator_result.py" def run_plan(*args: str) -> str: @@ -245,6 +246,7 @@ def test_operator_cleanup_is_bound_to_marker_clone_prefix_and_exact_run_dir() -> def test_operator_exports_and_fixed_validates_full_replay_before_cleanup() -> None: source = OPERATOR.read_text(encoding="utf-8") + result_helper = RESULT_HELPER.read_text(encoding="utf-8") assert 'CORY_REPLAY_FINGERPRINT="48ea5332c4f303dea02d503447f1ac8e1d8aa00e60d198a566d628b37dbdac1c"' in source assert 'export_path="$caller_home/.teleo-iap-result-$request_id.json"' in source @@ -255,10 +257,10 @@ def test_operator_exports_and_fixed_validates_full_replay_before_cleanup() -> No assert source.index('validate_downloaded_replay "$result_path" "$full_result" "$target_db"') < source.index( 'rm -f -- "$full_result"' ) - assert '"strict_six_of_six"' in source - assert '"fixed_fingerprint_before"' in source - assert '"fixed_counts_after"' in source - assert 'payload["remote_full_result_export_removed"] = True' in source + assert '"strict_six_of_six"' in result_helper + assert '"fixed_fingerprint_before"' in result_helper + assert '"fixed_counts_after"' in result_helper + assert 'payload["remote_full_result_export_removed"] = True' in result_helper def test_operator_uses_five_minute_key_private_modes_and_cleanup_trap() -> None: @@ -390,35 +392,31 @@ def test_clone_runner_scps_only_request_derived_bundle_over_iap(tmp_path: Path, observed_fingerprint = expected_fingerprint if valid_receipt else "0" * 64 full_receipt.write_text( json.dumps( - { - "status": "pass", - "target_database": "teleo_clone_cory_20260712t1940z", - "score": {"passes": 6, "expected_prompt_count": 6}, - "checks": {"all_runtime_checks": True}, - "database_fingerprint_before": { - "sha256": observed_fingerprint - }, - "database_fingerprint_after": { - "sha256": observed_fingerprint - }, - "canonical_status_before": { - "high_signal_rows": { - "claims": 1837, - "sources": 4145, - "claim_edges": 4916, - "claim_evidence": 4670, - "kb_proposals": 26, - } - }, - "canonical_status_after": { - "high_signal_rows": { - "claims": 1837, - "sources": 4145, - "claim_edges": 4916, - "claim_evidence": 4670, - "kb_proposals": 26, - } - }, + { + "status": "pass", + "target_database": "teleo_clone_cory_20260712t1940z", + "score": {"passes": 6, "expected_prompt_count": 6}, + "checks": {"all_runtime_checks": True}, + "database_fingerprint_before": {"sha256": observed_fingerprint}, + "database_fingerprint_after": {"sha256": observed_fingerprint}, + "canonical_status_before": { + "high_signal_rows": { + "claims": 1837, + "sources": 4145, + "claim_edges": 4916, + "claim_evidence": 4670, + "kb_proposals": 26, + } + }, + "canonical_status_after": { + "high_signal_rows": { + "claims": 1837, + "sources": 4145, + "claim_edges": 4916, + "claim_evidence": 4670, + "kb_proposals": 26, + } + }, }, sort_keys=True, )