150 lines
5.2 KiB
Python
150 lines
5.2 KiB
Python
"""``ConsciousnessNode.start_workflow_design`` fire workflow ray task 的提交逻辑。
|
||
|
||
历史上这里有一个常驻的 ``WorkflowRunningEngine`` actor 做中转，现已删除：
|
||
workflow 是一次性、有头有尾的执行，更适合直接以 ray task 形式触发。
|
||
本测试保证 ConsciousnessNode 在工作流生成后正确 fire ``run_workflow_task``，
|
||
并通过 ``put_pending`` 推送 SSE 进度（节点端写 pending → API 端 SSE 读 pending）。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from types import SimpleNamespace
|
||
from unittest.mock import AsyncMock, MagicMock
|
||
|
||
import pytest
|
||
|
||
from kilostar.core.work.workflow import workflow_engine as engine_module
|
||
from kilostar.core.work.workflow.workflow import KiloStarWorkflow
|
||
from kilostar.core.work.workflow.model import WorkflowMetadata
|
||
|
||
|
||
@pytest.fixture
|
||
def consciousness_instance():
|
||
from kilostar.core.individual.consciousness_node.consciousness_node import (
|
||
ConsciousnessNode,
|
||
)
|
||
from kilostar.utils.logger import get_logger
|
||
|
||
cls = ConsciousnessNode.__ray_actor_class__
|
||
obj = cls.__new__(cls)
|
||
obj.logger = get_logger("consciousness_node")
|
||
obj.agent = None
|
||
return obj
|
||
|
||
|
||
class _FakeActorRef:
|
||
"""模拟 ``ray_actor_hook("name").<name>`` 的链式取属性返回值。"""
|
||
|
||
def __init__(self, target):
|
||
self._target = target
|
||
|
||
def __getattr__(self, item):
|
||
return getattr(self._target, item)
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_start_workflow_design_fires_run_workflow_task(
|
||
consciousness_instance, monkeypatch
|
||
):
|
||
"""快乐路径：working 返回 ForWorkflowEngine，应 fire run_workflow_task 且推送 pending。"""
|
||
from kilostar.core.individual.consciousness_node.template import (
|
||
ForWorkflowEngine,
|
||
)
|
||
|
||
wf = KiloStarWorkflow(
|
||
title="t",
|
||
work_link=[],
|
||
workflow_metadata=WorkflowMetadata(),
|
||
)
|
||
|
||
consciousness_instance.working = AsyncMock(
|
||
return_value=ForWorkflowEngine(workflow=wf, reasoning="r")
|
||
)
|
||
|
||
postgres = MagicMock()
|
||
postgres.get_all_worker_individual = MagicMock()
|
||
postgres.get_all_worker_individual.remote = AsyncMock(return_value=[])
|
||
postgres.update_workflow_status = MagicMock()
|
||
postgres.update_workflow_status.remote = AsyncMock()
|
||
|
||
pending_writes: list[tuple[str, str]] = []
|
||
|
||
gwm = MagicMock()
|
||
gwm.put_pending = MagicMock()
|
||
gwm.put_pending.remote = AsyncMock(
|
||
side_effect=lambda tid, msg: pending_writes.append((tid, msg))
|
||
)
|
||
|
||
def _fake_hook(name):
|
||
if name == "postgres_database":
|
||
return SimpleNamespace(postgres_database=_FakeActorRef(postgres))
|
||
if name == "global_workflow_manager":
|
||
return SimpleNamespace(global_workflow_manager=_FakeActorRef(gwm))
|
||
raise KeyError(name)
|
||
|
||
import kilostar.core.individual.consciousness_node.consciousness_node as cmod
|
||
|
||
monkeypatch.setattr(cmod, "ray_actor_hook", _fake_hook)
|
||
|
||
captured: dict = {}
|
||
|
||
def _fake_task_remote(workflow_dict, trace_id):
|
||
captured["workflow_dict"] = workflow_dict
|
||
captured["trace_id"] = trace_id
|
||
return MagicMock()
|
||
|
||
monkeypatch.setattr(engine_module.run_workflow_task, "remote", _fake_task_remote)
|
||
|
||
await consciousness_instance.start_workflow_design("trace-123", "do something")
|
||
|
||
assert captured["trace_id"] == "trace-123"
|
||
assert captured["workflow_dict"]["title"] == "t"
|
||
# SSE 推送方向必须是 pending（put_pending）
|
||
assert any("正在为您构建" in msg for _, msg in pending_writes)
|
||
assert any("即将开始执行" in msg for _, msg in pending_writes)
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_start_workflow_design_failed_path_marks_failed(
|
||
consciousness_instance, monkeypatch
|
||
):
|
||
"""working 返回 None / 不匹配类型时应推送失败提示并把 workflow 状态置为 failed。"""
|
||
consciousness_instance.working = AsyncMock(return_value=None)
|
||
|
||
postgres = MagicMock()
|
||
postgres.get_all_worker_individual = MagicMock()
|
||
postgres.get_all_worker_individual.remote = AsyncMock(return_value=[])
|
||
postgres.update_workflow_status = MagicMock()
|
||
postgres.update_workflow_status.remote = AsyncMock()
|
||
|
||
pending_writes: list[tuple[str, str]] = []
|
||
gwm = MagicMock()
|
||
gwm.put_pending = MagicMock()
|
||
gwm.put_pending.remote = AsyncMock(
|
||
side_effect=lambda tid, msg: pending_writes.append((tid, msg))
|
||
)
|
||
|
||
def _fake_hook(name):
|
||
if name == "postgres_database":
|
||
return SimpleNamespace(postgres_database=_FakeActorRef(postgres))
|
||
if name == "global_workflow_manager":
|
||
return SimpleNamespace(global_workflow_manager=_FakeActorRef(gwm))
|
||
raise KeyError(name)
|
||
|
||
import kilostar.core.individual.consciousness_node.consciousness_node as cmod
|
||
|
||
monkeypatch.setattr(cmod, "ray_actor_hook", _fake_hook)
|
||
|
||
fired: list = []
|
||
monkeypatch.setattr(
|
||
engine_module.run_workflow_task,
|
||
"remote",
|
||
lambda *a, **kw: fired.append((a, kw)),
|
||
)
|
||
|
||
await consciousness_instance.start_workflow_design("trace-x", "cmd")
|
||
|
||
assert fired == [] # 没有 fire ray task
|
||
postgres.update_workflow_status.remote.assert_awaited_with("trace-x", "failed")
|
||
assert any("生成失败" in msg for _, msg in pending_writes)
|