test: validate ledger backfill flow

Co-authored-by: genichin <4282227+genichin@users.noreply.github.com>
This commit is contained in:
copilot-swe-agent[bot]
2026-08-24 10:25:35 +00:00
committed by GitHub
co-authored by genichin
parent e2c4bbba34
commit 3e74d35299
3 changed files with 17 additions and 14 deletions
+3 -3
View File
@@ -1,7 +1,7 @@
"""LedgerOutbox + LedgerWriter direct unit tests (REQ-019)."""
from __future__ import annotations
from datetime import datetime, timedelta, timezone
from datetime import UTC, datetime, timedelta
from trex_core.adapters.ledger.in_memory import InMemoryLedger
from trex_core.application.ledger_outbox import (
@@ -17,7 +17,7 @@ def _trade(execution_id="paper:PAPER-1:10") -> LedgerTrade:
return LedgerTrade(
asset_id="PF-1", order_id="PAPER-1", execution_id=execution_id,
side="buy", qty=10, price=1000.0,
executed_at=datetime(2026, 6, 4, tzinfo=timezone.utc),
executed_at=datetime(2026, 6, 4, tzinfo=UTC),
)
@@ -57,7 +57,7 @@ def test_writer_marks_conflict_blocked_and_does_not_retry():
def test_writer_retries_until_success():
# dcy-015: retryable rows now back off; share an injected clock between the
# outbox (list_pending "now") and the writer (next_retry_at) to advance past it.
now = {"t": datetime(2026, 6, 4, tzinfo=timezone.utc)}
now = {"t": datetime(2026, 6, 4, tzinfo=UTC)}
outbox = InMemoryLedgerOutbox(clock=lambda: now["t"])
ledger = InMemoryLedger(fail_n_then_succeed=1)
writer = LedgerWriter(outbox, ledger, clock=lambda: now["t"])