chore: commit current workspace changes
This commit is contained in:
@@ -40,7 +40,7 @@ def test_post_contract_includes_execution_id_and_path_uses_asset_id():
|
||||
adapter.record_trade(_trade(asset="PF-77"))
|
||||
assert captured["url"] == "https://portfolio-hub.test/v1/assets/PF-77/ledger/trades"
|
||||
assert "execution_id" in captured["body"]
|
||||
assert "\"side\": \"buy\"" in captured["body"]
|
||||
assert "\"side\":\"buy\"" in captured["body"]
|
||||
assert captured["auth"] == "Bearer tok"
|
||||
|
||||
|
||||
|
||||
@@ -11,7 +11,13 @@ import psycopg
|
||||
from trex_core.adapters.persistence.migrate import apply_migrations, migration_files
|
||||
from trex_core.domain.slot import EXPECTED_SLOT_FIELDS
|
||||
|
||||
_EXPECTED_TABLES = {"daily_slot", "daily_symbol_entries", "slot_transition_log", "intent_log"}
|
||||
_EXPECTED_TABLES = {
|
||||
"daily_slot",
|
||||
"daily_symbol_entries",
|
||||
"slot_transition_log",
|
||||
"intent_log",
|
||||
"ledger_outbox", # BAT-006 / REQ-019
|
||||
}
|
||||
|
||||
|
||||
def _columns(dsn: str, table: str) -> set[str]:
|
||||
|
||||
@@ -0,0 +1,72 @@
|
||||
"""PostgreSQL ledger outbox integration tests (REQ-019).
|
||||
|
||||
Skipped without TREX_POSTGRESQL_TEST_DSN (same gate as other PG suites).
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime, timezone
|
||||
|
||||
from trex_core.adapters.ledger.in_memory import InMemoryLedger
|
||||
from trex_core.adapters.persistence.postgresql import PostgreSQLLedgerOutbox
|
||||
from trex_core.application.ledger_outbox import (
|
||||
LedgerWriter,
|
||||
OutboxOutcome,
|
||||
WorkItemStatus,
|
||||
)
|
||||
from trex_core.application.ledger_port import LedgerTrade
|
||||
|
||||
|
||||
def _trade(execution_id="paper:1:10") -> LedgerTrade:
|
||||
return LedgerTrade(
|
||||
asset_id="PF-1", order_id="ORD-1", execution_id=execution_id,
|
||||
side="buy", qty=10, price=1000.0, fee=1.5, tax=0.0,
|
||||
executed_at=datetime(2026, 6, 4, tzinfo=timezone.utc),
|
||||
)
|
||||
|
||||
|
||||
def test_enqueue_persists_and_is_idempotent(pg_clean):
|
||||
outbox = PostgreSQLLedgerOutbox(pg_clean)
|
||||
a = outbox.enqueue(_trade())
|
||||
b = outbox.enqueue(_trade())
|
||||
assert a.correlation_id == b.correlation_id
|
||||
pending = outbox.list_pending()
|
||||
assert len(pending) == 1
|
||||
assert pending[0].status is WorkItemStatus.PENDING
|
||||
|
||||
|
||||
def test_writer_completes_persisted_item(pg_clean):
|
||||
outbox = PostgreSQLLedgerOutbox(pg_clean)
|
||||
ledger = InMemoryLedger()
|
||||
writer = LedgerWriter(outbox, ledger)
|
||||
outbox.enqueue(_trade(execution_id="pg-1"))
|
||||
stats = writer.run_once()
|
||||
assert stats.completed == 1 and ledger.count() == 1
|
||||
row = outbox.get("PF-1:ORD-1:pg-1")
|
||||
assert row is not None and row.status is WorkItemStatus.COMPLETED
|
||||
assert row.last_outcome is OutboxOutcome.COMPLETED
|
||||
|
||||
|
||||
def test_writer_marks_conflict_blocked_on_pg(pg_clean):
|
||||
outbox = PostgreSQLLedgerOutbox(pg_clean)
|
||||
ledger = InMemoryLedger(conflict_executions={"pg-x"})
|
||||
writer = LedgerWriter(outbox, ledger)
|
||||
outbox.enqueue(_trade(execution_id="pg-x"))
|
||||
writer.run_once()
|
||||
row = outbox.get("PF-1:ORD-1:pg-x")
|
||||
assert row.status is WorkItemStatus.CONFLICT_BLOCKED
|
||||
assert row.last_outcome is OutboxOutcome.CONFLICT_BLOCKED
|
||||
|
||||
|
||||
def test_writer_retries_on_pg(pg_clean):
|
||||
outbox = PostgreSQLLedgerOutbox(pg_clean)
|
||||
ledger = InMemoryLedger(fail_n_then_succeed=1)
|
||||
writer = LedgerWriter(outbox, ledger)
|
||||
outbox.enqueue(_trade(execution_id="pg-r"))
|
||||
writer.run_once()
|
||||
row1 = outbox.get("PF-1:ORD-1:pg-r")
|
||||
assert row1.status is WorkItemStatus.PENDING
|
||||
assert row1.last_outcome is OutboxOutcome.RETRYABLE_FAILURE
|
||||
writer.run_once()
|
||||
row2 = outbox.get("PF-1:ORD-1:pg-r")
|
||||
assert row2.status is WorkItemStatus.COMPLETED
|
||||
assert ledger.count() == 1
|
||||
@@ -89,19 +89,26 @@ def test_portfolio_id_absent_skips_ledger_but_transitions_slot():
|
||||
|
||||
|
||||
def test_duplicate_execution_id_results_in_zero_duplicate_in_ledger():
|
||||
"""REQ-019 AC-1: same execution_id enqueued twice = 1 ledger entry."""
|
||||
"""REQ-019 AC-1: same execution_id processed twice = 1 ledger entry."""
|
||||
ledger = InMemoryLedger()
|
||||
repo, svc, intents, execution, outbox, writer = _wire(PaperBroker(fill_after_polls=0),
|
||||
ledger=ledger)
|
||||
_buy(intents).slot_id
|
||||
_buy(intents)
|
||||
execution.run_once()
|
||||
writer.run_once()
|
||||
assert ledger.count() == 1
|
||||
# Re-enqueue the same trade explicitly (simulate accidental re-fill).
|
||||
# Same correlation_id arriving again (re-enqueue) is dedup'd at the outbox.
|
||||
first = outbox.all_items()[0]
|
||||
outbox.enqueue(first.trade) # idempotent on correlation_id
|
||||
stats = writer.run_once() # still terminal-completed; reports duplicate
|
||||
assert stats.duplicates >= 1
|
||||
outbox.enqueue(first.trade)
|
||||
assert len(outbox.all_items()) == 1
|
||||
# And even if the writer somehow re-processes (simulated by a fresh outbox
|
||||
# row for the same execution_id, e.g. after a manual replay), the ledger
|
||||
# adapter dedup'd it server-side and the outcome is duplicate_completed.
|
||||
replay_outbox = InMemoryLedgerOutbox()
|
||||
replay_writer = LedgerWriter(replay_outbox, ledger)
|
||||
replay_outbox.enqueue(first.trade)
|
||||
stats = replay_writer.run_once()
|
||||
assert stats.duplicates == 1 and stats.completed == 0
|
||||
assert ledger.count() == 1 # no duplicate in the ledger
|
||||
|
||||
|
||||
|
||||
+7
-1
@@ -9,7 +9,13 @@ import os
|
||||
|
||||
import pytest
|
||||
|
||||
_PG_TABLES = ("intent_log", "slot_transition_log", "daily_symbol_entries", "daily_slot")
|
||||
_PG_TABLES = (
|
||||
"ledger_outbox",
|
||||
"intent_log",
|
||||
"slot_transition_log",
|
||||
"daily_symbol_entries",
|
||||
"daily_slot",
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
|
||||
Reference in New Issue
Block a user