chongming-utils-task/tests/unit/test_state_machine.py

"""状态机行为测试。

覆盖三类东西：

1. 转移表被**真正强制执行**，而不是只写在文档里；
2. 控制动词语义（stop / pause / resume / retry / reset / force）与设计文档一致；
3. 审计记录与版本号的不变量。
"""

from __future__ import annotations

import pytest

from chongming_task.core.state_machine import StateMachine, Trigger
from chongming_task.errors import AttemptsExhausted, InvalidTransition
from chongming_task.models import Failure, FailureKind, Progress, RequestedAction
from chongming_task.states import TaskStatus
from chongming_task.testing import FakeClock


@pytest.fixture
def clock() -> FakeClock:
    return FakeClock()


@pytest.fixture
def machine(clock: FakeClock) -> StateMachine:
    return StateMachine.create("t1", "抽取", max_attempts=3, clock=clock)


def _run_to_running(machine: StateMachine, executor: str = "inline") -> None:
    machine.mark_queued()
    machine.mark_starting(executor=executor)
    machine.mark_running()


# ---------------------------------------------------------------------------
# 创建与转移表强制
# ---------------------------------------------------------------------------


def test_create_starts_pending(machine: StateMachine, clock: FakeClock) -> None:
    assert machine.status is TaskStatus.PENDING
    assert machine.snapshot.created_at == clock.now
    assert machine.snapshot.attempt == 0
    assert machine.history == ()


def test_illegal_transition_raises_and_leaves_state_untouched(
    machine: StateMachine,
) -> None:
    before = machine.snapshot
    with pytest.raises(InvalidTransition):
        machine.mark_running()
    assert machine.snapshot is before
    assert machine.history == ()


def test_invalid_transition_message_lists_allowed_targets(machine: StateMachine) -> None:
    with pytest.raises(InvalidTransition) as info:
        machine.mark_succeeded()
    assert info.value.current is TaskStatus.PENDING
    assert info.value.requested is TaskStatus.SUCCEEDED
    assert "非法状态转移" in str(info.value)
    assert "queued" in str(info.value)


def test_happy_path_pending_to_succeeded(machine: StateMachine, clock: FakeClock) -> None:
    _run_to_running(machine, executor="thread")
    record = machine.mark_succeeded(outcome={"rows": 10})

    assert record.from_status is TaskStatus.RUNNING
    assert record.to_status is TaskStatus.SUCCEEDED
    assert machine.snapshot.outcome == {"rows": 10}
    assert machine.snapshot.ended_at == clock.now
    assert machine.snapshot.attempt == 1
    assert machine.snapshot.validate() == ()
    assert [t.to_status for t in machine.history] == [
        TaskStatus.QUEUED,
        TaskStatus.STARTING,
        TaskStatus.RUNNING,
        TaskStatus.SUCCEEDED,
    ]


def test_status_version_increases_by_one_per_transition(machine: StateMachine) -> None:
    _run_to_running(machine)
    assert [t.version for t in machine.history] == [1, 2, 3]
    assert machine.snapshot.status_version == 3


def test_terminal_state_cannot_transition_at_all(machine: StateMachine) -> None:
    _run_to_running(machine)
    machine.mark_succeeded()
    for action in (
        machine.mark_queued,
        machine.mark_blocked,
        machine.mark_running,
        machine.mark_cancelled,
        machine.mark_lost,
        machine.mark_skipped,
    ):
        with pytest.raises(InvalidTransition):
            action()
    assert machine.status is TaskStatus.SUCCEEDED


# ---------------------------------------------------------------------------
# 高频旁路更新
# ---------------------------------------------------------------------------


def test_heartbeat_and_progress_do_not_touch_status_version(
    machine: StateMachine, clock: FakeClock
) -> None:
    """心跳是高频写入；若参与乐观并发，会与状态写入持续冲突。"""
    machine.mark_queued()
    before = machine.snapshot.status_version
    clock.advance(5)
    machine.heartbeat()
    machine.update_progress(Progress(done=1, total=4))
    assert machine.snapshot.status_version == before
    assert machine.snapshot.last_heartbeat_at == clock.now
    assert machine.snapshot.progress is not None
    assert machine.snapshot.progress.done == 1


