trex-core/tests/application/test_sync.py

"""sync_slot_to_broker tests (REQ-105, BAT-051, dcy-034).

TDD coverage:
- AC-1: cold-start INACTIVE→HOLDING + qty = broker + warning cleared
- AC-2: HOLDING qty mismatch → qty updated + warning cleared
- AC-3: broker-only slot (no slot for pid) is not created — tested in CLI layer
- AC-4: WAC seed from cost_basis_port when available
- AC-5: WAC seed degrade gracefully when port absent/fails
- AC-6: dry-run shows before/after without writing
- AC-7: live double-ack gate — tested in CLI layer
- AC-8: qty still mismatched after sync → warning NOT cleared
- AC-9: --all coverage — tested in CLI layer
- AC-10: INACTIVE→HOLDING transition actor=sync, reason=onboarding logged
- OQ-A: transition INACTIVE→HOLDING allowed in state machine
- NFR-3: PENDING_* slot skipped (no auto-fix on in-flight orders)
"""
from __future__ import annotations

import logging
from uuid import uuid4

import pytest

from trex_core.application.broker_port import BalanceItem
from trex_core.application.portfolio import InMemoryReconcileWarningStore
from trex_core.application.reconciliation import sync_slot_to_broker
from trex_core.domain.slot import SlotService, SlotState
from trex_core.domain.slot.in_memory_repository import InMemorySlotRepository
from trex_core.domain.slot.models import Slot


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------

INSTANCE = "test-instance"


def _make_slot(status: SlotState, qty: float | None = None, symbol: str = "005930",
               pid: str = "pf-005930", avg_price: float | None = None) -> Slot:
    return Slot(
        slot_id=uuid4(),
        status=status,
        symbol=symbol,
        qty=qty,
        avg_price=avg_price,
        portfolio_id=pid,
        instance_key=INSTANCE,
    )


class FakeBroker:
    def __init__(self, balances: dict[str, int]):
        self._b = balances

    def name(self): return "fake"
    def submit_order(self, req): ...
    def poll_fill(self, *args, **kw): ...
    def cancel_order(self, *args, **kw): ...
    def get_current_price(self, sym): return None

    def get_balance(self) -> list[BalanceItem]:
        return [BalanceItem(symbol=sym, hldg_qty=qty)
                for sym, qty in self._b.items() if qty > 0]


class FakeCostBasisPort:
    def __init__(self, values: dict[str, float | None]):
        self._v = values

    def get_cost_basis_with_diagnostics(self, portfolio_id: str):
        value = self._v.get(portfolio_id)
        if value is None:
            return None, "none", "missing"
        return value, "legacy", None

    # Backward compatibility for non-diagnostic callers
    def get_cost_basis(self, portfolio_id: str) -> float | None:
        value = self._v.get(portfolio_id)
        return value


class RaisingCostBasisPort:
    def get_cost_basis_with_diagnostics(self, portfolio_id: str):
        raise RuntimeError("simulated network error")


class DiagnosticCostBasisPort:
    def __init__(self, result):
        self._result = result

    def get_cost_basis_with_diagnostics(self, portfolio_id: str):
        return self._result


class LegacyOnlyCostBasisPort:
    """Compatibility seam with no diagnostic method."""

    def __init__(self, value):
        self._value = value

    def get_cost_basis(self, portfolio_id: str):
        return self._value


def _setup(slot: Slot, broker_balances: dict[str, int],
           cost_basis_port=None, dry_run=False,
           warn_store=None):
    repo = InMemorySlotRepository()
    repo.add_slot(slot)
    svc = SlotService(repo, INSTANCE)
    ws = warn_store if warn_store is not None else InMemoryReconcileWarningStore()
    broker = FakeBroker(broker_balances)
    result = sync_slot_to_broker(
        slot_id=str(slot.slot_id),
        slot_repo=repo,
        slot_service=svc,
        broker=broker,
        instance_key=INSTANCE,
        warn_store=ws,
        cost_basis_port=cost_basis_port,
        dry_run=dry_run,
    )
    return result, repo, ws


# ---------------------------------------------------------------------------
# AC-1: cold-start INACTIVE + broker qty > 0 → HOLDING, qty=broker, warning cleared
# ---------------------------------------------------------------------------

