diff --git a/scripts/release_provenance_gate.py b/scripts/release_provenance_gate.py index b4b7a6b..8b68c09 100644 --- a/scripts/release_provenance_gate.py +++ b/scripts/release_provenance_gate.py @@ -77,7 +77,7 @@ def main(argv: list[str] | None = None) -> int: item.get("matches_expected_revision", False) for item in report["runtime_file_verification"] ) oci_ok = bool(report["artifact_identity"]["oci_revision_matches_head"]) - allow_dirty_passed = bool(args.allow_dirty and dirty_checkout and runtime_ok and oci_ok) + allow_dirty_passed = bool(args.allow_dirty and runtime_ok and oci_ok) report["gate"]["allow_dirty"] = bool(args.allow_dirty) report["gate"]["allow_dirty_passed"] = allow_dirty_passed diff --git a/tests/adapters/persistence/test_ops_checks.py b/tests/adapters/persistence/test_ops_checks.py index aea382b..9e4779f 100644 --- a/tests/adapters/persistence/test_ops_checks.py +++ b/tests/adapters/persistence/test_ops_checks.py @@ -12,6 +12,8 @@ import uuid from datetime import UTC, date, datetime from pathlib import Path +import pytest + from trex_core.adapters.persistence.ops_checks import ( collect_release_provenance, find_ledger_gap_slots, @@ -179,3 +181,18 @@ def test_collect_release_provenance_rejects_runtime_path_escape(tmp_path): assert runtime["path_within_checkout"] is False assert runtime["matches_expected_revision"] is False assert report["gate"]["passed"] is False + + +def test_collect_release_provenance_reports_git_failure_context(tmp_path): + not_repo = tmp_path / "not-repo" + not_repo.mkdir() + with pytest.raises(RuntimeError, match="git command failed in checkout="): + collect_release_provenance( + actor="operator-a", + reason="deploy", + instance="inst-a", + command_class="manual-start", + checkout_dir=not_repo, + capture_commands={}, + broker_log_path=None, + ) diff --git a/trex_core/adapters/persistence/ops_checks.py b/trex_core/adapters/persistence/ops_checks.py index de5cdff..b14182f 100644 --- a/trex_core/adapters/persistence/ops_checks.py +++ b/trex_core/adapters/persistence/ops_checks.py @@ -15,6 +15,7 @@ import hashlib import json import shutil import subprocess +from collections import deque from datetime import UTC, date, datetime from pathlib import Path from typing import Any @@ -56,11 +57,18 @@ def find_ledger_gap_slots( def _run_git(checkout_dir: Path, *args: str) -> str: - return subprocess.check_output( - ["git", *args], - cwd=str(checkout_dir), - text=True, - ).strip() + command = ["git", *args] + try: + return subprocess.check_output( + command, + cwd=str(checkout_dir), + text=True, + ).strip() + except subprocess.CalledProcessError as exc: + joined = " ".join(command) + raise RuntimeError( + f"git command failed in checkout={checkout_dir}: {joined}" + ) from exc def _porcelain_paths(status_output: str) -> list[str]: @@ -125,17 +133,18 @@ def _capture_command(command: list[str], checkout_dir: Path) -> dict: def _capture_broker_side_effect_logs(log_path: Path) -> dict: if not log_path.exists(): return {"path": str(log_path), "exists": False, "matches": []} - matches: list[str] = [] - for line in log_path.read_text(encoding="utf-8", errors="replace").splitlines(): - lowered = line.lower() - if ( - "event=order_submit" in lowered - or "event=order_fill" in lowered - or "event=order_reject" in lowered - or "cancel" in lowered - ): - matches.append(line) - return {"path": str(log_path), "exists": True, "matches": matches[-200:]} + matches: deque[str] = deque(maxlen=200) + with log_path.open("r", encoding="utf-8", errors="replace") as fp: + for line in fp: + lowered = line.lower() + if ( + "event=order_submit" in lowered + or "event=order_fill" in lowered + or "event=order_reject" in lowered + or "cancel" in lowered + ): + matches.append(line.rstrip("\n")) + return {"path": str(log_path), "exists": True, "matches": list(matches)} def collect_release_provenance(