genichinandClaude Opus 4.8
017b8185e0
안 B 점진의 2/3단계. runtime 무변경(idle/closed 별칭 흡수)으로 cascade 통제. - state.py: SlotState.INACTIVE 추가 + idle/closed 별칭 normalize_state(); 전이 집합 inactive 기준 재정의(inactive↔pending_buy↔holding↔pending_sell), closed→idle 제거; is_allowed_transition 양쪽 정규화; slot_advisory_lock_id(instance_key) 헬퍼 - service.py: SlotService(instance_key) + advisory lock 인스턴스 스코프(self._lock_id) - __init__: normalize_state/slot_advisory_lock_id export; daemon: instance_key 전달 - 테스트 갱신: test_state(전이/멤버7/락/size12), test_service(closed→idle 제거→reallocation), test_slot_entry_count, test_postgresql_repository(closed→idle 거부+reallocation) 검증: 단위 375 passed / 35 skipped, PG-gated(trex_core_test) 8 passed (마이그레이션 009 드리프트 + lifecycle/reallocation), ruff clean. REQ-056/057 은 토대만 충족 → Approved 유지(entry_count/trading_date drop·literal inactive 전면 전환은 BAT-δ; confirm-req-implemented 는 dcy-020 후속 배치 완료 후). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
121 lines
4.5 KiB
Python
121 lines
4.5 KiB
Python
"""State machine unit tests — asset-centric redefinition (dcy-020 / REQ-057).
|
|
|
|
`inactive` is the canonical no-position state; `idle`/`closed` are aliases that
|
|
normalize to `inactive`. `closed->idle` recovery is removed. Covers REQ-057.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from trex_core.domain.slot.state import (
|
|
ALLOWED_TRANSITIONS,
|
|
SLOT_ADVISORY_LOCK_ID,
|
|
SlotState,
|
|
is_allowed_transition,
|
|
normalize_state,
|
|
slot_advisory_lock_id,
|
|
)
|
|
|
|
|
|
def test_slot_state_members():
|
|
"""dcy-020: 5 canonical + 2 aliases(idle/closed) = 7 enum members."""
|
|
assert {s.value for s in SlotState} == {
|
|
"inactive",
|
|
"idle",
|
|
"pending_buy",
|
|
"holding",
|
|
"pending_sell",
|
|
"closed",
|
|
"error",
|
|
}
|
|
|
|
|
|
def test_idle_and_closed_normalize_to_inactive():
|
|
assert normalize_state(SlotState.IDLE) is SlotState.INACTIVE
|
|
assert normalize_state(SlotState.CLOSED) is SlotState.INACTIVE
|
|
assert normalize_state(SlotState.INACTIVE) is SlotState.INACTIVE
|
|
assert normalize_state(SlotState.HOLDING) is SlotState.HOLDING
|
|
|
|
|
|
def test_advisory_lock_id_is_stable_constant():
|
|
"""SLOT_ADVISORY_LOCK_ID base must not change between releases (REQ-002 NFR-1)."""
|
|
assert isinstance(SLOT_ADVISORY_LOCK_ID, int)
|
|
assert SLOT_ADVISORY_LOCK_ID == 0x5107_4D31
|
|
|
|
|
|
def test_slot_advisory_lock_id_is_instance_scoped():
|
|
"""dcy-020/REQ-057: distinct instance_key -> distinct lock id; same -> same."""
|
|
a = slot_advisory_lock_id("inst-a")
|
|
b = slot_advisory_lock_id("inst-b")
|
|
assert a != b
|
|
assert slot_advisory_lock_id("inst-a") == a
|
|
# 'default' keeps a stable derived id
|
|
assert slot_advisory_lock_id("default") == slot_advisory_lock_id("default")
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"from_state,to_state",
|
|
[
|
|
# canonical (inactive-based)
|
|
(SlotState.INACTIVE, SlotState.PENDING_BUY),
|
|
(SlotState.PENDING_BUY, SlotState.HOLDING),
|
|
(SlotState.PENDING_BUY, SlotState.INACTIVE),
|
|
(SlotState.HOLDING, SlotState.PENDING_SELL),
|
|
(SlotState.PENDING_SELL, SlotState.INACTIVE),
|
|
(SlotState.PENDING_SELL, SlotState.HOLDING),
|
|
# alias equivalence: idle/closed normalize to inactive
|
|
(SlotState.IDLE, SlotState.PENDING_BUY), # idle->pending_buy
|
|
(SlotState.CLOSED, SlotState.PENDING_BUY), # closed->pending_buy (reallocation)
|
|
(SlotState.PENDING_SELL, SlotState.CLOSED), # sell filled (closed alias)
|
|
(SlotState.PENDING_BUY, SlotState.IDLE), # buy rollback (idle alias)
|
|
# to error (every non-error)
|
|
(SlotState.INACTIVE, SlotState.ERROR),
|
|
(SlotState.PENDING_BUY, SlotState.ERROR),
|
|
(SlotState.HOLDING, SlotState.ERROR),
|
|
(SlotState.PENDING_SELL, SlotState.ERROR),
|
|
(SlotState.CLOSED, SlotState.ERROR),
|
|
# operator resolve out of error
|
|
(SlotState.ERROR, SlotState.INACTIVE),
|
|
(SlotState.ERROR, SlotState.IDLE), # alias
|
|
(SlotState.ERROR, SlotState.HOLDING),
|
|
],
|
|
)
|
|
def test_allowed_transition(from_state: SlotState, to_state: SlotState):
|
|
assert is_allowed_transition(from_state, to_state)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"from_state,to_state",
|
|
[
|
|
# closed->idle recovery REMOVED (normalizes to inactive self-loop)
|
|
(SlotState.CLOSED, SlotState.IDLE),
|
|
# skipping intermediate states
|
|
(SlotState.INACTIVE, SlotState.HOLDING),
|
|
(SlotState.INACTIVE, SlotState.PENDING_SELL),
|
|
(SlotState.HOLDING, SlotState.INACTIVE),
|
|
(SlotState.HOLDING, SlotState.CLOSED),
|
|
(SlotState.PENDING_BUY, SlotState.PENDING_SELL),
|
|
(SlotState.PENDING_SELL, SlotState.PENDING_BUY),
|
|
# error has restricted exits — only inactive(+aliases) and holding
|
|
(SlotState.ERROR, SlotState.PENDING_BUY),
|
|
(SlotState.ERROR, SlotState.PENDING_SELL),
|
|
(SlotState.ERROR, SlotState.ERROR),
|
|
],
|
|
)
|
|
def test_disallowed_transition(from_state: SlotState, to_state: SlotState):
|
|
assert not is_allowed_transition(from_state, to_state)
|
|
|
|
|
|
def test_self_loops_are_not_allowed():
|
|
"""No state may transition to itself (including inactive and its aliases)."""
|
|
for s in SlotState:
|
|
assert not is_allowed_transition(s, s)
|
|
# alias self-loops also rejected
|
|
assert not is_allowed_transition(SlotState.IDLE, SlotState.CLOSED)
|
|
assert not is_allowed_transition(SlotState.CLOSED, SlotState.IDLE)
|
|
|
|
|
|
def test_allowed_transitions_set_size():
|
|
"""6 base (inactive-canonical) + 4 to-error + 2 operator-resolve = 12."""
|
|
assert len(ALLOWED_TRANSITIONS) == 12
|