def test_transitions_use_injected_clock(machine: StateMachine, clock: FakeClock) -> None:
    machine.mark_queued()
    queued_at = machine.snapshot.queued_at
    clock.advance(30)
    machine.mark_starting(executor="inline")
    assert machine.history[-1].at == clock.now
    assert queued_at != machine.history[-1].at


# ---------------------------------------------------------------------------
# 重试预算
# ---------------------------------------------------------------------------


def test_attempt_is_consumed_only_when_starting(machine: StateMachine) -> None:
    """预算只在"真正要跑"的那一刻消费。"""
    machine.mark_queued()
    assert machine.snapshot.attempt == 0
    machine.mark_starting(executor="inline")
    assert machine.snapshot.attempt == 1


def test_retry_revives_from_failed_and_keeps_history(clock: FakeClock) -> None:
    machine = StateMachine.create("t1", "x", max_attempts=3, clock=clock)
    _run_to_running(machine)
    machine.mark_failed(Failure(kind=FailureKind.EXCEPTION, message="boom"))

    record = machine.retry()
    assert record.trigger == Trigger.RETRY_REQUESTED
    assert machine.status is TaskStatus.QUEUED
    assert machine.snapshot.failure is None
    assert machine.snapshot.ended_at is None
    assert machine.snapshot.attempt == 1  # 历史不被改写
    machine.mark_starting(executor="inline")
    assert machine.snapshot.attempt == 2


def test_retry_rejected_when_budget_exhausted(clock: FakeClock) -> None:
    machine = StateMachine.create("t1", "x", max_attempts=1, clock=clock)
    _run_to_running(machine)
    machine.mark_failed(Failure(kind=FailureKind.EXCEPTION, message="boom"))
    with pytest.raises(AttemptsExhausted):
        machine.retry()


def test_forced_retry_raises_the_budget(clock: FakeClock) -> None:
    """force=True 是"我知道它已失败够多次了，再试一次"。"""
    machine = StateMachine.create("t1", "x", max_attempts=1, clock=clock)
    _run_to_running(machine)
    machine.mark_failed(Failure(kind=FailureKind.EXCEPTION, message="boom"))

    machine.retry(force=True)
    assert machine.snapshot.max_attempts == 2
    machine.mark_starting(executor="inline")
    assert machine.snapshot.attempt == 2


def test_retry_rejected_from_non_revivable_terminal(clock: FakeClock) -> None:
    """SUCCEEDED / SKIPPED 不可复活：前者重跑无意义，后者是策略决定的结果。"""
    machine = StateMachine.create("t1", "x", clock=clock)
    _run_to_running(machine)
    machine.mark_succeeded()
    assert not machine.can_retry
    with pytest.raises(InvalidTransition):
        machine.retry()


def test_retry_is_not_the_path_for_retry_wait(clock: FakeClock) -> None:
    """RETRY_WAIT 已安排好重试，不应再被 retry() 接受，否则会双倍消费预算。"""
    machine = StateMachine.create("t1", "x", max_attempts=3, clock=clock)
    _run_to_running(machine)
    machine.mark_retry_wait(Failure(kind=FailureKind.EXCEPTION, message="boom"))
    with pytest.raises(InvalidTransition):
        machine.retry()


# ---------------------------------------------------------------------------
# 停止语义
# ---------------------------------------------------------------------------


def test_stop_on_unstarted_task_lands_stopped_without_buffer(
    machine: StateMachine,
) -> None:
    """还没跑起来的任务应当立即落 STOPPED；显示"停止中"是虚假信息。"""
    record = machine.request_stop(actor="ops")
    assert record.to_status is TaskStatus.STOPPED
    assert machine.snapshot.requested_action is None
    assert machine.snapshot.is_terminal


