"""SlotService unit tests — transitions, inactive reset, capacity (asset-centric). Originally ported from trex-orbit (dcy-001 FR-3); rewritten at the dcy-020 cutover (BAT-030): permanent asset slots, 5-state machine, instance-scoped capacity. Covers REQ-002 + REQ-057/REQ-059. """ from __future__ import annotations import threading import time from uuid import uuid4 import pytest from trex_core.domain.slot import ( SlotNotFoundError, SlotService, SlotState, SlotStateMismatchError, SlotTransitionError, ) from trex_core.domain.slot.in_memory_repository import InMemorySlotRepository INST = "default" @pytest.fixture def repo(): return InMemorySlotRepository() @pytest.fixture def service(repo): return SlotService(repo, instance_key=INST) def _seed(repo, n=1): """Create n permanent asset slots (inactive) via universe reconcile.""" repo.reconcile_universe(INST, {f"pf-{i}": "SYM" for i in range(n)}) return sorted(repo.list_slots(INST), key=lambda s: s.portfolio_id) # ---------------------------------------------------------------------------- # transit_slot — happy paths # ---------------------------------------------------------------------------- class TestTransitHappyPath: def test_inactive_to_pending_buy_records_fields(self, service, repo): slot = _seed(repo)[0] updated = service.transit_slot( slot.slot_id, SlotState.INACTIVE, SlotState.PENDING_BUY, reason="orb_breakout", actor="buy_decision_thread", symbol="005930", qty=10, strategy_id="orb_v1", ) assert updated.status is SlotState.PENDING_BUY assert updated.symbol == "005930" assert updated.qty == 10 assert updated.strategy_id == "orb_v1" assert updated.portfolio_id == slot.portfolio_id # asset binding kept # log appended log = repo.transition_log_for(slot.slot_id) assert len(log) == 1 assert log[0].from_state is SlotState.INACTIVE assert log[0].to_state is SlotState.PENDING_BUY assert log[0].reason == "orb_breakout" assert log[0].actor == "buy_decision_thread" def test_pending_buy_to_holding(self, service, repo): slot = _seed(repo)[0] service.transit_slot( slot.slot_id, SlotState.INACTIVE, SlotState.PENDING_BUY, reason="orb_breakout", actor="buy", symbol="005930", ) service.transit_slot( slot.slot_id, SlotState.PENDING_BUY, SlotState.HOLDING, reason="filled", actor="order_exec", avg_price=70_000, ) cur = repo.fetch_slot(slot.slot_id) assert cur.status is SlotState.HOLDING assert cur.avg_price == 70_000 # dcy-012: limit_price (pending order limit) is cleared on entering holding. assert cur.limit_price is None def test_buy_rollback_resets_occupancy_keeps_asset_identity(self, service, repo): """pending_buy -> inactive (rejection rollback): position/order fields are cleared while the permanent slot keeps its asset binding.""" slot = _seed(repo)[0] service.transit_slot( slot.slot_id, SlotState.INACTIVE, SlotState.PENDING_BUY, reason="orb_breakout", actor="buy", symbol="005930", qty=10, strategy_id="orb_v1", ) rolled = service.transit_slot( slot.slot_id, SlotState.PENDING_BUY, SlotState.INACTIVE, reason="broker_reject:001", actor="order_exec", ) assert rolled.status is SlotState.INACTIVE assert rolled.qty is None assert rolled.strategy_id is None assert rolled.limit_price is None assert rolled.broker_order_id is None # asset identity preserved (dcy-020/REQ-056 permanent slot) assert rolled.portfolio_id == slot.portfolio_id assert rolled.instance_key == INST def test_sell_fill_returns_to_inactive_and_clears_position(self, service, repo): slot = _seed(repo)[0] service.transit_slot(slot.slot_id, SlotState.INACTIVE, SlotState.PENDING_BUY, reason="r", actor="a", symbol="005930", qty=10) service.transit_slot(slot.slot_id, SlotState.PENDING_BUY, SlotState.HOLDING, reason="r", actor="a", avg_price=70_000) service.transit_slot(slot.slot_id, SlotState.HOLDING, SlotState.PENDING_SELL, reason="r", actor="a") done = service.transit_slot(slot.slot_id, SlotState.PENDING_SELL, SlotState.INACTIVE, reason="sell_filled", actor="a") assert done.status is SlotState.INACTIVE assert done.qty is None and done.avg_price is None assert done.portfolio_id == slot.portfolio_id # the same asset slot can activate again (permanent reuse) again = service.transit_slot( slot.slot_id, SlotState.INACTIVE, SlotState.PENDING_BUY, reason="re-entry", actor="a", symbol="005930", ) assert again.status is SlotState.PENDING_BUY # ---------------------------------------------------------------------------- # transit_slot — rejections # ---------------------------------------------------------------------------- class TestTransitRejections: def test_disallowed_transition_raises(self, service, repo): """INACTIVE→PENDING_SELL is not an allowed transition (skips intermediate states). Note: INACTIVE→HOLDING is now allowed (BAT-051/REQ-105 OQ-A sync path), so we use INACTIVE→PENDING_SELL as the disallowed example instead. """ slot = _seed(repo)[0] with pytest.raises(SlotTransitionError): service.transit_slot( slot.slot_id, SlotState.INACTIVE, SlotState.PENDING_SELL, reason="x", actor="a", ) def test_state_mismatch_raises_when_current_state_drifted(self, service, repo): """Race detection — if slot already advanced, transit_slot from old state fails.""" slot = _seed(repo)[0] service.transit_slot( slot.slot_id, SlotState.INACTIVE, SlotState.PENDING_BUY, reason="r", actor="a", ) with pytest.raises(SlotStateMismatchError): # caller still thinks status=inactive but it's pending_buy service.transit_slot( slot.slot_id, SlotState.INACTIVE, SlotState.PENDING_BUY, reason="r", actor="a", ) def test_unknown_slot_raises(self, service): with pytest.raises(SlotNotFoundError): service.transit_slot( uuid4(), SlotState.INACTIVE, SlotState.PENDING_BUY, reason="r", actor="a", ) # ---------------------------------------------------------------------------- # operator error → inactive/holding (REQ-004) # ---------------------------------------------------------------------------- class TestOperatorResolve: def test_error_to_inactive_via_operator(self, service, repo): slot = _seed(repo)[0] # force into error service.transit_slot( slot.slot_id, SlotState.INACTIVE, SlotState.ERROR, reason="reconcile_mismatch", actor="system", ) result = service.transit_slot( slot.slot_id, SlotState.ERROR, SlotState.INACTIVE, reason="broker confirmed no position", actor="operator:alice", ) assert result.status is SlotState.INACTIVE log = repo.transition_log_for(slot.slot_id) assert log[-1].actor == "operator:alice" assert "broker confirmed" in log[-1].reason def test_error_to_pending_buy_rejected(self, service, repo): slot = _seed(repo)[0] service.transit_slot( slot.slot_id, SlotState.INACTIVE, SlotState.ERROR, reason="r", actor="s", ) with pytest.raises(SlotTransitionError): service.transit_slot( slot.slot_id, SlotState.ERROR, SlotState.PENDING_BUY, reason="r", actor="operator:alice", ) # ---------------------------------------------------------------------------- # compute_slot_capacity (REQ-002 AC-6; dcy-020/REQ-059) # ---------------------------------------------------------------------------- class TestCapacity: def test_all_inactive_no_cap(self, service, repo): _seed(repo, 10) cap = service.compute_slot_capacity() assert cap.inactive_count == 10 assert cap.holding_count == 0 assert cap.pending_count == 0 assert cap.active_count == 0 assert cap.available_count == 10 assert cap.total_slots == 10 assert cap.scarcity_ratio == 0.0 def test_mixed_states_under_m_cap(self, service, repo): slots = _seed(repo, 10) # move 3 to pending_buy + 2 to holding for s in slots[:3]: service.transit_slot( s.slot_id, SlotState.INACTIVE, SlotState.PENDING_BUY, reason="r", actor="a", ) for s in slots[3:5]: service.transit_slot( s.slot_id, SlotState.INACTIVE, SlotState.PENDING_BUY, reason="r", actor="a", ) service.transit_slot( s.slot_id, SlotState.PENDING_BUY, SlotState.HOLDING, reason="r", actor="a", ) cap = service.compute_slot_capacity(max_active_slots=8) assert cap.inactive_count == 5 assert cap.pending_count == 3 assert cap.holding_count == 2 assert cap.active_count == 5 assert cap.available_count == 3 # M(8) − active(5) assert cap.total_slots == 8 assert cap.scarcity_ratio == pytest.approx(1 - 3 / 8) def test_over_cap_clamps_available_to_zero(self, service, repo): slots = _seed(repo, 3) for s in slots: service.transit_slot(s.slot_id, SlotState.INACTIVE, SlotState.PENDING_BUY, reason="r", actor="a") cap = service.compute_slot_capacity(max_active_slots=2) assert cap.active_count == 3 assert cap.available_count == 0 assert cap.scarcity_ratio == 1.0 def test_empty_instance(self, service): cap = service.compute_slot_capacity() assert cap.total_slots == 0 assert cap.available_count == 0 assert cap.scarcity_ratio == 0.0 # ---------------------------------------------------------------------------- # advisory lock serialization (REQ-002 AC-5) # ---------------------------------------------------------------------------- class TestAdvisoryLockSerialization: def test_concurrent_transitions_are_serialized(self, service, repo): """Two threads competing for the same slot — one must observe state mismatch.""" slot = _seed(repo)[0] results: dict[str, Exception | None] = {} def worker(name: str): try: service.transit_slot( slot.slot_id, SlotState.INACTIVE, SlotState.PENDING_BUY, reason=f"r-{name}", actor=name, ) results[name] = None except Exception as exc: # noqa: BLE001 results[name] = exc t1 = threading.Thread(target=worker, args=("buy_thread",)) t2 = threading.Thread(target=worker, args=("sell_thread",)) t1.start() t2.start() t1.join() t2.join() # exactly one succeeded; the other observed state mismatch successes = [k for k, v in results.items() if v is None] failures = [k for k, v in results.items() if v is not None] assert len(successes) == 1 assert len(failures) == 1 assert isinstance(failures[0] and results[failures[0]], SlotStateMismatchError) def test_lock_hold_time_under_50ms(self, service, repo): """REQ-002 AC-5 numeric: transition under the in-memory repo finishes < 50 ms.""" slot = _seed(repo)[0] t0 = time.perf_counter() service.transit_slot( slot.slot_id, SlotState.INACTIVE, SlotState.PENDING_BUY, reason="r", actor="a", ) elapsed_ms = (time.perf_counter() - t0) * 1000 assert elapsed_ms < 50.0, f"transition took {elapsed_ms:.2f} ms" # ---------------------------------------------------------------------------- # dcy-021: position sizing — increase transition + INACTIVE/sell qty semantics # ---------------------------------------------------------------------------- class TestPositionSizing: def _holding(self, service, repo, qty=10.0, avg=1000.0): slot = _seed(repo)[0] service.transit_slot(slot.slot_id, SlotState.INACTIVE, SlotState.PENDING_BUY, reason="buy", actor="t", symbol="005930", qty=qty) return service.transit_slot(slot.slot_id, SlotState.PENDING_BUY, SlotState.HOLDING, reason="filled", actor="t", avg_price=avg, qty=qty) def test_holding_to_pending_buy_increase_preserves_avg(self, service, repo): """증액 pending: HOLDING→PENDING_BUY 가 허용되고 avg_price(가중평균용)가 보존된다.""" h = self._holding(service, repo) inc = service.transit_slot(h.slot_id, SlotState.HOLDING, SlotState.PENDING_BUY, reason="position_increase", actor="t", qty=5, limit_price=1200) assert inc.status is SlotState.PENDING_BUY assert inc.avg_price == 1000.0 # preserved for weighted-average at fill assert inc.qty == 5 # order qty during pending assert inc.limit_price == 1200 def test_full_sell_pending_clears_qty_to_none(self, service, repo): """전량 매도(qty 미지정) PENDING_SELL 진입 시 slot.qty = None('전량' sentinel).""" h = self._holding(service, repo) ps = service.transit_slot(h.slot_id, SlotState.HOLDING, SlotState.PENDING_SELL, reason="sell", actor="t") # qty omitted assert ps.status is SlotState.PENDING_SELL and ps.qty is None def test_partial_sell_pending_records_order_qty(self, service, repo): """부분 매도 PENDING_SELL 진입 시 slot.qty = 주문 수량.""" h = self._holding(service, repo) ps = service.transit_slot(h.slot_id, SlotState.HOLDING, SlotState.PENDING_SELL, reason="sell", actor="t", qty=4, limit_price=1100) assert ps.qty == 4 and ps.limit_price == 1100 assert ps.avg_price == 1000.0 # cost basis preserved