def test_sync_inactive_to_holding_on_cold_start():
    """AC-1: INACTIVE slot with broker qty + seeded cost-basis transitions to HOLDING."""
    slot = _make_slot(SlotState.INACTIVE, qty=None)
    ws = InMemoryReconcileWarningStore()
    ws.set_warned(str(slot.slot_id), portfolio_id="pf-005930", case_code="B")
    cb_port = FakeCostBasisPort({"pf-005930": 50000.0})

    result, repo, ws_out = _setup(slot, {"005930": 10}, cost_basis_port=cb_port, warn_store=ws)

    refreshed = repo.fetch_slot(slot.slot_id)
    assert refreshed.status is SlotState.HOLDING
    assert refreshed.qty == 10.0
    assert refreshed.avg_price == 50000.0
    assert result["warning_cleared"] is True
    assert not ws_out.is_warned(str(slot.slot_id))
    assert result["skipped"] is False
    assert result["seeded_avg_price"] is True
    assert result["cost_basis_source"] == "legacy"
    assert result["before"]["status"] == "inactive"
    assert result["after"]["status"] == "holding"
    assert result["after"]["qty"] == 10.0


def test_sync_inactive_to_holding_sets_transition_log():
    """AC-10: INACTIVE→HOLDING transition_log has actor=sync, reason=onboarding."""
    slot = _make_slot(SlotState.INACTIVE, qty=None)
    repo = InMemorySlotRepository()
    repo.add_slot(slot)
    svc = SlotService(repo, INSTANCE)
    ws = InMemoryReconcileWarningStore()
    broker = FakeBroker({"005930": 5})
    cb_port = FakeCostBasisPort({"pf-005930": 50000.0})

    sync_slot_to_broker(
        slot_id=str(slot.slot_id),
        slot_repo=repo,
        slot_service=svc,
        broker=broker,
        instance_key=INSTANCE,
        warn_store=ws,
        cost_basis_port=cb_port,
    )

    transitions = repo._transitions
    assert len(transitions) == 1
    t = transitions[0]
    assert t.from_state is SlotState.INACTIVE
    assert t.to_state is SlotState.HOLDING
    assert t.actor == "sync"
    assert "onboarding" in t.reason


# ---------------------------------------------------------------------------
# AC-2: HOLDING qty mismatch → qty updated, warning cleared
# ---------------------------------------------------------------------------

def test_sync_holding_qty_mismatch_updated():
    """AC-2: HOLDING slot with qty != broker → qty updated, warning cleared."""
    slot = _make_slot(SlotState.HOLDING, qty=5.0)
    ws = InMemoryReconcileWarningStore()
    ws.set_warned(str(slot.slot_id), portfolio_id="pf-005930", case_code="B")

    result, repo, ws_out = _setup(slot, {"005930": 10}, warn_store=ws)

    refreshed = repo.fetch_slot(slot.slot_id)
    assert refreshed.qty == 10.0
    assert refreshed.status is SlotState.HOLDING
    assert result["warning_cleared"] is True
    assert not ws_out.is_warned(str(slot.slot_id))


# ---------------------------------------------------------------------------
# AC-4: WAC seed from cost_basis_port when available
# ---------------------------------------------------------------------------

def test_sync_seeds_avg_price_from_cost_basis_port():
    """AC-4: When cost_basis_port returns WAC, avg_price is seeded."""
    slot = _make_slot(SlotState.INACTIVE, qty=None, pid="pf-005930")
    cb_port = FakeCostBasisPort({"pf-005930": 75000.0})

    result, repo, _ = _setup(slot, {"005930": 10}, cost_basis_port=cb_port)

    refreshed = repo.fetch_slot(slot.slot_id)
    assert refreshed.avg_price == 75000.0
    assert result.get("seeded_avg_price") is True


def test_sync_seeds_avg_price_holding_with_cost_basis():
    """AC-4: HOLDING slot also gets avg_price seeded from cost_basis_port."""
    slot = _make_slot(SlotState.HOLDING, qty=5.0, avg_price=60000.0, pid="pf-005930")
    cb_port = FakeCostBasisPort({"pf-005930": 70000.0})

    result, repo, _ = _setup(slot, {"005930": 10}, cost_basis_port=cb_port)

    refreshed = repo.fetch_slot(slot.slot_id)
    assert refreshed.avg_price == 70000.0


# ---------------------------------------------------------------------------
# AC-5: graceful degrade — no cost_basis_port or port fails
# ---------------------------------------------------------------------------

def test_sync_inactive_requires_cost_basis_live():
    """AC-5: INACTIVE live sync requires avg_price seed; missing cost_basis_port fails closed."""
    slot = _make_slot(SlotState.INACTIVE, qty=None, pid="pf-005930")

    result, repo, _ = _setup(slot, {"005930": 10}, cost_basis_port=None)

    refreshed = repo.fetch_slot(slot.slot_id)
    assert refreshed.status is SlotState.INACTIVE
    assert refreshed.qty is None
    assert refreshed.avg_price is None
    assert result["skipped"] is True
    assert result["error"] == "cost_basis_unavailable"
    assert result["dry_run"] is False
    assert result["skip_reason"] == "cost_basis_unavailable"
    assert result.get("seeded_avg_price") is False


