qiuruiandClaude Sonnet 4.6
900e388f08
- executor.py：__init__ 增加 event_bus 可选参数；_run_task 在 mark_done/mark_failed 后发布对应事件 - scheduler.py：_run_pending_queue 启动时尝试连接 Redis，可用时注入 EventBus，不可用时降级静默 - ops.py：新增 OpsHub 类，消费 ops-hub consumer group 事件，写入 DailyReport 并对 alert/version_pass 入队广播任务 - tests/test_executor.py：新增 test_executor_publishes_event_on_done，使用 fakeredis 验证事件发布 - tests/test_ops_hub.py：新增两个测试，覆盖事件→日报写入和 alert→任务分发链路 Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
216 lines
8.3 KiB
Python
216 lines
8.3 KiB
Python
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import pytest
|
||
from pathlib import Path
|
||
from unittest.mock import MagicMock, patch
|
||
from nmfs_agents.core.executor import Executor
|
||
from nmfs_agents.core.queue import TaskQueue, Task
|
||
from nmfs_agents.config import AgentsConfig, ProjectConfig, SchedulerConfig, ClaudeConfig, FeishuConfig
|
||
from nmfs_agents.tools.project_memory import ProjectMemory
|
||
|
||
|
||
def _cfg(tmp_path: Path) -> AgentsConfig:
|
||
return AgentsConfig(
|
||
projects={"yolo": ProjectConfig(path=tmp_path / "yolo", mode="report")},
|
||
scheduler=SchedulerConfig(max_concurrent=2),
|
||
claude=ClaudeConfig(api_key="fake"),
|
||
feishu=FeishuConfig(),
|
||
devices={},
|
||
)
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_run_next_marks_task_done(tmp_path):
|
||
cfg = _cfg(tmp_path)
|
||
q = TaskQueue(db_path=tmp_path / "tasks.db")
|
||
tid = q.enqueue(Task(project="yolo", type="fix_bug", title="测试",
|
||
priority=1, mode="report", agent_role="developer"))
|
||
mem = ProjectMemory(db_path=tmp_path / "mem.db")
|
||
executor = Executor(cfg, queue=q, memory=mem)
|
||
with patch("nmfs_agents.core.executor.DeveloperAgent") as MockAgent:
|
||
mock_agent = MockAgent.return_value
|
||
mock_agent.run.return_value = MagicMock(status="done", summary="测试完成")
|
||
await executor.run_next()
|
||
# 任务应标记为 done
|
||
with q._conn() as c:
|
||
row = c.execute("SELECT status FROM tasks WHERE id=?", (tid,)).fetchone()
|
||
assert row["status"] == "done"
|
||
|
||
|
||
def test_executor_writes_history_after_task(tmp_path):
|
||
"""Executor must write task result to ProjectMemory history after completion."""
|
||
import asyncio
|
||
from unittest.mock import patch
|
||
from nmfs_agents.core.executor import Executor
|
||
from nmfs_agents.core.queue import TaskQueue, Task
|
||
from nmfs_agents.tools.project_memory import ProjectMemory
|
||
from nmfs_agents.agents.developer import AgentResult
|
||
|
||
mem = ProjectMemory(db_path=tmp_path / "mem.db")
|
||
queue = TaskQueue(db_path=tmp_path / "q.db")
|
||
queue.enqueue(Task(project="yolo", type="fix_bug", title="修复零检出",
|
||
context="", priority=2, mode="report", agent_role="developer"))
|
||
|
||
cfg = _cfg(tmp_path)
|
||
executor = Executor(cfg, queue=queue, memory=mem)
|
||
|
||
mock_result = AgentResult(status="done", summary="已修复")
|
||
with patch("nmfs_agents.agents.developer.DeveloperAgent.run",
|
||
return_value=mock_result):
|
||
asyncio.run(executor.run_next())
|
||
|
||
history = mem.get_recent_history("yolo")
|
||
assert len(history) == 1
|
||
assert history[0]["status"] == "done"
|
||
assert history[0]["task_title"] == "修复零检出"
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_confirm_mode_marks_done_not_waiting(tmp_path):
|
||
"""confirm 模式任务完成后直接 mark_done（hook 级审批已在执行中发生，无需 task-level 等待）"""
|
||
cfg = _cfg(tmp_path)
|
||
q = TaskQueue(db_path=tmp_path / "tasks.db")
|
||
tid = q.enqueue(Task(project="yolo", type="fix_bug", title="修复",
|
||
priority=1, mode="confirm", agent_role="developer"))
|
||
mem = ProjectMemory(db_path=tmp_path / "mem.db")
|
||
executor = Executor(cfg, queue=q, memory=mem)
|
||
|
||
mock_result = MagicMock(status="done", summary="修复完成")
|
||
with patch("nmfs_agents.core.executor._make_agent") as mock_make:
|
||
mock_make.return_value.run.return_value = mock_result
|
||
await executor.run_next()
|
||
|
||
with q._conn() as c:
|
||
row = c.execute("SELECT status FROM tasks WHERE id=?", (tid,)).fetchone()
|
||
assert row["status"] == "done" # 不是 waiting_confirm
|
||
|
||
|
||
def test_notify_wx_called_on_task_run(tmp_path):
|
||
"""任务执行时调用 _notify_wx（task-start + task-end）"""
|
||
import asyncio
|
||
from unittest.mock import patch, MagicMock
|
||
from nmfs_agents.core.executor import Executor
|
||
from nmfs_agents.core.queue import TaskQueue, Task
|
||
from nmfs_agents.agents.developer import AgentResult
|
||
from nmfs_agents.tools.project_memory import ProjectMemory
|
||
|
||
cfg = _cfg(tmp_path)
|
||
cfg.claude_wx_url = "http://localhost:9999"
|
||
|
||
queue = TaskQueue(db_path=tmp_path / "q.db")
|
||
queue.enqueue(Task(project="yolo", type="fix_bug", title="测试通知",
|
||
context="", priority=1, mode="auto", agent_role="developer"))
|
||
mem = ProjectMemory(db_path=tmp_path / "mem.db")
|
||
executor = Executor(cfg, queue=queue, memory=mem)
|
||
|
||
notified_paths = []
|
||
def fake_notify(path, data):
|
||
notified_paths.append(path)
|
||
|
||
executor._notify_wx = fake_notify
|
||
mock_result = AgentResult(status="done", summary="完成")
|
||
with patch("nmfs_agents.agents.developer.DeveloperAgent.run", return_value=mock_result):
|
||
asyncio.run(executor.run_next())
|
||
|
||
assert "/agent/task-start" in notified_paths
|
||
assert "/agent/task-end" in notified_paths
|
||
|
||
|
||
def test_notify_wx_skipped_when_no_url(tmp_path):
|
||
"""claude_wx_url 未配置时，_notify_wx 不发请求"""
|
||
from nmfs_agents.core.executor import Executor
|
||
from nmfs_agents.config import AgentsConfig, ProjectConfig, SchedulerConfig, ClaudeConfig, FeishuConfig
|
||
|
||
cfg = AgentsConfig(
|
||
projects={},
|
||
scheduler=SchedulerConfig(),
|
||
claude=ClaudeConfig(api_key="fake"),
|
||
feishu=FeishuConfig(),
|
||
devices={},
|
||
claude_wx_url="", # 未配置
|
||
)
|
||
executor = Executor(cfg)
|
||
# 不应抛出异常，也不发请求
|
||
executor._notify_wx("/agent/task-start", {"task_id": 1})
|
||
|
||
|
||
@pytest.fixture
|
||
def mock_config(tmp_path):
|
||
from nmfs_agents.config import AgentsConfig, SchedulerConfig, ClaudeConfig, FeishuConfig
|
||
from unittest.mock import MagicMock
|
||
cfg = MagicMock(spec=AgentsConfig)
|
||
cfg.projects = {}
|
||
cfg.claude = ClaudeConfig(model="claude-sonnet-4-6", api_key=None)
|
||
cfg.scheduler = SchedulerConfig()
|
||
cfg.feishu = FeishuConfig()
|
||
cfg.claude_wx_url = ""
|
||
return cfg
|
||
|
||
|
||
@pytest.mark.parametrize("role,expected_class", [
|
||
("os-engineer", "OsEngineerAgent"),
|
||
("algo-researcher", "AlgoResearcherAgent"),
|
||
("base-validator", "BaseValidatorAgent"),
|
||
("base-architect", "BaseArchitectAgent"),
|
||
("system-tester", "SystemTesterAgent"),
|
||
("vision-analyst", "VisionAnalystAgent"),
|
||
("market-pm", "MarketPmAgent"),
|
||
("media-producer", "MediaProducerAgent"),
|
||
])
|
||
def test_executor_routes_new_roles(role, expected_class, mock_config, tmp_path):
|
||
from nmfs_agents.core.executor import _make_agent
|
||
from nmfs_agents.core.queue import Task
|
||
from nmfs_agents.tools.project_memory import ProjectMemory
|
||
|
||
task = Task(project="test", type="test", title="t",
|
||
context="", priority=3, mode="auto", agent_role=role)
|
||
mem = ProjectMemory(db_path=tmp_path / "m.db")
|
||
agent = _make_agent(mock_config, task, mem)
|
||
assert type(agent).__name__ == expected_class, \
|
||
f"role={role} 路由到 {type(agent).__name__}，期望 {expected_class}"
|
||
|
||
|
||
def test_executor_publishes_event_on_done(tmp_path):
|
||
import asyncio
|
||
import fakeredis
|
||
from unittest.mock import MagicMock, AsyncMock
|
||
from nmfs_agents.tools.event_bus import EventBus
|
||
from nmfs_agents.core.queue import Task, TaskQueue
|
||
from nmfs_agents.core.executor import Executor
|
||
from nmfs_agents.agents.developer import AgentResult
|
||
from nmfs_agents.config import AgentsConfig, SchedulerConfig, ClaudeConfig, FeishuConfig
|
||
import nmfs_agents.core.executor as _executor_mod
|
||
|
||
cfg = AgentsConfig(
|
||
projects={},
|
||
scheduler=SchedulerConfig(max_concurrent=2),
|
||
claude=ClaudeConfig(api_key="x"),
|
||
feishu=FeishuConfig(),
|
||
devices={},
|
||
)
|
||
|
||
r = fakeredis.FakeRedis(decode_responses=True)
|
||
bus = EventBus(redis_client=r)
|
||
|
||
q = TaskQueue(db_path=tmp_path / "t.db")
|
||
task = Task(project="antishake", type="code_improve", title="EIS 迭代",
|
||
priority=3, mode="auto", agent_role="algo-antishake")
|
||
q.enqueue(task)
|
||
|
||
executor = Executor(cfg, queue=q, event_bus=bus)
|
||
|
||
mock_agent = MagicMock()
|
||
mock_agent.run = MagicMock(return_value=AgentResult(status="done", summary="EIS 防抖达标"))
|
||
orig = _executor_mod._make_agent
|
||
_executor_mod._make_agent = lambda *a, **kw: mock_agent
|
||
try:
|
||
asyncio.run(executor.run_all_pending())
|
||
finally:
|
||
_executor_mod._make_agent = orig
|
||
|
||
events = bus.read_new("ops-hub", count=10, block_ms=100)
|
||
assert len(events) == 1
|
||
assert events[0].event_type == "task_done"
|
||
assert events[0].project == "antishake"
|