from __future__ import annotations import json from datetime import UTC, datetime from types import SimpleNamespace from uuid import UUID from trex_core.application.ledger_skip import LedgerSkipRecord from trex_core.runtime import cli def _cfg(*, live: bool = False): return SimpleNamespace( dsn="postgresql://x/y", rabbitmq=None, broker="kis", kis=SimpleNamespace(allow_live=live), ) def _record(*, status="unresolved", resolution_reason=None): return LedgerSkipRecord( skip_key="ledger-skip:inst-1:slot-1:sell:O-1:E-1:missing_execution_id", instance_key="inst-1", slot_id=UUID("00000000-0000-0000-0000-000000000001"), portfolio_id="PF-1", symbol="005930", order_id="O-1", execution_id="E-1", side="sell", qty=10.0, price=70000.0, executed_at=datetime(2026, 8, 24, 0, 1, 2, tzinfo=UTC), order_venue="SOR", execution_venue="unknown", skip_reason="missing_execution_id", status=status, resolution_reason=resolution_reason, resolved_at=( datetime(2026, 8, 24, 0, 5, 0, tzinfo=UTC) if status == "resolved" else None ), created_at=datetime(2026, 8, 24, 0, 1, 2, tzinfo=UTC), updated_at=datetime(2026, 8, 24, 0, 5, 0, tzinfo=UTC), ) def test_ledger_skips_parser_accepts_list_and_resolve(): parser = cli._build_parser() args = parser.parse_args(["ledger", "skips", "list", "--status", "all"]) assert args.ledger_cmd == "skips" assert args.ledger_skips_cmd == "list" args = parser.parse_args(["ledger", "skips", "resolve", "--skip-key", "abc", "--yes"]) assert args.ledger_skips_cmd == "resolve" def test_ledger_skips_list_outputs_records(monkeypatch, capsys): class _Store: def __init__(self, *_args, **_kwargs): pass def list_records(self, **_kwargs): return [_record()] monkeypatch.setattr( "trex_core.adapters.persistence.ledger_skip_store.PostgreSQLLedgerSkipStore", _Store, ) args = SimpleNamespace( cmd="ledger", ledger_cmd="skips", ledger_skips_cmd="list", date="2026-08-24", portfolio_id="PF-1", status="unresolved", limit=10, instance="inst-1", config=None, config_path=None, env_file=None, json=True, timeout=10.0, ) rc = cli._cmd_ledger_skips_list(args, _cfg()) assert rc == 0 out = json.loads(capsys.readouterr().out) assert out["count"] == 1 assert out["records"][0]["skip_reason"] == "missing_execution_id" def test_ledger_skips_resolve_marks_record_and_emits_alert(monkeypatch, tmp_path, capsys): class _Store: def __init__(self, *_args, **_kwargs): pass def resolve_by_key(self, skip_key, *, instance_key, resolution_reason): assert skip_key == "abc" assert instance_key == "inst-1" return _record(status="resolved", resolution_reason=resolution_reason) published = [] class _Emitter: def publish_alarm(self, code, title, *, fields=None, severity="critical"): published.append((code, severity, fields)) class _Transport: def close(self): return None monkeypatch.setattr( "trex_core.adapters.persistence.ledger_skip_store.PostgreSQLLedgerSkipStore", _Store, ) monkeypatch.setattr(cli, "_default_alarm_emitter", lambda cfg: (_Emitter(), _Transport())) args = SimpleNamespace( cmd="ledger", ledger_cmd="skips", ledger_skips_cmd="resolve", skip_key="abc", reason="operator_reconciled", instance="inst-1", yes=True, live=False, audit_dir=str(tmp_path), config=None, config_path=None, env_file=None, json=True, timeout=10.0, ) rc = cli._cmd_ledger_skips_resolve(args, _cfg(), lambda _: True) assert rc == 0 out = json.loads(capsys.readouterr().out) assert out["result"] == "resolved" assert out["audit_path"].endswith(".json") assert published[0][0] == "ledger_skip_resolved" def test_ledger_skips_resolve_live_requires_live_flag(monkeypatch, tmp_path, capsys): monkeypatch.setattr( "trex_core.adapters.persistence.ledger_skip_store.PostgreSQLLedgerSkipStore", lambda *_args, **_kwargs: object(), ) args = SimpleNamespace( cmd="ledger", ledger_cmd="skips", ledger_skips_cmd="resolve", skip_key="abc", reason="operator_reconciled", instance="inst-1", yes=True, live=False, audit_dir=str(tmp_path), config=None, config_path=None, env_file=None, json=True, timeout=10.0, ) rc = cli._cmd_ledger_skips_resolve(args, _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["request"]["instance"] == "inst-1" assert audit["result"]["status"] == "rejected" assert audit["audit_path"].endswith(".json") def test_ledger_skips_resolve_live_requires_yes_and_live(monkeypatch, tmp_path, capsys): monkeypatch.setattr( "trex_core.adapters.persistence.ledger_skip_store.PostgreSQLLedgerSkipStore", lambda *_args, **_kwargs: object(), ) args = SimpleNamespace( cmd="ledger", ledger_cmd="skips", ledger_skips_cmd="resolve", skip_key="abc", reason="operator_reconciled", instance="inst-1", yes=False, live=True, audit_dir=str(tmp_path), config=None, config_path=None, env_file=None, json=True, timeout=10.0, ) rc = cli._cmd_ledger_skips_resolve(args, _cfg(live=True), lambda _: True) assert rc == 2 err = capsys.readouterr().err assert "--yes --live" in err audit = json.loads(next(tmp_path.glob("*.json")).read_text(encoding="utf-8")) assert audit["result"]["reason"] == "live account requires --yes --live" def test_ledger_skips_resolve_not_found_emits_audit_report(monkeypatch, tmp_path, capsys): class _Store: def __init__(self, *_args, **_kwargs): pass def resolve_by_key(self, *_args, **_kwargs): return None monkeypatch.setattr( "trex_core.adapters.persistence.ledger_skip_store.PostgreSQLLedgerSkipStore", _Store, ) args = SimpleNamespace( cmd="ledger", ledger_cmd="skips", ledger_skips_cmd="resolve", skip_key="missing", reason="operator_reconciled", instance="inst-1", yes=True, live=False, audit_dir=str(tmp_path), config=None, config_path=None, env_file=None, json=False, timeout=10.0, ) rc = cli._cmd_ledger_skips_resolve(args, _cfg(), lambda _: True) assert rc == 1 out = json.loads(capsys.readouterr().out) assert out["result"]["status"] == "not_found" assert out["audit_path"].endswith(".json") audit = json.loads(next(tmp_path.glob("*.json")).read_text(encoding="utf-8")) assert audit["audit_path"].endswith(".json")