chakcy
f4de43c743
单任务受控执行（15 态状态机 + 审计轨迹）、DAG 管道、Controller 门面。
执行器三个：线程（默认，含 async def 任务）、内联（同步 / 测试）、进程
（spawn 子进程：硬性停止、绕开 GIL、崩溃隔离）。
调度：并发额度 + 每桶限流 + 具名资源信号量（按标签）+ 优先级排队。
持久层：TaskStore 契约 + 内存 / SQLite / Postgres 实现，契约测试随包发布。
跨进程与多节点：写穿 / 读穿、指令通道与租约、owned_buckets 划界、reconcile。
CLI、12 个可直接运行的示例、docs/ 下 5 份文档。
门禁：ruff / ruff format / mypy(strict) 干净；
pytest 563 passed（44 skipped 是需要真实 Postgres 的那批）；
带一台可连的 Postgres 时 607 passed。
119 lines
4.8 KiB
Python
119 lines
4.8 KiB
Python
"""用 hypothesis 随机生成事件序列，验证状态机的核心不变量。
|
|
|
|
比手写用例更强的地方：它能生成人手想不到的组合，例如
|
|
"暂停中收到停止"、"重试等待中被重试"、"重试预算耗尽后又被派发"这类长尾序列。
|
|
|
|
被验证的不变量
|
|
--------------
|
|
1. 被拒绝的动作**完全不产生副作用**（快照对象不变、历史不增长）。
|
|
2. 每次被接受的动作使 ``status_version`` 恰好 +1，并使历史恰好 +1 条。
|
|
3. 除显式审计的复活路径外，实际发生的转移必在转移表内。
|
|
4. 终态不可逃逸。
|
|
5. 每次被接受的动作之后，快照都通过 :meth:`TaskSnapshot.validate` 的跨字段一致性检查。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import TYPE_CHECKING
|
|
|
|
from hypothesis import HealthCheck, given, settings
|
|
from hypothesis import strategies as st
|
|
|
|
from chongming_task.core.state_machine import StateMachine, Trigger
|
|
from chongming_task.errors import AttemptsExhausted, InvalidTransition
|
|
from chongming_task.models import Failure, FailureKind
|
|
from chongming_task.states import TERMINAL, TRANSITIONS
|
|
from chongming_task.testing import FakeClock
|
|
|
|
if TYPE_CHECKING:
|
|
from collections.abc import Callable
|
|
|
|
_FAILURE = Failure(kind=FailureKind.EXCEPTION, message="boom")
|
|
|
|
#: 刻意绕过转移表的触发源，只有这三条审计路径合法。
|
|
_NON_TABLE_TRIGGERS = frozenset(
|
|
{
|
|
Trigger.RETRY_REQUESTED,
|
|
Trigger.RESET_REQUESTED,
|
|
Trigger.ADMIN_OVERRIDE,
|
|
}
|
|
)
|
|
|
|
#: 随机动作集合。
|
|
#:
|
|
#: 刻意不含 ``force_transition``：它按定义可跳到任意状态，与"随机事件序列"这一
|
|
#: 模型正交，其行为由 tests/unit/test_state_machine.py 的定向用例覆盖。
|
|
ACTIONS: tuple[tuple[str, Callable[[StateMachine], object]], ...] = (
|
|
("mark_blocked", lambda m: m.mark_blocked()),
|
|
("mark_queued", lambda m: m.mark_queued()),
|
|
("mark_starting", lambda m: m.mark_starting(executor="inline")),
|
|
("mark_running", lambda m: m.mark_running()),
|
|
("mark_succeeded", lambda m: m.mark_succeeded()),
|
|
("mark_failed", lambda m: m.mark_failed(_FAILURE)),
|
|
("mark_retry_wait", lambda m: m.mark_retry_wait(_FAILURE)),
|
|
("mark_retry_exhausted", lambda m: m.mark_retry_exhausted(_FAILURE)),
|
|
("mark_skipped", lambda m: m.mark_skipped()),
|
|
("mark_cancelled", lambda m: m.mark_cancelled()),
|
|
("mark_lost", lambda m: m.mark_lost()),
|
|
("request_stop", lambda m: m.request_stop()),
|
|
("mark_stopped", lambda m: m.mark_stopped()),
|
|
("request_pause", lambda m: m.request_pause()),
|
|
("mark_paused", lambda m: m.mark_paused()),
|
|
("abort_pause", lambda m: m.abort_pause()),
|
|
("resume", lambda m: m.resume()),
|
|
("retry", lambda m: m.retry()),
|
|
("force_retry", lambda m: m.retry(force=True)),
|
|
("reset", lambda m: m.reset()),
|
|
)
|
|
|
|
|
|
@settings(
|
|
max_examples=500,
|
|
deadline=None,
|
|
suppress_health_check=[HealthCheck.too_slow],
|
|
)
|
|
@given(
|
|
max_attempts=st.integers(min_value=1, max_value=4),
|
|
picks=st.lists(st.integers(min_value=0, max_value=len(ACTIONS) - 1), max_size=30),
|
|
)
|
|
def test_invariants_hold_for_any_action_sequence(max_attempts: int, picks: list[int]) -> None:
|
|
clock = FakeClock()
|
|
machine = StateMachine.create("t", "x", max_attempts=max_attempts, clock=clock)
|
|
|
|
for index in picks:
|
|
name, action = ACTIONS[index]
|
|
clock.advance(1)
|
|
before = machine.snapshot
|
|
history_len = len(machine.history)
|
|
|
|
try:
|
|
action(machine)
|
|
except (InvalidTransition, AttemptsExhausted):
|
|
# 不变量 1：被拒绝的动作必须完全无副作用。
|
|
assert machine.snapshot is before, f"{name} 被拒绝却改动了快照"
|
|
assert len(machine.history) == history_len, f"{name} 被拒绝却写了历史"
|
|
continue
|
|
|
|
after = machine.snapshot
|
|
record = machine.history[-1]
|
|
|
|
# 不变量 2：版本号与历史同步严格 +1。
|
|
assert len(machine.history) == history_len + 1, name
|
|
assert after.status_version == before.status_version + 1, name
|
|
|
|
# 不变量 3：只有审计路径可以绕过转移表。
|
|
if record.trigger not in _NON_TABLE_TRIGGERS:
|
|
assert after.status in TRANSITIONS[before.status], (
|
|
f"{name} 绕过了转移表：{before.status} -> {after.status}"
|
|
)
|
|
|
|
# 不变量 4：终态不可逃逸——**复活除外**，后者必须走审计路径。
|
|
# 终态出边为空集，因此这一条其实是上一条的推论；显式写出是为了失败时
|
|
# 能一眼分清"逃逸"与"绕表"。
|
|
if record.trigger not in _NON_TABLE_TRIGGERS:
|
|
assert before.status not in TERMINAL, f"{name} 从终态 {before.status} 逃逸"
|
|
|
|
# 不变量 5：快照跨字段一致。
|
|
problems = after.validate()
|
|
assert problems == (), f"{name} 产生不一致快照：{problems}"
|