nmfs-agents/tests/test_developer.py

from __future__ import annotations

import json
from pathlib import Path
from unittest.mock import patch, MagicMock

from nmfs_agents.agents.developer import DeveloperAgent
from nmfs_agents.config import AgentsConfig, ProjectConfig, SchedulerConfig, ClaudeConfig, FeishuConfig
from nmfs_agents.core.queue import Task


def _cfg(tmp_path: Path) -> AgentsConfig:
    return AgentsConfig(
        projects={"yolo": ProjectConfig(path=tmp_path / "yolo", mode="auto")},
        scheduler=SchedulerConfig(),
        claude=ClaudeConfig(api_key="fake-key"),
        feishu=FeishuConfig(),
        devices={},
    )


def test_developer_report_mode_uses_readonly_tools(tmp_path):
    """report 模式下 CLI 命令应只包含 Read/Glob/Grep 工具，不含 dangerously-skip-permissions。"""
    from nmfs_agents.agents.developer import AgentResult
    task = Task(project="yolo", type="fix_bug", title="分析零检出",
                priority=1, mode="report", agent_role="developer", id=1)
    cfg = _cfg(tmp_path)
    agent = DeveloperAgent(cfg)

    with patch("nmfs_agents.agents.developer._run_with_log",
               return_value=AgentResult(status="done", summary="分析完成：根本原因是 sigmoid 缺失")) as mock_run:
        result = agent.run(task)

    assert result.status == "done"
    assert "分析" in result.summary

    call_cmd = mock_run.call_args[0][0]  # first positional arg is cmd list
    assert "--allowedTools" in call_cmd
    idx = call_cmd.index("--allowedTools")
    assert "Read,Glob,Grep" in call_cmd[idx + 1]
    assert "--dangerously-skip-permissions" not in call_cmd


def test_developer_auto_mode_returns_summary(tmp_path):
    """auto/confirm 模式使用 --dangerously-skip-permissions（由 hooks 实施逐工具审批），返回摘要。"""
    from nmfs_agents.agents.developer import AgentResult
    (tmp_path / "yolo").mkdir()
    task = Task(project="yolo", type="code_review", title="检查代码质量",
                priority=3, mode="auto", agent_role="developer", id=2)
    cfg = _cfg(tmp_path)
    agent = DeveloperAgent(cfg)

    with patch("nmfs_agents.agents.developer._run_with_log",
               return_value=AgentResult(status="done", summary="代码审查完成，发现 2 处改进点")) as mock_run:
        result = agent.run(task)

    assert result.status == "done"
    assert result.summary != ""

    call_cmd = mock_run.call_args[0][0]
    assert "--dangerously-skip-permissions" in call_cmd


def test_developer_context_uses_description_not_hardcoded(tmp_path):
    """_build_context must not mention 'Rockchip RKNN' when description differs."""
    cfg = AgentsConfig(
        projects={"myproj": ProjectConfig(
            path=tmp_path, mode="report",
            description="通用 Web 后端服务"
        )},
        scheduler=SchedulerConfig(),
        claude=ClaudeConfig(model="claude-sonnet-4-6"),
        feishu=FeishuConfig(),
        devices={},
    )
    agent = DeveloperAgent(cfg)
    task = Task(project="myproj", type="fix_bug", title="修复登录",
                context="", priority=2, mode="report", agent_role="developer")
    ctx = agent._build_context(task)
    assert "通用 Web 后端服务" in ctx
    assert "Rockchip RKNN" not in ctx
    assert "myproj" in ctx


def test_developer_context_includes_memory(tmp_path):
    """_build_context includes ProjectMemory context when available."""
    from nmfs_agents.tools.project_memory import ProjectMemory

    mem = ProjectMemory(db_path=tmp_path / "mem.db")
    mem.set_goal("myproj", "稳定高性能后端", role="productizer")
    mem.set_fact("myproj", "test_cmd", "pytest tests/", role="developer")

    cfg = AgentsConfig(
        projects={"myproj": ProjectConfig(path=tmp_path, mode="auto",
                                          description="通用 Web 后端服务")},
        scheduler=SchedulerConfig(),
        claude=ClaudeConfig(model="claude-sonnet-4-6"),
        feishu=FeishuConfig(),
        devices={},
    )
    agent = DeveloperAgent(cfg, memory=mem)
    task = Task(project="myproj", type="fix_bug", title="修复登录",
                context="", priority=2, mode="auto", agent_role="developer")
    ctx = agent._build_context(task)
    assert "稳定高性能后端" in ctx
    assert "test_cmd" in ctx


def test_auto_mode_has_dangerously_skip_permissions(tmp_path):
    """auto/confirm 模式包含 --dangerously-skip-permissions，由 hooks 实施逐工具审批"""
    task = Task(project="yolo", type="fix_bug", title="优化精度",
                priority=1, mode="auto", agent_role="developer", id=5)
    cfg = _cfg(tmp_path)
    agent = DeveloperAgent(cfg)
    cmd, _ = agent._build_cmd(task)
    assert "--dangerously-skip-permissions" in cmd


