mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-10-29 03:18:14 +08:00
422 lines
15 KiB
Python
422 lines
15 KiB
Python
"""
|
||
工作流表的查询与状态流转行为。
|
||
|
||
调度器按「触发类型 + 状态」取工作流，取多了会把用户暂停的流程重新跑起来，
|
||
取少了则定时任务永远不触发。状态流转里的 `state != 'P'` 守卫是暂停语义的唯一实现，
|
||
`run_count` 的自增必须留在 SQL 侧，否则并发执行会丢计数。
|
||
"""
|
||
import asyncio
|
||
from dataclasses import FrozenInstanceError
|
||
|
||
import pytest
|
||
|
||
from app.application.workflow import WorkflowSnapshot
|
||
from app.db import base as db_base
|
||
from app.db.adapters.workflow import TransactionalWorkflowQueryRepository
|
||
from app.db.models.workflow import Workflow
|
||
from app.db.oper.workflow import WorkflowOper
|
||
from app.db.session import SessionFactory, async_session_scope
|
||
from app.schemas.workflow import Workflow as WorkflowResponse
|
||
|
||
|
||
@pytest.fixture(autouse=True)
|
||
def _track(db):
|
||
"""
|
||
把工作流表纳入用例级回收。
|
||
"""
|
||
db.watermark(Workflow)
|
||
|
||
|
||
def _flow(name: str, trigger_type: str = "timer", state: str = "W",
|
||
run_count: int = 0) -> Workflow:
|
||
"""
|
||
构造一条工作流记录。
|
||
"""
|
||
return Workflow(name=name, description=name, timer="0 * * * *",
|
||
trigger_type=trigger_type, state=state, run_count=run_count,
|
||
actions=[], flows=[], context={}, execution_state={})
|
||
|
||
|
||
async def _stage_async_action(workflow_id: int, action_id: str) -> None:
|
||
"""
|
||
用独占异步会话提交一次模型级暂存，模拟 Application UoW 边界。
|
||
"""
|
||
async with async_session_scope() as session:
|
||
await Workflow.async_update_current_action(
|
||
session,
|
||
wid=workflow_id,
|
||
action_id=action_id,
|
||
context={},
|
||
)
|
||
await session.commit()
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# 列表查询
|
||
# --------------------------------------------------------------------------- #
|
||
|
||
def test_list_and_get_by_name_match_async_twins(db):
|
||
"""
|
||
列举与按名查找的同步、异步结果必须一致。
|
||
"""
|
||
created = db.add(_flow("wf-name"))
|
||
|
||
assert Workflow.get_by_name(db.session, "wf-name").id == created.id
|
||
assert db.run_async_session(
|
||
lambda session: Workflow.async_get_by_name(session, "wf-name")
|
||
).id == created.id
|
||
assert Workflow.get_by_name(db.session, "wf-missing") is None
|
||
|
||
sync_ids = sorted(w.id for w in Workflow.list(db.session))
|
||
async_ids = sorted(w.id for w in db.run_async_session(Workflow.async_list))
|
||
assert sync_ids == async_ids
|
||
|
||
|
||
def test_workflow_oper_reuses_explicit_query_sessions(db, monkeypatch):
|
||
"""
|
||
WorkflowOper 绑定显式会话后不得再创建兼容查询会话。
|
||
"""
|
||
created = db.add(_flow("wf-explicit-session"))
|
||
monkeypatch.setattr(
|
||
db_base,
|
||
"run_sync_transaction",
|
||
lambda _operation: (_ for _ in ()).throw(
|
||
AssertionError("不应创建额外同步事务")
|
||
),
|
||
)
|
||
|
||
assert WorkflowOper(db.session).get_by_name(created.name).id == created.id
|
||
|
||
async def check() -> None:
|
||
"""
|
||
验证异步 Oper 同样复用调用方会话。
|
||
"""
|
||
async with async_session_scope() as session:
|
||
monkeypatch.setattr(
|
||
db_base,
|
||
"run_async_transaction",
|
||
lambda _operation: (_ for _ in ()).throw(
|
||
AssertionError("不应创建额外异步事务")
|
||
),
|
||
)
|
||
assert (await WorkflowOper(session).async_get_by_name(created.name)).id == created.id
|
||
|
||
asyncio.run(check())
|
||
|
||
|
||
def test_workflow_staged_writes_require_matching_caller_sessions(db) -> None:
|
||
"""
|
||
暂存入口拒绝缺失或错配会话，不能静默丢失删除或另开事务。
|
||
"""
|
||
with pytest.raises(RuntimeError, match="同步 Session"):
|
||
WorkflowOper().stage_delete(1)
|
||
|
||
async def check() -> None:
|
||
"""
|
||
分别覆盖同步删除和异步创建的会话类型边界。
|
||
"""
|
||
async with async_session_scope() as session:
|
||
with pytest.raises(RuntimeError, match="同步 Session"):
|
||
WorkflowOper(session).stage_delete(1)
|
||
for oper in (WorkflowOper(), WorkflowOper(db.session)):
|
||
with pytest.raises(RuntimeError, match="异步 Session"):
|
||
await oper.stage_create({"name": "wrong-session"})
|
||
|
||
asyncio.run(check())
|
||
|
||
|
||
def test_query_repository_returns_detached_deep_copied_snapshot(db):
|
||
"""
|
||
查询仓储必须在关闭短 Session 前投影，且 JSON 不与 ORM 记录共享。
|
||
"""
|
||
workflow = _flow("wf-snapshot")
|
||
workflow.actions = [{"id": "action-1", "config": {"value": 1}}]
|
||
workflow.flows = [{"source": "action-1", "target": "end"}]
|
||
workflow.context = {"nested": {"value": 1}}
|
||
created = db.add(workflow)
|
||
repository = TransactionalWorkflowQueryRepository(
|
||
sync_session=SessionFactory,
|
||
async_session=async_session_scope,
|
||
)
|
||
|
||
snapshot = repository.get(created.id)
|
||
|
||
assert isinstance(snapshot, WorkflowSnapshot)
|
||
assert snapshot.name == "wf-snapshot"
|
||
with pytest.raises(FrozenInstanceError):
|
||
snapshot.name = "changed"
|
||
snapshot.actions[0]["config"]["value"] = 2
|
||
snapshot.context["nested"]["value"] = 2
|
||
refreshed = repository.get(created.id)
|
||
assert refreshed.actions[0]["config"]["value"] == 1
|
||
assert refreshed.context["nested"]["value"] == 1
|
||
|
||
|
||
def test_query_repository_async_projection_survives_session_close(db):
|
||
"""
|
||
异步查询返回值在仓储退出 Session 作用域后仍可完整序列化。
|
||
"""
|
||
created = db.add(_flow("wf-async-snapshot"))
|
||
repository = TransactionalWorkflowQueryRepository(
|
||
sync_session=SessionFactory,
|
||
async_session=async_session_scope,
|
||
)
|
||
|
||
snapshot = asyncio.run(repository.async_get(created.id))
|
||
listed = asyncio.run(repository.async_list())
|
||
|
||
assert isinstance(snapshot, WorkflowSnapshot)
|
||
assert snapshot.name == "wf-async-snapshot"
|
||
assert created.id in {item.id for item in listed}
|
||
|
||
|
||
def test_query_repository_filters_paginates_and_counts_in_database(db):
|
||
"""
|
||
Agent 工作流筛选必须先进入 SQL，再按同一条件计数和分页。
|
||
"""
|
||
first = db.add(_flow("agent-page-first", trigger_type=None, state="W"))
|
||
second = db.add(_flow("agent-page-second", trigger_type="timer", state="W"))
|
||
db.add(
|
||
_flow("agent-page-paused", trigger_type="timer", state="P"),
|
||
_flow("unrelated-workflow", trigger_type="timer", state="W"),
|
||
)
|
||
repository = TransactionalWorkflowQueryRepository(
|
||
sync_session=SessionFactory,
|
||
async_session=async_session_scope,
|
||
)
|
||
|
||
total = asyncio.run(
|
||
repository.async_count(
|
||
state="W",
|
||
name="agent-page-",
|
||
trigger_type="timer",
|
||
)
|
||
)
|
||
page = asyncio.run(
|
||
repository.async_list(
|
||
state="W",
|
||
name="agent-page-",
|
||
trigger_type="timer",
|
||
page=2,
|
||
count=1,
|
||
)
|
||
)
|
||
|
||
assert total == 2
|
||
assert [item.id for item in page] == [second.id]
|
||
assert first.id < second.id
|
||
|
||
|
||
def test_workflow_snapshot_validates_against_api_response_contract(db):
|
||
"""
|
||
冻结快照可直接序列化为 API 合同且不会暴露内部执行上下文。
|
||
"""
|
||
created = db.add(_flow("wf-api-snapshot"))
|
||
repository = TransactionalWorkflowQueryRepository(
|
||
sync_session=SessionFactory,
|
||
async_session=async_session_scope,
|
||
)
|
||
|
||
response = WorkflowResponse.model_validate(repository.get(created.id))
|
||
payload = response.model_dump()
|
||
|
||
assert payload["id"] == created.id
|
||
assert payload["name"] == "wf-api-snapshot"
|
||
assert "context" not in payload
|
||
|
||
|
||
def test_enabled_workflows_exclude_paused(db):
|
||
"""
|
||
启用列表排除暂停状态。
|
||
|
||
暂停是用户显式的「别再跑了」，被列出即等于暂停开关失效。
|
||
"""
|
||
db.add(_flow("wf-waiting", state="W"), _flow("wf-running", state="R"),
|
||
_flow("wf-paused", state="P"))
|
||
|
||
names = {w.name for w in Workflow.get_enabled_workflows(db.session)}
|
||
|
||
assert {"wf-waiting", "wf-running"} <= names
|
||
assert "wf-paused" not in names
|
||
assert "wf-paused" not in {
|
||
w.name for w in db.run_async_session(Workflow.async_get_enabled_workflows)
|
||
}
|
||
|
||
|
||
def test_timer_triggered_includes_legacy_null_trigger_type(db):
|
||
"""
|
||
定时触发列表要包含 trigger_type 为空的历史数据。
|
||
|
||
该列是后加的，老工作流为空；严格等于 'timer' 会让它们从此再也不被调度，
|
||
而用户看到的只是「任务不跑了」。
|
||
"""
|
||
db.add(_flow("wf-timer", trigger_type="timer"),
|
||
_flow("wf-legacy", trigger_type=None),
|
||
_flow("wf-event", trigger_type="event"),
|
||
_flow("wf-timer-paused", trigger_type="timer", state="P"))
|
||
|
||
names = {w.name for w in Workflow.get_timer_triggered_workflows(db.session)}
|
||
|
||
assert {"wf-timer", "wf-legacy"} <= names
|
||
assert "wf-event" not in names
|
||
assert "wf-timer-paused" not in names
|
||
|
||
|
||
def test_event_triggered_requires_explicit_type(db):
|
||
"""
|
||
事件触发列表只认显式的 'event'，且同样排除暂停。
|
||
"""
|
||
db.add(_flow("wf-event", trigger_type="event"),
|
||
_flow("wf-event-paused", trigger_type="event", state="P"),
|
||
_flow("wf-legacy", trigger_type=None))
|
||
|
||
names = {w.name for w in Workflow.get_event_triggered_workflows(db.session)}
|
||
|
||
assert names >= {"wf-event"}
|
||
assert "wf-event-paused" not in names
|
||
assert "wf-legacy" not in names
|
||
|
||
|
||
def test_trigger_lists_match_async_twins(db):
|
||
"""
|
||
定时与事件两条触发列表的同步、异步结果必须一致。
|
||
"""
|
||
db.add(_flow("wf-t", trigger_type="timer"), _flow("wf-e", trigger_type="event"))
|
||
|
||
assert sorted(w.id for w in Workflow.get_timer_triggered_workflows(db.session)) == \
|
||
sorted(w.id for w in db.run_async_session(
|
||
Workflow.async_get_timer_triggered_workflows
|
||
))
|
||
assert sorted(w.id for w in Workflow.get_event_triggered_workflows(db.session)) == \
|
||
sorted(w.id for w in db.run_async_session(
|
||
Workflow.async_get_event_triggered_workflows
|
||
))
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# 状态流转
|
||
# --------------------------------------------------------------------------- #
|
||
|
||
def test_update_state_and_start_write_the_state(db):
|
||
"""
|
||
状态更新与启动直接落库，供调度器读到最新状态。
|
||
"""
|
||
flow = db.add(_flow("wf-state"))
|
||
|
||
Workflow.update_state(db.session, flow.id, "F")
|
||
assert Workflow.get_by_name(db.session, "wf-state").state == "F"
|
||
|
||
Workflow.start(db.session, flow.id)
|
||
assert Workflow.get_by_name(db.session, "wf-state").state == "R"
|
||
|
||
|
||
def test_fail_and_success_respect_the_paused_guard(db):
|
||
"""
|
||
暂停中的工作流不接受成功/失败结果写入。
|
||
|
||
守卫丢失后，一个还在跑的旧任务收尾时会把用户刚设的暂停状态改掉，
|
||
下一轮调度它又被跑起来。
|
||
"""
|
||
paused = db.add(_flow("wf-paused", state="P"))
|
||
|
||
Workflow.fail(db.session, paused.id, "出错了")
|
||
assert Workflow.get_by_name(db.session, "wf-paused").state == "P"
|
||
|
||
Workflow.success(db.session, paused.id, "完成")
|
||
assert Workflow.get_by_name(db.session, "wf-paused").state == "P"
|
||
|
||
|
||
def test_fail_records_result_and_timestamp(db):
|
||
"""
|
||
失败要同时写入结果与最后执行时间，供界面展示失败原因。
|
||
"""
|
||
flow = db.add(_flow("wf-fail", state="R"))
|
||
|
||
Workflow.fail(db.session, flow.id, "网络超时")
|
||
|
||
updated = Workflow.get_by_name(db.session, "wf-fail")
|
||
assert (updated.state, updated.result) == ("F", "网络超时")
|
||
assert updated.last_time
|
||
|
||
|
||
def test_success_increments_run_count_in_sql(db):
|
||
"""
|
||
执行次数必须在 SQL 侧自增。
|
||
|
||
先读后写会在并发执行时丢计数；连续两次成功后必须是 2。
|
||
"""
|
||
flow = db.add(_flow("wf-count", state="R", run_count=0))
|
||
|
||
Workflow.success(db.session, flow.id, "第一次")
|
||
Workflow.success(db.session, flow.id, "第二次")
|
||
|
||
updated = Workflow.get_by_name(db.session, "wf-count")
|
||
assert updated.run_count == 2
|
||
assert updated.state == "S"
|
||
|
||
|
||
def test_reset_clears_progress_and_optionally_the_count(db):
|
||
"""
|
||
重置清空执行进度；是否清零执行次数由参数决定。
|
||
|
||
默认保留计数是为了让「重跑」不丢失历史执行统计。
|
||
"""
|
||
flow = db.add(_flow("wf-reset", state="F", run_count=5))
|
||
Workflow.update_current_action(db.session, flow.id, "action-1",
|
||
{"k": "v"}, {"step": 1})
|
||
|
||
Workflow.reset(db.session, flow.id)
|
||
kept = Workflow.get_by_name(db.session, "wf-reset")
|
||
assert (kept.state, kept.result, kept.current_action) == ("W", None, None)
|
||
assert kept.context == {} and kept.execution_state == {}
|
||
assert kept.run_count == 5
|
||
|
||
Workflow.reset(db.session, flow.id, reset_count=True)
|
||
assert Workflow.get_by_name(db.session, "wf-reset").run_count == 0
|
||
|
||
|
||
def test_update_current_action_appends_without_duplicating(db):
|
||
"""
|
||
已执行动作按逗号追加且不重复登记。
|
||
|
||
重复登记会让「已执行」列表无限膨胀，重跑时的跳过判断也随之失准。
|
||
"""
|
||
flow = db.add(_flow("wf-action"))
|
||
|
||
Workflow.update_current_action(db.session, flow.id, "a1", {"n": 1})
|
||
Workflow.update_current_action(db.session, flow.id, "a2", {"n": 2})
|
||
Workflow.update_current_action(db.session, flow.id, "a1", {"n": 3})
|
||
|
||
updated = Workflow.get_by_name(db.session, "wf-action")
|
||
assert updated.current_action == "a1,a2"
|
||
assert updated.context == {"n": 3}
|
||
|
||
|
||
def test_update_current_action_leaves_execution_state_untouched_when_omitted(db):
|
||
"""
|
||
不传执行状态时保持原值，避免一次进度更新把结构化状态清空。
|
||
"""
|
||
flow = db.add(_flow("wf-keep-state"))
|
||
Workflow.update_current_action(db.session, flow.id, "a1", {}, {"step": 7})
|
||
|
||
Workflow.update_current_action(db.session, flow.id, "a2", {"n": 1})
|
||
|
||
assert Workflow.get_by_name(db.session, "wf-keep-state").execution_state == {"step": 7}
|
||
|
||
|
||
def test_update_current_action_matches_async_twin(db):
|
||
"""
|
||
同步与异步的动作追加必须给出相同的字符串。
|
||
"""
|
||
sync_flow = db.add(_flow("wf-sync-action"))
|
||
async_flow = db.add(_flow("wf-async-action"))
|
||
|
||
for action in ("a1", "a2", "a1"):
|
||
Workflow.update_current_action(db.session, sync_flow.id, action, {})
|
||
# 同步 Model 方法只暂存 SQL；由测试持有的事务边界先提交，避免与异步会话争锁。
|
||
db.session.commit()
|
||
asyncio.run(_stage_async_action(async_flow.id, action))
|
||
|
||
assert Workflow.get_by_name(db.session, "wf-sync-action").current_action == \
|
||
Workflow.get_by_name(db.session, "wf-async-action").current_action
|