136 lines
4.8 KiB
Python
136 lines
4.8 KiB
Python
|
"""状态表的结构性不变量。
|
||
|
|
|
||
|
|
这些测试只约束 :mod:`chongming_task.states` 的数据本身，不依赖状态机实现。
|
||
|
|
它们的作用是：让"转移表是唯一真相"成为可执行的断言，而不是文档里的承诺。
|
||
|
|
"""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import pytest
|
||
|
|
|
||
|
|
from chongming_task.states import (
|
||
|
|
NON_TERMINAL,
|
||
|
|
REVIVABLE,
|
||
|
|
TERMINAL,
|
||
|
|
TRANSITIONS,
|
||
|
|
TaskStatus,
|
||
|
|
allowed_from,
|
||
|
|
can_transition,
|
||
|
|
is_terminal,
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
def test_transition_table_is_exhaustive() -> None:
|
||
|
|
"""每个规范状态都必须有表项，否则状态机会在运行期 KeyError。"""
|
||
|
|
assert set(TRANSITIONS) == set(TaskStatus)
|
||
|
|
|
||
|
|
|
||
|
|
def test_alias_does_not_add_a_table_key() -> None:
|
||
|
|
"""``FINISHED`` 只是别名，不应在表中占据独立条目。"""
|
||
|
|
assert len(TRANSITIONS) == len(set(TaskStatus))
|
||
|
|
|
||
|
|
|
||
|
|
def test_terminal_set_is_derived_from_the_table() -> None:
|
||
|
|
"""终态必须由表推导，避免"表"与"集合"两处维护而不一致。"""
|
||
|
|
derived = frozenset(status for status, nxt in TRANSITIONS.items() if not nxt)
|
||
|
|
assert derived == TERMINAL
|
||
|
|
|
||
|
|
|
||
|
|
def test_terminal_and_non_terminal_partition_the_enum() -> None:
|
||
|
|
partition = TERMINAL | NON_TERMINAL
|
||
|
|
assert not (TERMINAL & NON_TERMINAL)
|
||
|
|
assert partition == set(TaskStatus)
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.parametrize("status", sorted(TERMINAL))
|
||
|
|
def test_terminal_states_have_no_outgoing_transitions(status: TaskStatus) -> None:
|
||
|
|
assert allowed_from(status) == frozenset()
|
||
|
|
for target in TaskStatus:
|
||
|
|
assert not can_transition(status, target)
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.parametrize("status", sorted(NON_TERMINAL))
|
||
|
|
def test_non_terminal_states_have_outgoing_transitions(status: TaskStatus) -> None:
|
||
|
|
"""非终态若无出边，任务会永久卡死。"""
|
||
|
|
assert allowed_from(status), f"{status} 必须有至少一条出边"
|
||
|
|
|
||
|
|
|
||
|
|
def test_no_self_loops() -> None:
|
||
|
|
"""自环会掩盖状态机缺陷，并让 status_version 无意义地增长。"""
|
||
|
|
for source, targets in TRANSITIONS.items():
|
||
|
|
assert source not in targets, f"{source} 存在自环"
|
||
|
|
|
||
|
|
|
||
|
|
def test_all_targets_are_known_statuses() -> None:
|
||
|
|
known = set(TaskStatus)
|
||
|
|
for targets in TRANSITIONS.values():
|
||
|
|
assert targets <= known
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.parametrize("start", sorted(NON_TERMINAL))
|
||
|
|
def test_every_non_terminal_state_can_reach_a_terminal_state(start: TaskStatus) -> None:
|
||
|
|
"""不存在只能内部互跳、永远收敛不了的死角。"""
|
||
|
|
seen: set[TaskStatus] = set()
|
||
|
|
stack: list[TaskStatus] = [start]
|
||
|
|
while stack:
|
||
|
|
current = stack.pop()
|
||
|
|
if current in TERMINAL:
|
||
|
|
return
|
||
|
|
if current in seen:
|
||
|
|
continue
|
||
|
|
seen.add(current)
|
||
|
|
stack.extend(allowed_from(current))
|
||
|
|
pytest.fail(f"{start} 无法到达任何终态")
|
||
|
|
|
||
|
|
|
||
|
|
# ---------------------------------------------------------------------------
|
||
|
|
# 以下用例锁定设计决策本身，防止后续重构悄悄改变语义
|
||
|
|
# ---------------------------------------------------------------------------
|
||
|
|
|
||
|
|
|
||
|
|
def test_paused_can_resume_either_in_place_or_back_to_the_queue() -> None:
|
||
|
|
"""恢复目标取决于暂停发生在哪一层：
|
||
|
|
|
||
|
|
* 就地暂停（经 ``PAUSING``）→ ``RUNNING``，槽位从未释放；
|
||
|
|
* 未启动即暂停 → ``QUEUED``，重新排队。
|
||
|
|
"""
|
||
|
|
assert can_transition(TaskStatus.PAUSED, TaskStatus.RUNNING)
|
||
|
|
assert can_transition(TaskStatus.PAUSED, TaskStatus.QUEUED)
|
||
|
|
|
||
|
|
|
||
|
|
def test_stopping_may_still_succeed() -> None:
|
||
|
|
"""停止请求发出后任务恰好自然结束，是合法结果，不能判为失败。"""
|
||
|
|
assert can_transition(TaskStatus.STOPPING, TaskStatus.SUCCEEDED)
|
||
|
|
|
||
|
|
|
||
|
|
def test_pending_cannot_jump_straight_to_running() -> None:
|
||
|
|
"""PENDING 必须经过 QUEUED/STARTING，不允许绕过并发槽与执行器启动。"""
|
||
|
|
assert not can_transition(TaskStatus.PENDING, TaskStatus.RUNNING)
|
||
|
|
|
||
|
|
|
||
|
|
def test_running_cannot_return_to_queued_directly() -> None:
|
||
|
|
"""RUNNING 回落 QUEUED 只能经由 PAUSED，否则会绕过暂停语义。"""
|
||
|
|
assert not can_transition(TaskStatus.RUNNING, TaskStatus.QUEUED)
|
||
|
|
|
||
|
|
|
||
|
|
def test_revivable_is_a_subset_of_terminal() -> None:
|
||
|
|
assert REVIVABLE <= TERMINAL
|
||
|
|
|
||
|
|
|
||
|
|
def test_success_and_skip_are_not_revivable() -> None:
|
||
|
|
"""成功重跑没有意义；SKIPPED 是策略决定的结果，不应被 retry 复活。"""
|
||
|
|
assert TaskStatus.SUCCEEDED not in REVIVABLE
|
||
|
|
assert TaskStatus.SKIPPED not in REVIVABLE
|
||
|
|
|
||
|
|
|
||
|
|
def test_legacy_finished_alias() -> None:
|
||
|
|
assert TaskStatus.FINISHED is TaskStatus.SUCCEEDED
|
||
|
|
assert "finished" not in {s.value for s in TaskStatus}
|
||
|
|
|
||
|
|
|
||
|
|
def test_helpers_agree_with_the_table() -> None:
|
||
|
|
assert is_terminal(TaskStatus.SUCCEEDED)
|
||
|
|
assert not is_terminal(TaskStatus.RUNNING)
|
||
|
|
assert can_transition(TaskStatus.RUNNING, TaskStatus.STOPPING)
|
||
|
|
assert not can_transition(TaskStatus.SUCCEEDED, TaskStatus.RUNNING)
|