def test_sync_inactive_cost_basis_port_raises_fail_closed(caplog):
    """AC-5: INACTIVE live sync fails closed if cost_basis_port diagnostics raises."""
    slot = _make_slot(SlotState.INACTIVE, qty=None, pid="pf-005930")
    cb_port = RaisingCostBasisPort()

    with caplog.at_level(logging.WARNING, logger="trex_core.application.reconciliation"):
        result, repo, _ = _setup(slot, {"005930": 10}, cost_basis_port=cb_port)

    refreshed = repo.fetch_slot(slot.slot_id)
    assert refreshed.status is SlotState.INACTIVE
    assert refreshed.qty is None
    assert result["skipped"] is True
    assert result["error"] == "cost_basis_unavailable"
    assert "cost_basis_unavailable" in result.get("skip_reason", "")
    assert "graceful degrade" in caplog.text or "cost_basis_port" in caplog.text


def test_sync_degrades_gracefully_cost_basis_returns_none():
    """AC-5: When cost_basis_port returns None, qty is updated, avg_price unchanged."""
    slot = _make_slot(SlotState.HOLDING, qty=5.0, avg_price=60000.0, pid="pf-005930")
    cb_port = FakeCostBasisPort({"pf-005930": None})

    result, repo, _ = _setup(slot, {"005930": 10}, cost_basis_port=cb_port)

    refreshed = repo.fetch_slot(slot.slot_id)
    assert refreshed.qty == 10.0
    assert refreshed.avg_price == 60000.0  # unchanged


@pytest.mark.parametrize("invalid_wac", (0.0, -1.0, float("nan"), float("inf")))
@pytest.mark.parametrize("dry_run", (True, False))
def test_sync_legacy_only_invalid_wac_fails_closed_without_writes(invalid_wac, dry_run):
    """QC-6: legacy-only invalid WAC never bypasses inactive fail-closed barrier."""
    slot = _make_slot(SlotState.INACTIVE, qty=None, pid="pf-005930")
    ws = InMemoryReconcileWarningStore()
    ws.set_warned(str(slot.slot_id), portfolio_id="pf-005930", case_code="B")

    result, repo, ws_out = _setup(
        slot,
        {"005930": 10},
        cost_basis_port=LegacyOnlyCostBasisPort(invalid_wac),
        dry_run=dry_run,
        warn_store=ws,
    )

    refreshed = repo.fetch_slot(slot.slot_id)
    assert refreshed is not None
    assert refreshed.status is SlotState.INACTIVE
    assert refreshed.qty is None
    assert refreshed.avg_price is None
    assert ws_out.is_warned(str(slot.slot_id))
    assert result["after"] == result["before"]
    assert result["warning_cleared"] is False
    assert result["seeded_avg_price"] is False
    assert result["cost_basis_source"] == "none"
    assert result["cost_basis_reason"] == "invalid"
    assert result["skipped"] is True
    assert result["skip_reason"] == "cost_basis_unavailable"
    if dry_run:
        assert result["dry_run"] is True
        assert "error" not in result
    else:
        assert result["dry_run"] is False
        assert result["error"] == "cost_basis_unavailable"


def test_sync_normalizes_diagnostic_source_aliases_and_rejects_unknown_sources():
    """REQ-123: diagnostic seam never exposes raw provider field names."""
    for raw_source in ("avg_cost", "wac", "cost_basis"):
        slot = _make_slot(SlotState.INACTIVE, qty=None)
        result, repo, _ = _setup(
            slot,
            {"005930": 10},
            cost_basis_port=DiagnosticCostBasisPort((50000.0, raw_source, None)),
        )
        refreshed = repo.fetch_slot(slot.slot_id)
        assert refreshed is not None
        assert refreshed.status is SlotState.HOLDING
        assert result["cost_basis_source"] == "legacy"
        assert result["cost_basis_reason"] is None

    for raw_source in ("provider_private", 7, None):
        slot = _make_slot(SlotState.INACTIVE, qty=None)
        result, repo, _ = _setup(
            slot,
            {"005930": 10},
            cost_basis_port=DiagnosticCostBasisPort((50000.0, raw_source, "raw.detail")),
        )
        refreshed = repo.fetch_slot(slot.slot_id)
        assert refreshed is not None
        assert refreshed.status is SlotState.INACTIVE
        assert result["cost_basis_source"] == "none"
        assert result["cost_basis_reason"] == "diagnostic_source_invalid"