def test_stop_on_running_task_goes_through_stopping(machine: StateMachine) -> None:
    _run_to_running(machine)
    record = machine.request_stop(actor="ops", reason="发布窗口")
    assert record.to_status is TaskStatus.STOPPING
    assert machine.snapshot.requested_action is RequestedAction.STOP
    assert not machine.snapshot.is_terminal

    final = machine.mark_stopped()
    assert final.to_status is TaskStatus.STOPPED
    assert machine.snapshot.requested_action is None
    assert machine.snapshot.validate() == ()


def test_forced_stop_uses_distinct_trigger(machine: StateMachine) -> None:
    _run_to_running(machine, executor="thread")
    machine.request_stop(actor="ops")
    record = machine.mark_stopped(forced=True, reason="优雅窗口超时")
    assert record.trigger == Trigger.STOP_FORCED
    assert not record.is_override


def test_task_may_still_succeed_after_stop_requested(machine: StateMachine) -> None:
    """停止请求发出后任务恰好跑完，是合法结果，不能判为失败。"""
    _run_to_running(machine)
    machine.request_stop(actor="ops")
    record = machine.mark_succeeded()
    assert record.to_status is TaskStatus.SUCCEEDED


def test_stop_capability_is_false_on_terminal(machine: StateMachine) -> None:
    _run_to_running(machine)
    machine.mark_succeeded()
    assert not machine.can_stop
    with pytest.raises(InvalidTransition):
        machine.request_stop(actor="ops")


# ---------------------------------------------------------------------------
# 暂停语义
# ---------------------------------------------------------------------------


def test_pause_on_unstarted_task_lands_paused_directly(machine: StateMachine) -> None:
    """排队中的任务本就静止，pause() 应立即落 PAUSED，不该显示"暂停中"。"""
    machine.mark_queued()
    record = machine.request_pause(actor="ops")
    assert record.to_status is TaskStatus.PAUSED
    assert machine.status is TaskStatus.PAUSED
    assert machine.snapshot.requested_action is None
    assert machine.can_resume


def test_pause_while_pausing_is_not_short_circuited(machine: StateMachine) -> None:
    """已在 PAUSING 的任务不能被"加急"成 PAUSED：它还没到达检查点。"""
    _run_to_running(machine)
    machine.request_pause()
    assert machine.status is TaskStatus.PAUSING
    with pytest.raises(InvalidTransition):
        machine.request_pause()
    assert machine.status is TaskStatus.PAUSING


def test_starting_always_records_started_at(machine: StateMachine) -> None:
    """任何到达 RUNNING 的路径都必须带有 started_at（含暂停撤销路径）。"""
    machine.mark_queued()
    machine.mark_starting(executor="inline")
    assert machine.snapshot.started_at is not None
    machine.request_pause()
    machine.abort_pause(reason="未到达检查点")
    assert machine.status is TaskStatus.RUNNING
    assert machine.snapshot.validate() == ()


def test_pause_and_resume_round_trip(machine: StateMachine) -> None:
    _run_to_running(machine)
    machine.request_pause(actor="ops")
    assert machine.snapshot.requested_action is RequestedAction.PAUSE

    machine.mark_paused()
    assert machine.status is TaskStatus.PAUSED

    machine.resume()
    assert machine.status is TaskStatus.RUNNING
    assert machine.snapshot.requested_action is None


def test_inplace_resume_returns_to_running_not_queued(machine: StateMachine) -> None:
    """就地暂停的任务始终占着执行槽，调用栈还在内存里，恢复就是回到 RUNNING。

    我们没有办法把任意的 Python 调用栈换出内存再换回来，所以"暂停后重新排队"
    对就地暂停并不成立——那会导致任务状态与实际执行情况脱节。
    """
    _run_to_running(machine)
    machine.request_pause()
    machine.mark_paused()
    machine.resume()
    assert machine.status is TaskStatus.RUNNING
    assert machine.snapshot.queued_at is None or machine.snapshot.started_at is not None


