trex-core/tests/application/test_messaging_intent_consumer.py

"""Intent admission RPC handler + dedup tests (REQ-021/022/023, dcy-006).

The handler calls IntentService.submit_*, so dedup uses the IntentStore
(intent_log) gate (REQ-007). Two messages with the same correlation_id
trigger exactly ONE slot transition (REQ-023 AC-1).
"""
from __future__ import annotations

from trex_core.adapters.persistence.in_memory_intent_store import InMemoryIntentStore
from trex_core.application.intents import IntentService
from trex_core.application.messaging import RpcRequest
from trex_core.application.messaging_handlers import make_intent_rpc_handler
from trex_core.domain.slot import SlotService, SlotState
from trex_core.domain.slot.in_memory_repository import InMemorySlotRepository



def _wire(max_active=3):
    repo = InMemorySlotRepository()
    svc = SlotService(repo)
    # dcy-011 / REQ-007: buy resolves symbol from the universe via resolver.
    intents = IntentService(svc, repo, InMemoryIntentStore(),
                            resolve_symbol=lambda pid: "005930",
                            max_active_slots=max_active)
    repo.reconcile_universe("default", {"PF-1": "SYM", "PF-2": "SYM", "PF-3": "SYM", "PF-9": "SYM"})
    return repo, svc, intents


def _buy_req(intent_id: str, portfolio_id: str = "PF-1") -> RpcRequest:
    # dcy-011 / REQ-021: buy payload carries portfolio_id (required); no symbol.
    return RpcRequest(
        payload={
            "intent_type": "buy",
            "intent_id": intent_id,
            "portfolio_id": portfolio_id,
            "qty": 10,
        },
        correlation_id=intent_id,
    )


def test_buy_admission_returns_admitted_and_transitions_slot():
    repo, _, intents = _wire()
    handler = make_intent_rpc_handler(intents)
    resp = handler(_buy_req("buy-1"))
    assert resp.payload["status"] == "admitted"
    assert resp.payload["slot_id"]
    from uuid import UUID
    slot = repo.fetch_slot(UUID(resp.payload["slot_id"]))
    assert slot.status is SlotState.PENDING_BUY and slot.symbol == "005930"


def test_duplicate_correlation_id_dedups_via_intent_log():
    """REQ-023 AC-1: same intent_id ⇒ slot transition exactly once."""
    repo, _, intents = _wire()
    handler = make_intent_rpc_handler(intents)
    r1 = handler(_buy_req("buy-dup"))
    r2 = handler(_buy_req("buy-dup"))
    assert r1.payload == r2.payload
    pending = [s for s in repo.list_slots("default") if s.status is SlotState.PENDING_BUY]
    assert len(pending) == 1


def test_buy_rejected_when_capacity_full():
    repo, _, intents = _wire(max_active=3)
    handler = make_intent_rpc_handler(intents)
    for i, pid in enumerate(("PF-1", "PF-2", "PF-3")):
        assert handler(_buy_req(f"buy-{i}", portfolio_id=pid)).payload["status"] == "admitted"
    resp = handler(_buy_req("overflow", portfolio_id="PF-9"))
    assert resp.payload["status"] == "rejected"
    assert resp.payload["reason"] == "capacity_full"


def test_sell_admission_from_holding():
    repo, svc, intents = _wire()
    handler = make_intent_rpc_handler(intents)
    sid = handler(_buy_req("buy-h")).payload["slot_id"]
    from uuid import UUID
    svc.transit_slot(UUID(sid), SlotState.PENDING_BUY, SlotState.HOLDING,
                     reason="filled", actor="t")
    resp = handler(RpcRequest(
        payload={"intent_type": "sell", "intent_id": "sell-h", "slot_id": sid},
        correlation_id="sell-h",
    ))
    assert resp.payload["status"] == "admitted"
    assert repo.fetch_slot(UUID(sid)).status is SlotState.PENDING_SELL


