lhl
2f4397a8d6
- Issue3: 新建 src/genesis/state_machine.py（9 状态白名单转移） - cancel 记录 cancelled_from 进 cancelled 终态，resume 回中断点 - 仅执行中状态可取消（awaiting_*/done 不可），非法转移抛 StateTransitionError - 同步 agent-runtime-design.md §3.2（状态图/规则表 9 状态 + cancelled 行） - 新增 10 用例（含覆盖补齐 2 项），全量 177 passed / 100.00%（981 stmts/252 br）
128 lines
3.5 KiB
Python
128 lines
3.5 KiB
Python
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from genesis.state_machine import StateTransitionError, SessionStateMachine
|
|
|
|
|
|
# ---------- 正常流转 ----------
|
|
|
|
def test_initial_state_is_uploading():
|
|
sm = SessionStateMachine()
|
|
assert sm.state == "uploading"
|
|
|
|
|
|
def test_unknown_initial_state_rejected():
|
|
with pytest.raises(StateTransitionError):
|
|
SessionStateMachine(initial="bogus")
|
|
|
|
|
|
def test_unknown_target_state_rejected():
|
|
sm = SessionStateMachine()
|
|
with pytest.raises(StateTransitionError):
|
|
sm.transition("bogus")
|
|
|
|
|
|
def test_happy_path_full_flow():
|
|
sm = SessionStateMachine()
|
|
sm.transition("parsing")
|
|
sm.transition("awaiting_parse_confirm")
|
|
sm.transition("impact_running")
|
|
sm.transition("awaiting_impact_confirm")
|
|
sm.transition("writing")
|
|
sm.transition("qa")
|
|
sm.transition("done")
|
|
assert sm.state == "done"
|
|
|
|
|
|
# ---------- 非法转移拒绝 ----------
|
|
|
|
def test_illegal_transition_rejected():
|
|
sm = SessionStateMachine()
|
|
with pytest.raises(StateTransitionError):
|
|
sm.transition("writing") # uploading → writing 非法
|
|
|
|
|
|
def test_done_is_terminal_no_transition_out():
|
|
sm = SessionStateMachine()
|
|
sm.transition("parsing")
|
|
sm.transition("awaiting_parse_confirm")
|
|
sm.transition("impact_running")
|
|
sm.transition("awaiting_impact_confirm")
|
|
sm.transition("writing")
|
|
sm.transition("qa")
|
|
sm.transition("done")
|
|
with pytest.raises(StateTransitionError):
|
|
sm.transition("writing")
|
|
|
|
|
|
# ---------- T3: cancelled + resume ----------
|
|
|
|
def test_cancel_from_running_state():
|
|
sm = SessionStateMachine()
|
|
sm.transition("parsing")
|
|
sm.transition("awaiting_parse_confirm")
|
|
sm.transition("impact_running")
|
|
sm.transition("awaiting_impact_confirm")
|
|
sm.transition("writing")
|
|
sm.cancel()
|
|
assert sm.state == "cancelled"
|
|
assert sm.cancelled_from == "writing"
|
|
|
|
|
|
def test_cancel_from_done_is_illegal():
|
|
sm = SessionStateMachine()
|
|
sm.transition("parsing")
|
|
sm.transition("awaiting_parse_confirm")
|
|
sm.transition("impact_running")
|
|
sm.transition("awaiting_impact_confirm")
|
|
sm.transition("writing")
|
|
sm.transition("qa")
|
|
sm.transition("done")
|
|
with pytest.raises(StateTransitionError):
|
|
sm.cancel()
|
|
|
|
|
|
def test_resume_returns_to_interrupted_state():
|
|
sm = SessionStateMachine()
|
|
sm.transition("parsing")
|
|
sm.transition("awaiting_parse_confirm")
|
|
sm.transition("impact_running")
|
|
sm.transition("awaiting_impact_confirm")
|
|
sm.transition("writing")
|
|
sm.cancel()
|
|
sm.resume()
|
|
assert sm.state == "writing"
|
|
|
|
|
|
def test_resume_without_cancel_is_illegal():
|
|
sm = SessionStateMachine()
|
|
sm.transition("parsing")
|
|
with pytest.raises(StateTransitionError):
|
|
sm.resume() # 非 cancelled 状态不能 resume
|
|
|
|
|
|
def test_cancelled_cannot_jump_to_arbitrary_state():
|
|
sm = SessionStateMachine()
|
|
sm.transition("parsing")
|
|
sm.cancel()
|
|
with pytest.raises(StateTransitionError):
|
|
sm.transition("done") # cancelled 只能 resume 回中断前状态
|
|
|
|
|
|
def test_resume_after_resume_allows_forward_progress():
|
|
# 取消 → resume → 再次取消 → resume 循环稳定（cancelled_from 保持最后一次中断点）
|
|
sm = SessionStateMachine()
|
|
sm.transition("parsing")
|
|
sm.transition("awaiting_parse_confirm")
|
|
sm.transition("impact_running")
|
|
sm.cancel()
|
|
sm.resume()
|
|
assert sm.state == "impact_running"
|
|
sm.transition("awaiting_impact_confirm")
|
|
sm.transition("writing")
|
|
sm.cancel()
|
|
assert sm.cancelled_from == "writing"
|
|
sm.resume()
|
|
assert sm.state == "writing"
|