def test_sync_preserves_only_sanitized_adapter_failure_reasons():
    """REQ-123: public result reasons are stable and do not expose parser detail."""
    cases = (
        (None, "canonical", "invalid", "invalid"),
        (None, "legacy", "missing", "missing"),
        (None, "none", "transport", "transport"),
        (None, "none", "http_status_503", "http_status_503"),
        (None, "none", "invalid_json", "invalid_json"),
        (None, "none", "invalid_payload", "invalid_payload"),
    )
    for value, source, reason, expected_reason in cases:
        slot = _make_slot(SlotState.INACTIVE, qty=None)
        result, repo, _ = _setup(
            slot,
            {"005930": 10},
            cost_basis_port=DiagnosticCostBasisPort((value, source, reason)),
            dry_run=True,
        )
        refreshed = repo.fetch_slot(slot.slot_id)
        assert refreshed is not None
        assert refreshed.status is SlotState.INACTIVE
        assert result["cost_basis_source"] in {"canonical", "legacy", "none"}
        assert result["cost_basis_reason"] == expected_reason


# ---------------------------------------------------------------------------
# AC-6: dry-run — no writes, shows before/after
# ---------------------------------------------------------------------------

def test_sync_dry_run_inactive_valid_canonical_seed_previews_without_writes():
    """QC-5: canonical seed dry-run previews holding state without mutations."""
    slot = _make_slot(SlotState.INACTIVE, qty=None)
    ws = InMemoryReconcileWarningStore()
    ws.set_warned(str(slot.slot_id), portfolio_id="pf-005930", case_code="B")
    cb_port = DiagnosticCostBasisPort((75000.0, "canonical", None))

    result, repo, ws_out = _setup(
        slot,
        {"005930": 5},
        cost_basis_port=cb_port,
        dry_run=True,
        warn_store=ws,
    )

    refreshed = repo.fetch_slot(slot.slot_id)
    assert refreshed is not None
    assert refreshed.status is SlotState.INACTIVE
    assert refreshed.qty is None
    assert refreshed.avg_price is None
    assert ws_out.is_warned(str(slot.slot_id))
    assert result["dry_run"] is True
    assert result["warning_cleared"] is False
    assert result["after"]["status"] == "holding"
    assert result["after"]["qty"] == 5.0
    assert result["after"]["avg_price"] == 75000.0
    assert result["seeded_avg_price"] is True
    assert result["cost_basis_source"] == "canonical"
    assert result["cost_basis_reason"] is None


def test_sync_dry_run_no_write():
    """AC-6: dry-run does NOT modify the slot, returns before/after preview."""
    slot = _make_slot(SlotState.INACTIVE, qty=None)
    ws = InMemoryReconcileWarningStore()
    ws.set_warned(str(slot.slot_id), portfolio_id="pf-005930", case_code="B")

    result, repo, ws_out = _setup(slot, {"005930": 10}, dry_run=True, warn_store=ws)

    refreshed = repo.fetch_slot(slot.slot_id)
    # slot must NOT be modified
    assert refreshed.status is SlotState.INACTIVE
    assert refreshed.qty is None
    # warning must NOT be cleared
    assert ws_out.is_warned(str(slot.slot_id))
    # result has dry_run flag and no-write preview
    assert result["dry_run"] is True
    assert result["after"]["status"] == "inactive"
    assert result["after"]["qty"] is None
    assert result["warning_cleared"] is False
    assert result["skip_reason"] == "cost_basis_unavailable"


# ---------------------------------------------------------------------------
# AC-8: qty still mismatched after sync → warning NOT cleared
# ---------------------------------------------------------------------------

def test_sync_warning_not_cleared_when_qty_still_mismatched():
    """AC-8: If qty update somehow left a mismatch, warning is not cleared."""
    # Simulate: slot.qty stays at original value after "update" — we'll use
    # a direct repo scenario where update_slot is patched to do nothing.
    # Instead, test via direct manipulation: a HOLDING slot with broker qty=10,
    # but the update fails to persist (skip by using a read-only scenario).
    # Real path: after sync the refreshed qty WILL match broker, so we need
    # to test the 'if after_qty != broker_qty' branch via a different scenario.
    # We'll test this via the "broker returns 0" (broker_qty_zero skip) path.
    slot = _make_slot(SlotState.HOLDING, qty=5.0)
    ws = InMemoryReconcileWarningStore()
    ws.set_warned(str(slot.slot_id), portfolio_id="pf-005930", case_code="B")

    # broker_qty=0 → should skip (A-case guard) and not clear warning
    result, repo, ws_out = _setup(slot, {"005930": 0}, warn_store=ws)

    assert result["skipped"] is True
    assert result["skip_reason"] == "broker_qty_zero"
    assert ws_out.is_warned(str(slot.slot_id))  # warning NOT cleared