def test_resume_of_a_never_started_pause_returns_to_queued(machine: StateMachine) -> None:
    """排队期间被暂停的任务从未启动过，恢复必须重新排队。"""
    machine.mark_queued()
    assert machine.request_pause().to_status is TaskStatus.PAUSED
    assert machine.snapshot.attempt == 0

    machine.resume()
    assert machine.status is TaskStatus.QUEUED


def test_resume_of_a_retry_wait_pause_returns_to_queued(clock: FakeClock) -> None:
    machine = StateMachine.create("t1", "x", max_attempts=3, clock=clock)
    _run_to_running(machine)
    machine.mark_retry_wait(Failure(kind=FailureKind.EXCEPTION, message="boom"))

    assert machine.request_pause().to_status is TaskStatus.PAUSED
    machine.resume()
    assert machine.status is TaskStatus.QUEUED


def test_aborted_pause_reverts_to_running(machine: StateMachine) -> None:
    """暂停没能生效时必须如实回滚，而不是假装已暂停、让用户以为资源已释放。"""
    _run_to_running(machine, executor="thread")
    machine.request_pause()
    record = machine.abort_pause(reason="任务未调用 checkpoint")
    assert record.to_status is TaskStatus.RUNNING
    assert record.trigger == Trigger.PAUSE_ABORTED
    assert machine.snapshot.requested_action is None


# ---------------------------------------------------------------------------
# 人工干预
# ---------------------------------------------------------------------------


def test_force_transition_requires_audit_fields(machine: StateMachine) -> None:
    with pytest.raises(ValueError, match="actor"):
        machine.force_transition(TaskStatus.SUCCEEDED, actor="   ", reason="ok")
    with pytest.raises(ValueError, match="reason"):
        machine.force_transition(TaskStatus.SUCCEEDED, actor="ops", reason="")


def test_force_transition_bypasses_table_and_marks_override(machine: StateMachine) -> None:
    """PENDING -> SUCCEEDED 在转移表中非法，人工改写必须能通过，但要留痕。"""
    record = machine.force_transition(
        TaskStatus.SUCCEEDED, actor="ops@example", reason="已人工核验产物"
    )
    assert record.to_status is TaskStatus.SUCCEEDED
    assert record.is_override
    assert record.actor == "ops@example"
    assert machine.snapshot.overridden
    assert machine.snapshot.validate() == ()


def test_reset_returns_to_pending_and_clears_attempt_state(clock: FakeClock) -> None:
    machine = StateMachine.create("t1", "x", max_attempts=2, clock=clock)
    _run_to_running(machine)
    machine.mark_failed(Failure(kind=FailureKind.EXCEPTION, message="boom"))

    machine.reset(actor="ops", reason="重新跑一遍")
    snapshot = machine.snapshot
    assert snapshot.status is TaskStatus.PENDING
    assert snapshot.attempt == 0
    assert snapshot.failure is None
    assert snapshot.ended_at is None
    assert snapshot.started_at is None
    assert not snapshot.overridden


# ---------------------------------------------------------------------------
# 能力预检（供控制面提前置灰按钮）
# ---------------------------------------------------------------------------


def test_capability_prechecks_match_actions(machine: StateMachine) -> None:
    assert machine.can_stop
    assert not machine.can_pause
    assert not machine.can_resume

    _run_to_running(machine)
    assert machine.can_stop
    assert machine.can_pause
    assert not machine.can_resume

    machine.request_pause()
    machine.mark_paused()
    assert machine.can_resume
    assert machine.can_stop


def test_lost_is_terminal_but_revivable(machine: StateMachine) -> None:
    _run_to_running(machine, executor="thread")
    machine.mark_lost(reason="心跳超时")
    assert machine.snapshot.is_terminal
    assert machine.can_retry