mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-10-30 03:48:03 +08:00
460 lines
17 KiB
Python
460 lines
17 KiB
Python
"""
|
||
工作流准入、真实线程生命周期、截止时间和图定义的行为回归。
|
||
"""
|
||
|
||
import asyncio
|
||
import json
|
||
import threading
|
||
from types import SimpleNamespace
|
||
from unittest.mock import AsyncMock, Mock
|
||
|
||
import pytest
|
||
|
||
from app import workflow as workflow_package
|
||
from app.application.workflow import WorkflowDefinitionCommand, WorkflowMutationCommand
|
||
from app.chain import workflow as workflow_chain
|
||
from app.schemas.workflow import Action, ActionResult
|
||
|
||
|
||
def _workflow(workflow_id=1, *, actions=None, flows=None, current_action=None):
|
||
"""
|
||
构造包含串行两个动作的脱离数据库工作流快照。
|
||
"""
|
||
return SimpleNamespace(
|
||
id=workflow_id, name=f"workflow-{workflow_id}", current_action=current_action,
|
||
actions=actions if actions is not None else [
|
||
{"id": "A", "name": "A", "type": "ObservedAction"},
|
||
{"id": "B", "name": "B", "type": "ObservedAction"},
|
||
],
|
||
flows=flows if flows is not None else [{"id": "AB", "source": "A", "target": "B"}],
|
||
context={}, execution_config={}, execution_state={},
|
||
)
|
||
|
||
|
||
def _manager(action_type):
|
||
"""
|
||
提供真实准入锁和动作调度能力，不启动应用或加载外部插件。
|
||
"""
|
||
manager = object.__new__(workflow_package.WorkflowManager)
|
||
manager._lock = threading.RLock()
|
||
manager._actions = {"ObservedAction": action_type}
|
||
manager._event_workflows = {}
|
||
manager._accepting_executions = True
|
||
manager._executions = {}
|
||
return manager
|
||
|
||
|
||
def _install_runtime(monkeypatch, manager, workflows):
|
||
"""
|
||
把真实执行链绑定到可观察的内存查询和事务端口。
|
||
"""
|
||
port = SimpleNamespace(reset=Mock(return_value=True), start=Mock(return_value=True),
|
||
step=Mock(return_value=True), success=Mock(return_value=True), fail=Mock(return_value=True))
|
||
monkeypatch.setattr(workflow_chain, "get_workflow_manager", lambda: manager)
|
||
monkeypatch.setattr(workflow_chain, "get_configured_workflow_execution", lambda: port)
|
||
monkeypatch.setattr(workflow_chain, "get_configured_workflow_query",
|
||
lambda: SimpleNamespace(get_sync=workflows.get))
|
||
monkeypatch.setattr(workflow_package.runtime_stop_state, "is_workflow_stopped", lambda _wid: False)
|
||
return port
|
||
|
||
|
||
def test_same_workflow_from_begin_cannot_reset_active_execution(monkeypatch):
|
||
"""
|
||
真实节点线程阻塞时，同 ID 重跑必须在重置和写运行中状态之前被拒绝。
|
||
"""
|
||
entered, release = threading.Event(), threading.Event()
|
||
calls = []
|
||
|
||
class ObservedAction:
|
||
"""
|
||
记录动作并把第一个节点阻塞到竞争请求结束。
|
||
"""
|
||
|
||
def __init__(self, action_id):
|
||
"""
|
||
保存当前节点 ID。
|
||
"""
|
||
self.action_id = action_id
|
||
|
||
def execute_with_inputs(self, workflow_id, params, inputs, runtime, context):
|
||
"""
|
||
在真实池线程内阻塞节点 A，其他节点直接成功。
|
||
"""
|
||
_ = workflow_id, params, inputs, runtime
|
||
calls.append(self.action_id)
|
||
if self.action_id == "A":
|
||
entered.set()
|
||
assert release.wait(2)
|
||
return ActionResult(success=True, message="", context=context)
|
||
|
||
manager = _manager(ObservedAction)
|
||
port = _install_runtime(monkeypatch, manager, {1: _workflow(current_action="A,B")})
|
||
results = []
|
||
thread = threading.Thread(target=lambda: results.append(workflow_chain.WorkflowChain.process(1)), daemon=True)
|
||
thread.start()
|
||
try:
|
||
assert entered.wait(1), "from_begin 应清除旧节点状态并真正执行 A"
|
||
refused = workflow_chain.WorkflowChain.process(1, from_begin=True)
|
||
assert refused[0] is False
|
||
port.reset.assert_called_once_with(1)
|
||
port.start.assert_called_once_with(1)
|
||
assert calls == ["A"]
|
||
finally:
|
||
release.set()
|
||
thread.join(2)
|
||
assert not thread.is_alive()
|
||
assert results == [(True, "")]
|
||
assert calls == ["A", "B"]
|
||
assert manager._executions == {}
|
||
|
||
|
||
@pytest.mark.parametrize("succeeded", [True, False])
|
||
def test_workflow_retains_owner_until_terminal_transaction_commits(monkeypatch, succeeded):
|
||
"""
|
||
成功和失败终态提交阻塞时均保留同 ID 所有权，防止覆盖后续运行。
|
||
"""
|
||
entered, release = threading.Event(), threading.Event()
|
||
|
||
class ObservedAction:
|
||
"""
|
||
以预定结果结束动作，令执行进入终态提交。
|
||
"""
|
||
|
||
def __init__(self, action_id):
|
||
"""
|
||
接收动作标识并保持无副作用。
|
||
"""
|
||
_ = action_id
|
||
|
||
def execute_with_inputs(self, workflow_id, params, inputs, runtime, context):
|
||
"""
|
||
直接结束，成功和失败走各自真实写入分支。
|
||
"""
|
||
_ = workflow_id, params, inputs, runtime
|
||
return ActionResult(success=succeeded, message="failed", context=context)
|
||
|
||
manager = _manager(ObservedAction)
|
||
port = _install_runtime(monkeypatch, manager, {1: _workflow()})
|
||
|
||
def persist_terminal(*_args, **_kwargs):
|
||
"""
|
||
让终态事务在尚未完成提交时保持阻塞。
|
||
"""
|
||
entered.set()
|
||
assert release.wait(2)
|
||
return True
|
||
|
||
terminal = port.success if succeeded else port.fail
|
||
terminal.side_effect = persist_terminal
|
||
results = []
|
||
thread = threading.Thread(target=lambda: results.append(workflow_chain.WorkflowChain.process(1)), daemon=True)
|
||
thread.start()
|
||
try:
|
||
assert entered.wait(1)
|
||
assert manager._executions
|
||
assert workflow_chain.WorkflowChain.process(1)[0] is False
|
||
owner = next(iter(manager._executions.values()))
|
||
assert owner.wait_stopped(timeout=0.01) is False
|
||
port.reset.assert_called_once()
|
||
port.start.assert_called_once()
|
||
finally:
|
||
release.set()
|
||
thread.join(2)
|
||
assert not thread.is_alive()
|
||
assert results[0][0] is succeeded
|
||
assert manager._executions == {}
|
||
|
||
|
||
def test_different_workflows_can_execute_concurrently(monkeypatch):
|
||
"""
|
||
同 ID 互斥不应把不同工作流串行化。
|
||
"""
|
||
entered = {1: threading.Event(), 2: threading.Event()}
|
||
release = threading.Event()
|
||
|
||
class ObservedAction:
|
||
"""
|
||
在两个工作流的首节点上记录并发进入。
|
||
"""
|
||
|
||
def __init__(self, action_id):
|
||
"""
|
||
保存节点 ID。
|
||
"""
|
||
self.action_id = action_id
|
||
|
||
def execute_with_inputs(self, workflow_id, params, inputs, runtime, context):
|
||
"""
|
||
首节点保持阻塞，以证明两个不同 ID 同时被准入。
|
||
"""
|
||
_ = params, inputs, runtime
|
||
if self.action_id == "A":
|
||
entered[workflow_id].set()
|
||
assert release.wait(2)
|
||
return ActionResult(success=True, message="", context=context)
|
||
|
||
manager = _manager(ObservedAction)
|
||
_install_runtime(monkeypatch, manager, {wid: _workflow(wid) for wid in entered})
|
||
results = []
|
||
threads = [threading.Thread(target=lambda wid=wid: results.append(workflow_chain.WorkflowChain.process(wid)),
|
||
daemon=True) for wid in entered]
|
||
for thread in threads:
|
||
thread.start()
|
||
try:
|
||
assert all(event.wait(1) for event in entered.values())
|
||
assert len(manager._executions) == 2
|
||
finally:
|
||
release.set()
|
||
for thread in threads:
|
||
thread.join(2)
|
||
assert results == [(True, ""), (True, "")]
|
||
|
||
|
||
@pytest.mark.parametrize("rejected_operation", ["reset", "start"])
|
||
def test_failed_start_transaction_releases_unstarted_owner(monkeypatch, rejected_operation):
|
||
"""
|
||
重置或启动事务返回 False 时，不得执行节点或提交成功终态。
|
||
"""
|
||
manager = _manager(Mock())
|
||
port = _install_runtime(monkeypatch, manager, {1: _workflow()})
|
||
getattr(port, rejected_operation).return_value = False
|
||
success, message = workflow_chain.WorkflowChain.process(1)
|
||
assert success is False
|
||
assert "失败" in message
|
||
assert manager._executions == {}
|
||
manager._actions["ObservedAction"].assert_not_called()
|
||
port.success.assert_not_called()
|
||
port.fail.assert_not_called()
|
||
if rejected_operation == "reset":
|
||
port.start.assert_not_called()
|
||
|
||
|
||
@pytest.mark.parametrize("commit_failure", [False, RuntimeError("terminal failed")])
|
||
def test_terminal_commit_failure_retains_owner_for_retry(monkeypatch, commit_failure):
|
||
"""
|
||
终态未提交时不能报告成功或释放 owner，收口重试成功后才允许新执行。
|
||
"""
|
||
action_type = Mock()
|
||
action_type.get_contract.return_value = {}
|
||
action_type.return_value.execute_with_inputs.return_value = ActionResult(success=True, message="")
|
||
manager = _manager(action_type)
|
||
port = _install_runtime(monkeypatch, manager, {1: _workflow()})
|
||
port.success.side_effect = [commit_failure, True]
|
||
with pytest.raises(RuntimeError):
|
||
workflow_chain.WorkflowChain.process(1)
|
||
assert manager._executions
|
||
assert workflow_chain.WorkflowChain.process(1)[0] is False
|
||
port.start.assert_called_once()
|
||
owner = next(iter(manager._executions.values()))
|
||
assert owner.wait_stopped(timeout=1) is True
|
||
assert port.success.call_count == 2
|
||
assert manager._executions == {}
|
||
|
||
|
||
@pytest.mark.parametrize("loop", [False, True])
|
||
def test_action_timeout_checks_result_and_deadline_after_sleep(monkeypatch, loop):
|
||
"""
|
||
同步返回和循环等待跨过截止时间时均失败，等待后不能再次调用动作。
|
||
"""
|
||
clock = [0.0]
|
||
calls, sleeps = [], []
|
||
|
||
class ObservedAction:
|
||
"""
|
||
用离线时钟模拟同步阻塞或未完成的循环动作。
|
||
"""
|
||
|
||
done = False
|
||
|
||
def __init__(self, action_id):
|
||
"""
|
||
保留构造兼容，不引入外部能力。
|
||
"""
|
||
_ = action_id
|
||
|
||
def execute_with_inputs(self, workflow_id, params, inputs, runtime, context):
|
||
"""
|
||
非循环跨过截止时间，循环首轮保持未完成。
|
||
"""
|
||
_ = workflow_id, params, inputs, runtime
|
||
calls.append("execute")
|
||
if not loop:
|
||
clock[0] = 2
|
||
return ActionResult(success=True, message="", context=context)
|
||
|
||
def simulated_sleep(_wid, seconds, _token):
|
||
"""
|
||
推进测试时钟并记录调度器裁剪后的等待时间。
|
||
"""
|
||
sleeps.append(seconds)
|
||
clock[0] += seconds
|
||
|
||
manager = _manager(ObservedAction)
|
||
monkeypatch.setattr(workflow_package, "monotonic", lambda: clock[0])
|
||
monkeypatch.setattr(manager, "_sleep_with_cancel", simulated_sleep)
|
||
monkeypatch.setattr(workflow_package.runtime_stop_state, "is_workflow_stopped", lambda _wid: False)
|
||
result = manager.execute(1, Action(id="A", type="ObservedAction", timeout=1,
|
||
data={"loop": loop, "loop_interval": 10}))
|
||
assert result.success is False
|
||
assert "超时" in result.message
|
||
assert calls == ["execute"]
|
||
assert sleeps == ([1] if loop else [])
|
||
|
||
|
||
def test_cooperative_action_can_observe_deadline_cancel_token(monkeypatch):
|
||
"""
|
||
支持取消的动作通过运行期令牌看到本次尝试的截止时间。
|
||
"""
|
||
clock = [0.0]
|
||
|
||
class ObservedAction:
|
||
"""
|
||
在动作内部协作检查截止时间，而非让管理器分离后台线程。
|
||
"""
|
||
|
||
def __init__(self, action_id):
|
||
"""
|
||
接收动作标识。
|
||
"""
|
||
_ = action_id
|
||
|
||
def execute_with_inputs(self, workflow_id, params, inputs, runtime, context):
|
||
"""
|
||
直到统一令牌报告取消才退出，并验证传入截止时间。
|
||
"""
|
||
_ = workflow_id, params, inputs
|
||
token = runtime["cancel_token"]
|
||
assert runtime["deadline"] == 1
|
||
while not token.is_cancelled():
|
||
clock[0] += 0.25
|
||
return ActionResult(success=False, message="cancelled", context=context)
|
||
|
||
manager = _manager(ObservedAction)
|
||
monkeypatch.setattr(workflow_package, "monotonic", lambda: clock[0])
|
||
monkeypatch.setattr(workflow_package.runtime_stop_state, "is_workflow_stopped", lambda _wid: False)
|
||
result = manager.execute(1, Action(id="A", type="ObservedAction", timeout=1))
|
||
assert result.success is False
|
||
assert "超时" in result.message
|
||
assert clock[0] == 1
|
||
|
||
|
||
def test_noncooperative_timeout_keeps_owner_until_thread_really_returns(monkeypatch):
|
||
"""
|
||
到达截止时间也不能分离阻塞线程或提前放行同 ID 的下一次执行。
|
||
"""
|
||
entered, release, finished = threading.Event(), threading.Event(), threading.Event()
|
||
clock = [0.0]
|
||
|
||
class ObservedAction:
|
||
"""
|
||
模拟不读取协作令牌的第三方同步动作。
|
||
"""
|
||
|
||
def __init__(self, action_id):
|
||
"""
|
||
接收动作标识。
|
||
"""
|
||
_ = action_id
|
||
|
||
def execute_with_inputs(self, workflow_id, params, inputs, runtime, context):
|
||
"""
|
||
忽略令牌，直到测试显式释放才真实返回。
|
||
"""
|
||
_ = workflow_id, params, inputs, runtime
|
||
entered.set()
|
||
assert release.wait(2)
|
||
return ActionResult(success=True, message="", context=context)
|
||
|
||
manager = _manager(ObservedAction)
|
||
definition = _workflow(actions=[{"id": "A", "name": "A", "type": "ObservedAction", "timeout": 1}], flows=[])
|
||
port = _install_runtime(monkeypatch, manager, {1: definition})
|
||
monkeypatch.setattr(workflow_package, "monotonic", lambda: clock[0])
|
||
results = []
|
||
|
||
def run_workflow():
|
||
"""
|
||
在独立宿主线程内运行，只有完整 process 返回才设置完成标志。
|
||
"""
|
||
try:
|
||
results.append(workflow_chain.WorkflowChain.process(1))
|
||
finally:
|
||
finished.set()
|
||
|
||
thread = threading.Thread(target=run_workflow, daemon=True)
|
||
thread.start()
|
||
try:
|
||
assert entered.wait(1)
|
||
clock[0] = 2
|
||
assert finished.wait(0.05) is False
|
||
assert manager._executions
|
||
assert workflow_chain.WorkflowChain.process(1)[0] is False
|
||
port.reset.assert_called_once()
|
||
port.fail.assert_not_called()
|
||
port.success.assert_not_called()
|
||
finally:
|
||
release.set()
|
||
thread.join(2)
|
||
assert results[0][0] is False
|
||
assert "超时" in port.step.call_args.kwargs["execution_state"]["nodes"]["A"]["message"]
|
||
port.fail.assert_called_once()
|
||
assert manager._executions == {}
|
||
|
||
|
||
_INVALID_GRAPHS = [
|
||
([{"id": "A"}, {"id": "A"}], [], "重复"),
|
||
([{"id": "A"}], [{"source": "A", "target": "missing"}], "不存在"),
|
||
([{"id": "A"}], [{"source": "A", "target": ""}], "端点"),
|
||
([{"id": "A"}, {"id": "B"}],
|
||
[{"source": "A", "target": "B"}, {"source": "B", "target": "A"}], "循环"),
|
||
([{"id": "A"}, {"id": "B"}, {"id": "C"}],
|
||
[{"source": "B", "target": "C"}, {"source": "C", "target": "B"}], "循环"),
|
||
([{"id": "A"}, {"id": "B"}],
|
||
[{"id": "same", "source": "A", "target": "B"}, {"id": "same", "source": "A", "target": "B"}], "重复"),
|
||
]
|
||
|
||
|
||
@pytest.mark.parametrize("actions,flows,error", _INVALID_GRAPHS)
|
||
def test_persisted_invalid_graph_cannot_report_runtime_success(monkeypatch, actions, flows, error):
|
||
"""
|
||
即使坏定义已持久化，运行入口也应拒绝且不写成功或启动状态。
|
||
"""
|
||
manager = _manager(object)
|
||
port = _install_runtime(monkeypatch, manager, {1: _workflow(actions=actions, flows=flows)})
|
||
success, message = workflow_chain.WorkflowChain.process(1)
|
||
assert success is False
|
||
assert error in message
|
||
port.start.assert_not_called()
|
||
port.success.assert_not_called()
|
||
|
||
|
||
@pytest.mark.parametrize("actions,flows,error", _INVALID_GRAPHS)
|
||
@pytest.mark.parametrize("operation", ["create", "update", "fork"])
|
||
def test_invalid_graph_is_rejected_before_definition_write(actions, flows, error, operation):
|
||
"""
|
||
创建、更新和复用共享同一图校验，错误不触发任何数据库写入。
|
||
"""
|
||
repository = SimpleNamespace(async_get_by_name=AsyncMock(return_value=None),
|
||
stage_create=AsyncMock(), stage_update=Mock(),
|
||
get=Mock(return_value=SimpleNamespace(trigger_type="manual", actions=[], flows=[])))
|
||
if operation == "update":
|
||
unit = SimpleNamespace(commit=Mock(), rollback=Mock())
|
||
command = WorkflowMutationCommand(
|
||
repository=repository, unit_of_work=unit, add_timer=Mock(), remove_timer=Mock(),
|
||
load_event=Mock(), remove_event=Mock(), refresh_event=Mock(), stop_running=Mock(), delete_cache=Mock(),
|
||
)
|
||
result = command.update({"id": 1, "actions": actions, "flows": flows})
|
||
repository.stage_update.assert_not_called()
|
||
unit.commit.assert_not_called()
|
||
else:
|
||
unit = SimpleNamespace(commit=AsyncMock(), rollback=AsyncMock())
|
||
command = WorkflowDefinitionCommand(repository=repository, unit_of_work=unit,
|
||
stop_running=Mock(), async_delete_cache=AsyncMock())
|
||
payload = {"name": "invalid", "actions": actions, "flows": flows}
|
||
if operation == "fork":
|
||
payload.update(actions=json.dumps(actions), flows=json.dumps(flows))
|
||
result = asyncio.run(getattr(command, operation)(payload))
|
||
repository.stage_create.assert_not_awaited()
|
||
unit.commit.assert_not_awaited()
|
||
assert result.success is False
|
||
assert error in result.message
|