# ---------------------------------------------------------------------------
# NFR-3: PENDING_* slot is skipped (no auto-fix on in-flight orders)
# ---------------------------------------------------------------------------

def test_sync_pending_buy_slot_skipped():
    """NFR-3: PENDING_BUY slot is skipped (in-flight, do not touch)."""
    slot = _make_slot(SlotState.PENDING_BUY, qty=None)
    result, repo, _ = _setup(slot, {"005930": 10})

    refreshed = repo.fetch_slot(slot.slot_id)
    assert refreshed.status is SlotState.PENDING_BUY  # unchanged
    assert result["skipped"] is True
    assert "in_flight" in result.get("skip_reason", "")


def test_sync_pending_sell_slot_skipped():
    """NFR-3: PENDING_SELL slot is skipped."""
    slot = _make_slot(SlotState.PENDING_SELL, qty=10.0)
    result, repo, _ = _setup(slot, {"005930": 10})

    refreshed = repo.fetch_slot(slot.slot_id)
    assert refreshed.status is SlotState.PENDING_SELL
    assert result["skipped"] is True


# ---------------------------------------------------------------------------
# Broker-only (no slot) — not handled by sync_slot_to_broker
# (that path is filtered in CLI layer, but we verify slot_not_found)
# ---------------------------------------------------------------------------

def test_sync_invalid_slot_id_returns_error():
    """Invalid slot_id returns error result."""
    repo = InMemorySlotRepository()
    svc = SlotService(repo, INSTANCE)
    ws = InMemoryReconcileWarningStore()
    broker = FakeBroker({"005930": 10})

    result = sync_slot_to_broker(
        slot_id="not-a-uuid",
        slot_repo=repo,
        slot_service=svc,
        broker=broker,
        instance_key=INSTANCE,
        warn_store=ws,
    )
    assert result["skipped"] is True
    assert result["error"] == "invalid_slot_id"


def test_sync_missing_slot_returns_not_found():
    """Slot ID is valid UUID but not in repo → slot_not_found."""
    repo = InMemorySlotRepository()
    svc = SlotService(repo, INSTANCE)
    ws = InMemoryReconcileWarningStore()
    broker = FakeBroker({"005930": 10})

    result = sync_slot_to_broker(
        slot_id=str(uuid4()),
        slot_repo=repo,
        slot_service=svc,
        broker=broker,
        instance_key=INSTANCE,
        warn_store=ws,
    )
    assert result["skipped"] is True
    assert result["error"] == "slot_not_found"


# ---------------------------------------------------------------------------
# Unwarned slot sync (no warning in store) — should still sync qty and return
# ---------------------------------------------------------------------------

def test_sync_unwarned_slot_still_syncs_qty():
    """sync_slot_to_broker works on slots without an active warning,
    but INACTIVE still requires avg_price for live write."""
    slot = _make_slot(SlotState.INACTIVE, qty=None, pid="pf-005930")
    ws = InMemoryReconcileWarningStore()  # no warning set
    cb_port = FakeCostBasisPort({"pf-005930": 50000.0})

    result, repo, _ = _setup(slot, {"005930": 7}, cost_basis_port=cb_port, warn_store=ws)

    refreshed = repo.fetch_slot(slot.slot_id)
    assert refreshed.status is SlotState.HOLDING
    assert refreshed.qty == 7.0
    assert refreshed.avg_price == 50000.0
    # warning_cleared is True because clear_warned() was called successfully
    # (idempotent no-op when there was no warning)
    assert result["warning_cleared"] is True


# ---------------------------------------------------------------------------
# OQ-A actor restriction: general strategy path still goes INACTIVE→PENDING_BUY
# ---------------------------------------------------------------------------

def test_general_strategy_path_still_uses_pending_buy():
    """OQ-A: normal strategy entry path (INACTIVE→PENDING_BUY) is unaffected."""
    from trex_core.domain.slot.state import SlotState, is_allowed_transition
    assert is_allowed_transition(SlotState.INACTIVE, SlotState.PENDING_BUY)
    assert is_allowed_transition(SlotState.PENDING_BUY, SlotState.HOLDING)
    # sync shortcut also allowed
    assert is_allowed_transition(SlotState.INACTIVE, SlotState.HOLDING)