def test_auto_mode_sets_trust_env(tmp_path):
    """auto 模式注入 CLAUDE_WX_AGENT_MODE=trust"""
    import json as _json
    task = Task(project="yolo", type="fix_bug", title="优化精度",
                priority=1, mode="auto", agent_role="developer", id=5)
    cfg = _cfg(tmp_path)
    agent = DeveloperAgent(cfg)
    env = agent._build_env(task)
    assert env.get("CLAUDE_WX_AGENT_MODE") == "trust"
    agent_ctx = _json.loads(env.get("CLAUDE_WX_AGENT", "{}"))
    assert agent_ctx["project"] == "yolo"
    assert agent_ctx["title"] == "优化精度"


def test_confirm_mode_sets_default_env(tmp_path):
    """confirm 模式注入 CLAUDE_WX_AGENT_MODE=default"""
    task = Task(project="yolo", type="fix_bug", title="修复零检出",
                priority=1, mode="confirm", agent_role="developer", id=6)
    cfg = _cfg(tmp_path)
    agent = DeveloperAgent(cfg)
    env = agent._build_env(task)
    assert env.get("CLAUDE_WX_AGENT_MODE") == "default"


def test_confirm_mode_no_allowed_tools_restriction(tmp_path):
    """confirm 模式不限制 --allowedTools（由 hooks 控制），包含 dangerously-skip-permissions"""
    task = Task(project="yolo", type="fix_bug", title="修复零检出",
                priority=1, mode="confirm", agent_role="developer", id=6)
    cfg = _cfg(tmp_path)
    agent = DeveloperAgent(cfg)
    cmd, _ = agent._build_cmd(task)
    assert "--allowedTools" not in cmd
    assert "--dangerously-skip-permissions" in cmd


def test_unknown_mode_env_falls_back_to_trust(tmp_path):
    """未知 mode 时 CLAUDE_WX_AGENT_MODE 回退为 trust（有意设计）"""
    task = Task(project="yolo", type="fix_bug", title="测试",
                priority=1, mode="unknown_mode", agent_role="developer", id=99)
    cfg = _cfg(tmp_path)
    agent = DeveloperAgent(cfg)
    env = agent._build_env(task)
    assert env.get("CLAUDE_WX_AGENT_MODE") == "trust"


def test_build_env_without_task_no_agent_vars(tmp_path):
    """_build_env(task=None) 时不注入 CLAUDE_WX_AGENT_MODE/CLAUDE_WX_AGENT"""
    cfg = _cfg(tmp_path)
    agent = DeveloperAgent(cfg)
    env = agent._build_env()
    assert "CLAUDE_WX_AGENT_MODE" not in env
    assert "CLAUDE_WX_AGENT" not in env


def test_mcp_sequential_thinking_injected(tmp_path):
    """dev-embedded 角色应注入 MCP 配置（--mcp-config 临时文件）"""
    from nmfs_agents.core.queue import Task
    agent = DeveloperAgent(_cfg(tmp_path))
    task = Task(project="yolo", type="fix_bug", title="修复崩溃",
                priority=1, mode="auto", agent_role="dev-embedded")
    cmd, tmp_files = agent._build_cmd(task)
    cmd_str = " ".join(cmd)
    # MCP 通过 --mcp-config 临时 JSON 文件注入，不是 --mcp-server 参数
    assert "--mcp-config" in cmd_str
    # 清理临时文件
    import os
    for f in tmp_files:
        try:
            os.unlink(f)
        except OSError:
            pass


def test_mcp_playwright_injected_for_tester(tmp_path):
    """tester 角色应通过 --mcp-config 注入 playwright"""
    from nmfs_agents.core.queue import Task
    import os
    agent = DeveloperAgent(_cfg(tmp_path))
    task = Task(project="yolo", type="fix_bug", title="测试",
                priority=1, mode="auto", agent_role="tester")
    cmd, tmp_files = agent._build_cmd(task)
    cmd_str = " ".join(cmd)
    assert "--mcp-config" in cmd_str
    for f in tmp_files:
        try:
            os.unlink(f)
        except OSError:
            pass


def test_report_mode_no_mcp(tmp_path):
    """report 模式仍然限制 allowedTools，不注入 MCP 配置"""
    from nmfs_agents.core.queue import Task
    agent = DeveloperAgent(_cfg(tmp_path))
    task = Task(project="yolo", type="fix_bug", title="分析",
                priority=1, mode="report", agent_role="dev-embedded")
    cmd, _ = agent._build_cmd(task)
    assert "--allowedTools" in cmd
    assert "--mcp-config" not in " ".join(cmd)


def test_build_device_context_with_devices(tmp_path):
    """_build_device_context 应包含 SSH 连接示例和设备信息"""
    from nmfs_agents.config import DeviceConfig
    cfg = AgentsConfig(
        projects={"yolo": ProjectConfig(path=tmp_path / "yolo", mode="auto")},
        scheduler=SchedulerConfig(),
        claude=ClaudeConfig(api_key="fake"),
        feishu=FeishuConfig(),
        devices={"rk3588": DeviceConfig(
            host="192.168.1.1", user="pi", password="pi",
            workspace="/home/pi/work",
        )},
    )
    agent = DeveloperAgent(cfg)
    ctx = agent._build_device_context()
    assert "SSH" in ctx
    assert "192.168.1.1" in ctx
    assert "sshpass" in ctx


