feat(ops): add release provenance gate and audit checkout isolation

Co-authored-by: genichin <4282227+genichin@users.noreply.github.com>
This commit is contained in:
copilot-swe-agent[bot]
2026-09-13 16:12:55 +09:00
committed by Hermes Agent
co-authored by genichin
parent a6ae800346
commit f3a6b2c410
4 changed files with 392 additions and 11 deletions
+78 -4
View File
@@ -7,13 +7,19 @@ find_ledger_gap_slots detects only the gaps.
"""
from __future__ import annotations
import subprocess
import uuid
from datetime import date, datetime, timezone
from datetime import UTC, date, datetime
from pathlib import Path
from trex_core.adapters.persistence.ops_checks import find_ledger_gap_slots
from trex_core.adapters.persistence.ops_checks import (
collect_release_provenance,
find_ledger_gap_slots,
isolate_dirty_checkout_files,
)
_DAY = date(2026, 6, 10)
_TS = datetime(2026, 6, 10, 6, 0, tzinfo=timezone.utc)
_TS = datetime(2026, 6, 10, 6, 0, tzinfo=UTC)
def _insert_slot(conn, *, status, portfolio_id, symbol="005930"):
@@ -79,10 +85,78 @@ def test_no_gaps_when_all_recorded(pg_clean):
def test_date_scope_filters_by_sell_fill_day(pg_clean):
import psycopg
dsn = pg_clean
other_day_ts = datetime(2026, 6, 9, 6, 0, tzinfo=timezone.utc)
other_day_ts = datetime(2026, 6, 9, 6, 0, tzinfo=UTC)
with psycopg.connect(dsn, autocommit=True) as conn:
sid = _insert_slot(conn, status="inactive", portfolio_id="PF-OLD")
_insert_sell_fill(conn, sid, ts=other_day_ts)
assert find_ledger_gap_slots(dsn, trading_date=_DAY) == []
gaps = find_ledger_gap_slots(dsn, trading_date=date(2026, 6, 9))
assert {str(g["slot_id"]) for g in gaps} == {sid}
def _init_git_repo(tmp_path: Path) -> tuple[Path, str]:
repo = tmp_path / "repo"
repo.mkdir()
subprocess.run(["git", "init"], cwd=repo, check=True, capture_output=True)
subprocess.run(["git", "config", "user.name", "test"], cwd=repo, check=True, capture_output=True)
subprocess.run(["git", "config", "user.email", "test@example.com"], cwd=repo, check=True, capture_output=True)
(repo / "tracked.txt").write_text("v1\n", encoding="utf-8")
subprocess.run(["git", "add", "tracked.txt"], cwd=repo, check=True, capture_output=True)
subprocess.run(["git", "commit", "-m", "init"], cwd=repo, check=True, capture_output=True)
head = subprocess.check_output(["git", "rev-parse", "HEAD"], cwd=repo, text=True).strip()
return repo, head
def test_collect_release_provenance_passes_on_clean_checkout(tmp_path):
repo, head = _init_git_repo(tmp_path)
report = collect_release_provenance(
actor="operator-a",
reason="deploy",
instance="inst-a",
command_class="manual-start",
checkout_dir=repo,
expected_oci_revision=head,
image_digest="sha256:" + ("a" * 64),
runtime_files=["tracked.txt"],
capture_commands={},
broker_log_path=None,
)
assert report["checkout"]["git_clean"] is True
assert report["artifact_identity"]["oci_revision_matches_head"] is True
assert report["runtime_file_verification"][0]["matches_expected_revision"] is True
assert report["gate"]["passed"] is True
def test_collect_release_provenance_blocks_on_revision_or_runtime_mismatch(tmp_path):
repo, old_head = _init_git_repo(tmp_path)
(repo / "tracked.txt").write_text("v2\n", encoding="utf-8")
subprocess.run(["git", "add", "tracked.txt"], cwd=repo, check=True, capture_output=True)
subprocess.run(["git", "commit", "-m", "v2"], cwd=repo, check=True, capture_output=True)
report = collect_release_provenance(
actor="operator-a",
reason="deploy",
instance="inst-a",
command_class="manual-start",
checkout_dir=repo,
expected_oci_revision=old_head,
runtime_files=["tracked.txt"],
capture_commands={},
broker_log_path=None,
)
assert report["checkout"]["git_clean"] is True
assert report["artifact_identity"]["oci_revision_matches_head"] is False
assert report["runtime_file_verification"][0]["matches_expected_revision"] is False
assert report["gate"]["passed"] is False
def test_isolate_dirty_checkout_files_copies_dirty_and_untracked(tmp_path):
repo, _ = _init_git_repo(tmp_path)
(repo / "tracked.txt").write_text("dirty\n", encoding="utf-8")
(repo / "new.txt").write_text("new\n", encoding="utf-8")
isolated = tmp_path / "isolated"
manifest = isolate_dirty_checkout_files(checkout_dir=repo, isolate_dir=isolated)
assert set(manifest["dirty_paths"]) == {"tracked.txt", "new.txt"}
assert set(manifest["copied_paths"]) == {"tracked.txt", "new.txt"}
assert (isolated / "tracked.txt").read_text(encoding="utf-8") == "dirty\n"
assert (isolated / "new.txt").read_text(encoding="utf-8") == "new\n"
assert (isolated / "manifest.json").is_file()