def test_bad_payload_returns_rejected_not_exception():
    """Handler must not raise — bad input ⇒ structured rejection."""
    _, _, intents = _wire()
    handler = make_intent_rpc_handler(intents)
    # missing slot_id for sell
    resp = handler(RpcRequest(
        payload={"intent_type": "sell", "intent_id": "bad"},
        correlation_id="bad",
    ))
    assert resp.payload["status"] == "rejected"
    assert resp.payload["reason"].startswith("bad_request:")


def test_unknown_intent_type_rejected():
    _, _, intents = _wire()
    handler = make_intent_rpc_handler(intents)
    resp = handler(RpcRequest(
        payload={"intent_type": "ping", "intent_id": "x"},
        correlation_id="x",
    ))
    assert resp.payload["status"] == "rejected"
    assert "unknown_intent_type" in resp.payload["reason"]


def test_correlation_id_is_used_as_intent_id_when_missing_in_payload():
    """The transport's correlation_id is the canonical idempotency key."""
    repo, _, intents = _wire()
    handler = make_intent_rpc_handler(intents)
    req = RpcRequest(
        payload={
            "intent_type": "buy",
            "portfolio_id": "PF-1",
            "qty": 5,
        },
        correlation_id="cid-only",
    )
    r1 = handler(req)
    r2 = handler(req)
    assert r1.payload["status"] == "admitted"
    assert r1.payload == r2.payload
    pending = [s for s in repo.list_slots("default") if s.status is SlotState.PENDING_BUY]
    assert len(pending) == 1


# ---- dcy-011 / REQ-007/021: portfolio_id required + symbol resolve ----------

def test_buy_without_key_rejected():
    """dcy-020/REQ-061: missing slot_id AND portfolio_id → reject."""
    repo, _, intents = _wire()
    handler = make_intent_rpc_handler(intents)
    req = RpcRequest(
        payload={"intent_type": "buy", "intent_id": "no-pf", "qty": 5},
        correlation_id="no-pf",
    )
    resp = handler(req)
    assert resp.payload["status"] == "rejected"
    assert resp.payload["reason"] == "slot_id_or_portfolio_required"


def test_buy_unknown_portfolio_rejected():
    """FR-3 / S-1: portfolio_id with no asset slot → unknown_portfolio reject."""
    repo, _, intents = _wire()
    handler = make_intent_rpc_handler(intents)
    resp = handler(_buy_req("u1", portfolio_id="PF-X"))
    assert resp.payload["status"] == "rejected"
    assert resp.payload["reason"] == "unknown_portfolio"


def test_buy_symbol_resolved_from_universe():
    """FR-2 / S-1: symbol is resolved server-side from portfolio_id (no client symbol)."""
    repo, _, intents = _wire()  # resolver → "005930"
    handler = make_intent_rpc_handler(intents)
    resp = handler(_buy_req("r1", portfolio_id="PF-9"))
    assert resp.payload["status"] == "admitted"
    from uuid import UUID
    slot = repo.fetch_slot(UUID(resp.payload["slot_id"]))
    assert slot.symbol == "005930" and slot.portfolio_id == "PF-9"


def test_buy_payload_allocated_capital_is_ignored():
    """dcy-023 / REQ-007 AC-6: allocated_capital is no longer a contract field.
    Senders (e.g. trex-orb-v1) may still include it — admission proceeds
    normally and the value lands nowhere (no Slot/BuyIntent field exists)."""
    repo, _, intents = _wire()
    handler = make_intent_rpc_handler(intents)
    req = RpcRequest(
        payload={
            "intent_type": "buy",
            "intent_id": "buy-cap",
            "portfolio_id": "PF-1",
            "qty": 10,
            "allocated_capital": 1_000_000,   # unknown key — ignored
        },
        correlation_id="buy-cap",
    )
    resp = handler(req)
    assert resp.payload["status"] == "admitted"
    from uuid import UUID
    slot = repo.fetch_slot(UUID(resp.payload["slot_id"]))
    assert slot.status is SlotState.PENDING_BUY and slot.qty == 10
    assert not hasattr(slot, "allocated_capital")