def test_build_device_context_empty_when_no_devices(tmp_path):
    """无设备配置时 _build_device_context 返回空字符串"""
    agent = DeveloperAgent(_cfg(tmp_path))
    assert agent._build_device_context() == ""


def test_plan_first_prompt_in_context(tmp_path):
    """非 report 模式下 system context 应包含 sequentialthinking"""
    from nmfs_agents.core.queue import Task
    agent = DeveloperAgent(_cfg(tmp_path))
    task = Task(project="yolo", type="fix_bug", title="修复",
                priority=1, mode="auto", agent_role="dev-embedded")
    ctx = agent._build_context(task)
    assert "sequentialthinking" in ctx


def test_report_mode_no_plan_first(tmp_path):
    """report 模式不应包含 Plan-First 提示"""
    from nmfs_agents.core.queue import Task
    agent = DeveloperAgent(_cfg(tmp_path))
    task = Task(project="yolo", type="fix_bug", title="分析",
                priority=1, mode="report", agent_role="dev-embedded")
    ctx = agent._build_context(task)
    assert "sequentialthinking" not in ctx


def test_build_device_context_includes_edge_service_for_linux(tmp_path):
    """Linux 设备有 EdgeValidatorService 时应在 context 中包含 HTTP URL"""
    from nmfs_agents.config import DeviceConfig
    cfg = AgentsConfig(
        projects={},
        scheduler=SchedulerConfig(),
        claude=ClaudeConfig(api_key="fake"),
        feishu=FeishuConfig(),
        devices={"rk3588": DeviceConfig(
            host="192.168.1.1", user="pi", password="pi",
            workspace="/home/pi", type="linux", connect="ssh",
        )},
    )
    agent = DeveloperAgent(cfg)
    ctx = agent._build_device_context()
    assert "EdgeValidatorService" in ctx or "edge_validator" in ctx.lower() or "8899" in ctx
    assert "边端落地" in ctx


def test_build_device_context_includes_serial_info(tmp_path):
    """ESP32 串口设备应在 context 中包含 pyserial 和端口路径"""
    from nmfs_agents.config import DeviceConfig
    cfg = AgentsConfig(
        projects={},
        scheduler=SchedulerConfig(),
        claude=ClaudeConfig(api_key="fake"),
        feishu=FeishuConfig(),
        devices={"esp32-dev": DeviceConfig(
            type="esp32", connect="serial", port="/dev/ttyUSB0", baud=115200,
            host="", user="", password="", workspace="",
        )},
    )
    agent = DeveloperAgent(cfg)
    ctx = agent._build_device_context()
    assert "/dev/ttyUSB0" in ctx
    assert "pyserial" in ctx or "serial" in ctx.lower()


def test_build_device_context_injects_env_models(tmp_path):
    """_build_device_context 在 __env__ 有数据时应注入已知模型摘要。"""
    from nmfs_agents.config import DeviceConfig
    from nmfs_agents.tools.project_memory import ProjectMemory

    cfg = AgentsConfig(
        projects={},
        scheduler=SchedulerConfig(),
        claude=ClaudeConfig(api_key="fake"),
        feishu=FeishuConfig(),
        devices={"rk3588": DeviceConfig(
            host="192.168.1.1", user="pi", password="pi",
            workspace="/home/pi", type="linux", connect="ssh",
        )},
    )
    mem = ProjectMemory(db_path=tmp_path / "m.db")
    mem.set_fact("__env__", "local.models.rknn", "/data/a.rknn,/data/b.rknn", role="env-collector")
    mem.set_fact("__env__", "rk3588.models.rknn", "/home/pi/model.rknn", role="env-collector")
    mem.set_fact("__env__", "rk3588.disk", "/dev/root 29G 8G 19G", role="env-collector")
    mem.set_fact("__env__", "collected_at", "2026-03-09T10:00:00", role="env-collector")

    agent = DeveloperAgent(cfg, memory=mem)
    ctx = agent._build_device_context()

    assert "已知模型" in ctx
    assert "/data/a.rknn" in ctx
    assert "rk3588" in ctx
    assert "19G" in ctx or "29G" in ctx


def test_build_device_context_no_env_data(tmp_path):
    """__env__ 无数据时，_build_device_context 不注入模型摘要（不崩溃）。"""
    from nmfs_agents.config import DeviceConfig
    from nmfs_agents.tools.project_memory import ProjectMemory

    cfg = AgentsConfig(
        projects={},
        scheduler=SchedulerConfig(),
        claude=ClaudeConfig(api_key="fake"),
        feishu=FeishuConfig(),
        devices={"rk3588": DeviceConfig(
            host="192.168.1.1", user="pi", password="pi",
            workspace="/home/pi", type="linux", connect="ssh",
        )},
    )
    mem = ProjectMemory(db_path=tmp_path / "m.db")
    agent = DeveloperAgent(cfg, memory=mem)
    ctx = agent._build_device_context()
    assert "已知模型" not in ctx