from __future__ import annotations import json from types import SimpleNamespace from trex_core.application.ledger_outbox import OutboxOutcome from trex_core.application.ledger_port import LedgerTrade from trex_core.runtime import cli def _cfg(*, live: bool = False): return SimpleNamespace( dsn="postgresql://x/y", broker="kis", kis=SimpleNamespace(allow_live=live), portfolio_hub=SimpleNamespace(base_url="http://pf.local", token="tok", timeout=5.0), ) def _args(tmp_path, **overrides): data = { "date": "2026-08-24", "execution_id": "kis-daily-ccld:20260824:12345678:0001234567:10", "dry_run": False, "yes": True, "live": True, "json": True, "audit_dir": str(tmp_path), "instance": "inst-1", "config": None, "config_path": None, "env_file": None, "timeout": 10.0, "ledger_cmd": "backfill", "cmd": "ledger", } data.update(overrides) return SimpleNamespace(**data) def _evidence(): return { "symbol": "005930", "side": "sell", "qty": 10.0, "price": 70000.0, "fee": 105.0, "tax": 1400.0, "order_id": "12345678:0001234567", "order_venue": "SOR", "execution_venue": None, "executed_at": "2026-08-24T00:01:02+00:00", "raw": { "pdno": "005930", "sll_buy_dvsn_cd": "01", "tot_ccld_qty": "10", "avg_prvs": "70000", "ord_dt": "20260824", "ord_gno_brno": "12345678", "odno": "0001234567", "excg_id_dvsn_cd": "SOR", }, } def test_ledger_backfill_parser_accepts_required_flags(): parser = cli._build_parser() args = parser.parse_args([ "ledger", "backfill", "--date", "2026-08-24", "--execution-id", "kis-daily-ccld:20260824:12345678:0001234567:10", "--dry-run", ]) assert args.cmd == "ledger" assert args.ledger_cmd == "backfill" assert args.dry_run is True def test_ledger_backfill_dry_run_emits_trade_and_writes_audit(monkeypatch, tmp_path, capsys): monkeypatch.setattr(cli, "_lookup_ledger_backfill_evidence", lambda cfg, **kwargs: (_evidence(), SimpleNamespace(mode="env", identity_label="kis-config"))) monkeypatch.setattr(cli, "_resolve_ledger_backfill_slot", lambda dsn, **kwargs: {"slot_id": "slot-1", "portfolio_id": "PF-1", "status": "inactive", "symbol": "005930"}) monkeypatch.setattr(cli, "_lookup_outbox_by_execution_id", lambda dsn, **kwargs: []) rc = cli._cmd_ledger_backfill(_args(tmp_path, dry_run=True), _cfg(), lambda _: True) assert rc == 0 out = json.loads(capsys.readouterr().out) assert out["result"]["status"] == "dry_run" assert out["trade"]["correlation_id"] == "PF-1:12345678:0001234567:kis-daily-ccld:20260824:12345678:0001234567:10" assert out["trade"]["order_venue"] == "SOR" assert out["trade"]["execution_venue"] is None audit = json.loads(next(tmp_path.glob("*.json")).read_text(encoding="utf-8")) assert audit["request"]["execution_id"] == _args(tmp_path).execution_id def test_ledger_backfill_live_requires_live_flag(monkeypatch, tmp_path, capsys): monkeypatch.setattr(cli, "_lookup_ledger_backfill_evidence", lambda cfg, **kwargs: (_evidence(), SimpleNamespace(mode="env", identity_label="kis-config"))) monkeypatch.setattr(cli, "_resolve_ledger_backfill_slot", lambda dsn, **kwargs: {"slot_id": "slot-1", "portfolio_id": "PF-1", "status": "inactive", "symbol": "005930"}) monkeypatch.setattr(cli, "_lookup_outbox_by_execution_id", lambda dsn, **kwargs: []) rc = cli._cmd_ledger_backfill(_args(tmp_path, live=False), _cfg(live=True), lambda _: True) assert rc == 2 err = capsys.readouterr().err assert "--live" in err audit = json.loads(next(tmp_path.glob("*.json")).read_text(encoding="utf-8")) assert audit["result"]["status"] == "rejected" def test_ledger_backfill_existing_completed_row_is_idempotent(monkeypatch, tmp_path, capsys): monkeypatch.setattr(cli, "_lookup_ledger_backfill_evidence", lambda cfg, **kwargs: (_evidence(), SimpleNamespace(mode="env", identity_label="kis-config"))) monkeypatch.setattr(cli, "_resolve_ledger_backfill_slot", lambda dsn, **kwargs: {"slot_id": "slot-1", "portfolio_id": "PF-1", "status": "inactive", "symbol": "005930"}) existing = [{ "correlation_id": "PF-1:12345678:0001234567:kis-daily-ccld:20260824:12345678:0001234567:10", "asset_id": "PF-1", "order_id": "12345678:0001234567", "execution_id": "kis-daily-ccld:20260824:12345678:0001234567:10", "status": "completed", "attempts": 1, "last_outcome": "completed", "last_error": None, }] monkeypatch.setattr(cli, "_lookup_outbox_by_execution_id", lambda dsn, **kwargs: existing) class _Outbox: def enqueue(self, trade): raise AssertionError("completed execution must not be re-enqueued") def get(self, correlation_id): return SimpleNamespace( status=SimpleNamespace(value="completed"), attempts=1, last_outcome=SimpleNamespace(value="completed"), last_error=None, ) class _Writer: def __init__(self, *_args, **_kwargs): pass def process_correlation_id(self, correlation_id): raise AssertionError("completed execution must not be re-processed") import trex_core.adapters.ledger.portfolio_hub as hub_mod import trex_core.adapters.persistence.postgresql as pg_mod monkeypatch.setattr(pg_mod, "PostgreSQLLedgerOutbox", lambda dsn: _Outbox()) monkeypatch.setattr(hub_mod, "PortfolioHubLedger", lambda *args, **kwargs: object()) monkeypatch.setattr(cli, "LedgerWriter", _Writer) rc = cli._cmd_ledger_backfill(_args(tmp_path), _cfg(), lambda _: True) assert rc == 0 out = json.loads(capsys.readouterr().out) assert out["result"]["status"] == "already_recorded" def test_ledger_backfill_live_enqueue_and_processes_exact_trade(monkeypatch, tmp_path, capsys): monkeypatch.setattr(cli, "_lookup_ledger_backfill_evidence", lambda cfg, **kwargs: (_evidence(), SimpleNamespace(mode="env", identity_label="kis-config"))) monkeypatch.setattr(cli, "_resolve_ledger_backfill_slot", lambda dsn, **kwargs: {"slot_id": "slot-1", "portfolio_id": "PF-1", "status": "inactive", "symbol": "005930"}) monkeypatch.setattr(cli, "_lookup_outbox_by_execution_id", lambda dsn, **kwargs: []) class _Outbox: def __init__(self): self.trade = None def enqueue(self, trade: LedgerTrade): self.trade = trade def get(self, correlation_id): assert self.trade is not None return SimpleNamespace( status=SimpleNamespace(value="completed"), attempts=1, last_outcome=SimpleNamespace(value="completed"), last_error=None, ) outbox = _Outbox() class _Writer: def __init__(self, seen_outbox, *_args, **_kwargs): self.seen_outbox = seen_outbox def process_correlation_id(self, correlation_id): assert self.seen_outbox.trade is not None assert correlation_id == self.seen_outbox.trade.correlation_id return OutboxOutcome.COMPLETED import trex_core.adapters.ledger.portfolio_hub as hub_mod import trex_core.adapters.persistence.postgresql as pg_mod monkeypatch.setattr(pg_mod, "PostgreSQLLedgerOutbox", lambda dsn: outbox) monkeypatch.setattr(hub_mod, "PortfolioHubLedger", lambda *args, **kwargs: object()) monkeypatch.setattr(cli, "LedgerWriter", _Writer) rc = cli._cmd_ledger_backfill(_args(tmp_path), _cfg(), lambda _: True) assert rc == 0 out = json.loads(capsys.readouterr().out) assert out["result"]["status"] == "completed" assert outbox.trade is not None assert outbox.trade.order_venue == "SOR" assert outbox.trade.venue is None