wanwu/deer-flow
1
0
Fork 0
mirror of https://gitee.com/wanwujie/deer-flow synced 2026-04-08 16:24:45 +08:00
Code Issues Packages Projects Releases Wiki Activity
Files
0c7c96d75e286c7513ea1d0cb9625afe6617dd21
deer-flow/backend/tests/test_custom_agent.py
JeffJiang 7de94394d4 feat(agent):Supports custom agent and chat experience with refactoring (#957) 
* feat: add agent management functionality with creation, editing, and deletion

* feat: enhance agent creation and chat experience

- Added AgentWelcome component to display agent description on new thread creation.
- Improved agent name validation with availability check during agent creation.
- Updated NewAgentPage to handle agent creation flow more effectively, including enhanced error handling and user feedback.
- Refactored chat components to streamline message handling and improve user experience.
- Introduced new bootstrap skill for personalized onboarding conversations, including detailed conversation phases and a structured SOUL.md template.
- Updated localization files to reflect new features and error messages.
- General code cleanup and optimizations across various components and hooks.

* Refactor workspace layout and agent management components

- Updated WorkspaceLayout to use useLayoutEffect for sidebar state initialization.
- Removed unused AgentFormDialog and related edit functionality from AgentCard.
- Introduced ArtifactTrigger component to manage artifact visibility.
- Enhanced ChatBox to handle artifact selection and display.
- Improved message list rendering logic to avoid loading states.
- Updated localization files to remove deprecated keys and add new translations.
- Refined hooks for local settings and thread management to improve performance and clarity.
- Added temporal awareness guidelines to deep research skill documentation.

* feat: refactor chat components and introduce thread management hooks

* feat: improve artifact file detail preview logic and clean up console logs

* feat: refactor lead agent creation logic and improve logging details

* feat: validate agent name format and enhance error handling in agent setup

* feat: simplify thread search query by removing unnecessary metadata

* feat: update query key in useDeleteThread and useRenameThread for consistency

* feat: add isMock parameter to thread and artifact handling for improved testing

* fix: reorder import of setup_agent for consistency in builtins module

* feat: append mock parameter to thread links in CaseStudySection for testing purposes

* fix: update load_agent_soul calls to use cfg.name for improved clarity

* fix: update date format in apply_prompt_template for consistency

* feat: integrate isMock parameter into artifact content loading for enhanced testing

* docs: add license section to SKILL.md for clarity and attribution

* feat(agent): enhance model resolution and agent configuration handling

* chore: remove unused import of _resolve_model_name from agents

* feat(agent): remove unused field

* fix(agent): set default value for requested_model_name in _resolve_model_name function

* feat(agent): update get_available_tools call to handle optional agent_config and improve middleware function signature

---------

Co-authored-by: Willem Jiang <willem.jiang@gmail.com>
2026-03-03 21:32:01 +08:00

516 lines
20 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""Tests for custom agent support."""
from __future__ import annotations
from pathlib import Path
from unittest.mock import patch
import pytest
import yaml
from fastapi.testclient import TestClient
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _make_paths(base_dir: Path):
"""Return a Paths instance pointing to base_dir."""
from src.config.paths import Paths
return Paths(base_dir=base_dir)
def _write_agent(base_dir: Path, name: str, config: dict, soul: str = "You are helpful.") -> None:
"""Write an agent directory with config.yaml and SOUL.md."""
agent_dir = base_dir / "agents" / name
agent_dir.mkdir(parents=True, exist_ok=True)
config_copy = dict(config)
if "name" not in config_copy:
config_copy["name"] = name
with open(agent_dir / "config.yaml", "w") as f:
yaml.dump(config_copy, f)
(agent_dir / "SOUL.md").write_text(soul, encoding="utf-8")
# ===========================================================================
# 1. Paths class – agent path methods
# ===========================================================================
class TestPaths:
def test_agents_dir(self, tmp_path):
paths = _make_paths(tmp_path)
assert paths.agents_dir == tmp_path / "agents"
def test_agent_dir(self, tmp_path):
paths = _make_paths(tmp_path)
assert paths.agent_dir("code-reviewer") == tmp_path / "agents" / "code-reviewer"
def test_agent_memory_file(self, tmp_path):
paths = _make_paths(tmp_path)
assert paths.agent_memory_file("code-reviewer") == tmp_path / "agents" / "code-reviewer" / "memory.json"
def test_user_md_file(self, tmp_path):
paths = _make_paths(tmp_path)
assert paths.user_md_file == tmp_path / "USER.md"
def test_paths_are_different_from_global(self, tmp_path):
paths = _make_paths(tmp_path)
assert paths.memory_file != paths.agent_memory_file("my-agent")
assert paths.memory_file == tmp_path / "memory.json"
assert paths.agent_memory_file("my-agent") == tmp_path / "agents" / "my-agent" / "memory.json"
# ===========================================================================
# 2. AgentConfig – Pydantic parsing
# ===========================================================================
class TestAgentConfig:
def test_minimal_config(self):
from src.config.agents_config import AgentConfig
cfg = AgentConfig(name="my-agent")
assert cfg.name == "my-agent"
assert cfg.description == ""
assert cfg.model is None
assert cfg.tool_groups is None
def test_full_config(self):
from src.config.agents_config import AgentConfig
cfg = AgentConfig(
name="code-reviewer",
description="Specialized for code review",
model="deepseek-v3",
tool_groups=["file:read", "bash"],
)
assert cfg.name == "code-reviewer"
assert cfg.model == "deepseek-v3"
assert cfg.tool_groups == ["file:read", "bash"]
def test_config_from_dict(self):
from src.config.agents_config import AgentConfig
data = {"name": "test-agent", "description": "A test", "model": "gpt-4"}
cfg = AgentConfig(**data)
assert cfg.name == "test-agent"
assert cfg.model == "gpt-4"
assert cfg.tool_groups is None
# ===========================================================================
# 3. load_agent_config
# ===========================================================================
class TestLoadAgentConfig:
def test_load_valid_config(self, tmp_path):
config_dict = {"name": "code-reviewer", "description": "Code review agent", "model": "deepseek-v3"}
_write_agent(tmp_path, "code-reviewer", config_dict)
with patch("src.config.agents_config.get_paths", return_value=_make_paths(tmp_path)):
from src.config.agents_config import load_agent_config
cfg = load_agent_config("code-reviewer")
assert cfg.name == "code-reviewer"
assert cfg.description == "Code review agent"
assert cfg.model == "deepseek-v3"
def test_load_missing_agent_raises(self, tmp_path):
with patch("src.config.agents_config.get_paths", return_value=_make_paths(tmp_path)):
from src.config.agents_config import load_agent_config
with pytest.raises(FileNotFoundError):
load_agent_config("nonexistent-agent")
def test_load_missing_config_yaml_raises(self, tmp_path):
# Create directory without config.yaml
(tmp_path / "agents" / "broken-agent").mkdir(parents=True)
with patch("src.config.agents_config.get_paths", return_value=_make_paths(tmp_path)):
from src.config.agents_config import load_agent_config
with pytest.raises(FileNotFoundError):
load_agent_config("broken-agent")
def test_load_config_infers_name_from_dir(self, tmp_path):
"""Config without 'name' field should use directory name."""
agent_dir = tmp_path / "agents" / "inferred-name"
agent_dir.mkdir(parents=True)
(agent_dir / "config.yaml").write_text("description: My agent\n")
(agent_dir / "SOUL.md").write_text("Hello")
with patch("src.config.agents_config.get_paths", return_value=_make_paths(tmp_path)):
from src.config.agents_config import load_agent_config
cfg = load_agent_config("inferred-name")
assert cfg.name == "inferred-name"
def test_load_config_with_tool_groups(self, tmp_path):
config_dict = {"name": "restricted", "tool_groups": ["file:read", "file:write"]}
_write_agent(tmp_path, "restricted", config_dict)
with patch("src.config.agents_config.get_paths", return_value=_make_paths(tmp_path)):
from src.config.agents_config import load_agent_config
cfg = load_agent_config("restricted")
assert cfg.tool_groups == ["file:read", "file:write"]
def test_legacy_prompt_file_field_ignored(self, tmp_path):
"""Unknown fields like the old prompt_file should be silently ignored."""
agent_dir = tmp_path / "agents" / "legacy-agent"
agent_dir.mkdir(parents=True)
(agent_dir / "config.yaml").write_text("name: legacy-agent\nprompt_file: system.md\n")
(agent_dir / "SOUL.md").write_text("Soul content")
with patch("src.config.agents_config.get_paths", return_value=_make_paths(tmp_path)):
from src.config.agents_config import load_agent_config
cfg = load_agent_config("legacy-agent")
assert cfg.name == "legacy-agent"
# ===========================================================================
# 4. load_agent_soul
# ===========================================================================
class TestLoadAgentSoul:
def test_reads_soul_file(self, tmp_path):
expected_soul = "You are a specialized code review expert."
_write_agent(tmp_path, "code-reviewer", {"name": "code-reviewer"}, soul=expected_soul)
with patch("src.config.agents_config.get_paths", return_value=_make_paths(tmp_path)):
from src.config.agents_config import AgentConfig, load_agent_soul
cfg = AgentConfig(name="code-reviewer")
soul = load_agent_soul(cfg.name)
assert soul == expected_soul
def test_missing_soul_file_returns_none(self, tmp_path):
agent_dir = tmp_path / "agents" / "no-soul"
agent_dir.mkdir(parents=True)
(agent_dir / "config.yaml").write_text("name: no-soul\n")
# No SOUL.md created
with patch("src.config.agents_config.get_paths", return_value=_make_paths(tmp_path)):
from src.config.agents_config import AgentConfig, load_agent_soul
cfg = AgentConfig(name="no-soul")
soul = load_agent_soul(cfg.name)
assert soul is None
def test_empty_soul_file_returns_none(self, tmp_path):
agent_dir = tmp_path / "agents" / "empty-soul"
agent_dir.mkdir(parents=True)
(agent_dir / "config.yaml").write_text("name: empty-soul\n")
(agent_dir / "SOUL.md").write_text(" \n ")
with patch("src.config.agents_config.get_paths", return_value=_make_paths(tmp_path)):
from src.config.agents_config import AgentConfig, load_agent_soul
cfg = AgentConfig(name="empty-soul")
soul = load_agent_soul(cfg.name)
assert soul is None
# ===========================================================================
# 5. list_custom_agents
# ===========================================================================
class TestListCustomAgents:
def test_empty_when_no_agents_dir(self, tmp_path):
with patch("src.config.agents_config.get_paths", return_value=_make_paths(tmp_path)):
from src.config.agents_config import list_custom_agents
agents = list_custom_agents()
assert agents == []
def test_discovers_multiple_agents(self, tmp_path):
_write_agent(tmp_path, "agent-a", {"name": "agent-a"})
_write_agent(tmp_path, "agent-b", {"name": "agent-b", "description": "B"})
with patch("src.config.agents_config.get_paths", return_value=_make_paths(tmp_path)):
from src.config.agents_config import list_custom_agents
agents = list_custom_agents()
names = [a.name for a in agents]
assert "agent-a" in names
assert "agent-b" in names
def test_skips_dirs_without_config_yaml(self, tmp_path):
# Valid agent
_write_agent(tmp_path, "valid-agent", {"name": "valid-agent"})
# Invalid dir (no config.yaml)
(tmp_path / "agents" / "invalid-dir").mkdir(parents=True)
with patch("src.config.agents_config.get_paths", return_value=_make_paths(tmp_path)):
from src.config.agents_config import list_custom_agents
agents = list_custom_agents()
assert len(agents) == 1
assert agents[0].name == "valid-agent"
def test_skips_non_directory_entries(self, tmp_path):
# Create the agents dir with a file (not a dir)
agents_dir = tmp_path / "agents"
agents_dir.mkdir(parents=True)
(agents_dir / "not-a-dir.txt").write_text("hello")
_write_agent(tmp_path, "real-agent", {"name": "real-agent"})
with patch("src.config.agents_config.get_paths", return_value=_make_paths(tmp_path)):
from src.config.agents_config import list_custom_agents
agents = list_custom_agents()
assert len(agents) == 1
assert agents[0].name == "real-agent"
def test_returns_sorted_by_name(self, tmp_path):
_write_agent(tmp_path, "z-agent", {"name": "z-agent"})
_write_agent(tmp_path, "a-agent", {"name": "a-agent"})
_write_agent(tmp_path, "m-agent", {"name": "m-agent"})
with patch("src.config.agents_config.get_paths", return_value=_make_paths(tmp_path)):
from src.config.agents_config import list_custom_agents
agents = list_custom_agents()
names = [a.name for a in agents]
assert names == sorted(names)
# ===========================================================================
# 7. Memory isolation: _get_memory_file_path
# ===========================================================================
class TestMemoryFilePath:
def test_global_memory_path(self, tmp_path):
"""None agent_name should return global memory file."""
import src.agents.memory.updater as updater_mod
with patch("src.agents.memory.updater.get_paths", return_value=_make_paths(tmp_path)):
path = updater_mod._get_memory_file_path(None)
assert path == tmp_path / "memory.json"
def test_agent_memory_path(self, tmp_path):
"""Providing agent_name should return per-agent memory file."""
import src.agents.memory.updater as updater_mod
with patch("src.agents.memory.updater.get_paths", return_value=_make_paths(tmp_path)):
path = updater_mod._get_memory_file_path("code-reviewer")
assert path == tmp_path / "agents" / "code-reviewer" / "memory.json"
def test_different_paths_for_different_agents(self, tmp_path):
import src.agents.memory.updater as updater_mod
with patch("src.agents.memory.updater.get_paths", return_value=_make_paths(tmp_path)):
path_global = updater_mod._get_memory_file_path(None)
path_a = updater_mod._get_memory_file_path("agent-a")
path_b = updater_mod._get_memory_file_path("agent-b")
assert path_global != path_a
assert path_global != path_b
assert path_a != path_b
# ===========================================================================
# 8. Gateway API – Agents endpoints
# ===========================================================================
def _make_test_app(tmp_path: Path):
"""Create a FastAPI app with the agents router, patching paths to tmp_path."""
from fastapi import FastAPI
from src.gateway.routers.agents import router
app = FastAPI()
app.include_router(router)
return app
@pytest.fixture()
def agent_client(tmp_path):
"""TestClient with agents router, using tmp_path as base_dir."""
paths_instance = _make_paths(tmp_path)
with patch("src.config.agents_config.get_paths", return_value=paths_instance), patch("src.gateway.routers.agents.get_paths", return_value=paths_instance):
app = _make_test_app(tmp_path)
with TestClient(app) as client:
client._tmp_path = tmp_path # type: ignore[attr-defined]
yield client
class TestAgentsAPI:
def test_list_agents_empty(self, agent_client):
response = agent_client.get("/api/agents")
assert response.status_code == 200
data = response.json()
assert data["agents"] == []
def test_create_agent(self, agent_client):
payload = {
"name": "code-reviewer",
"description": "Reviews code",
"soul": "You are a code reviewer.",
}
response = agent_client.post("/api/agents", json=payload)
assert response.status_code == 201
data = response.json()
assert data["name"] == "code-reviewer"
assert data["description"] == "Reviews code"
assert data["soul"] == "You are a code reviewer."
def test_create_agent_invalid_name(self, agent_client):
payload = {"name": "Code Reviewer!", "soul": "test"}
response = agent_client.post("/api/agents", json=payload)
assert response.status_code == 422
def test_create_duplicate_agent_409(self, agent_client):
payload = {"name": "my-agent", "soul": "test"}
agent_client.post("/api/agents", json=payload)
# Second create should fail
response = agent_client.post("/api/agents", json=payload)
assert response.status_code == 409
def test_list_agents_after_create(self, agent_client):
agent_client.post("/api/agents", json={"name": "agent-one", "soul": "p1"})
agent_client.post("/api/agents", json={"name": "agent-two", "soul": "p2"})
response = agent_client.get("/api/agents")
assert response.status_code == 200
names = [a["name"] for a in response.json()["agents"]]
assert "agent-one" in names
assert "agent-two" in names
def test_get_agent(self, agent_client):
agent_client.post("/api/agents", json={"name": "test-agent", "soul": "Hello world"})
response = agent_client.get("/api/agents/test-agent")
assert response.status_code == 200
data = response.json()
assert data["name"] == "test-agent"
assert data["soul"] == "Hello world"
def test_get_missing_agent_404(self, agent_client):
response = agent_client.get("/api/agents/nonexistent")
assert response.status_code == 404
def test_update_agent_soul(self, agent_client):
agent_client.post("/api/agents", json={"name": "update-me", "soul": "original"})
response = agent_client.put("/api/agents/update-me", json={"soul": "updated"})
assert response.status_code == 200
assert response.json()["soul"] == "updated"
def test_update_agent_description(self, agent_client):
agent_client.post("/api/agents", json={"name": "desc-agent", "description": "old desc", "soul": "p"})
response = agent_client.put("/api/agents/desc-agent", json={"description": "new desc"})
assert response.status_code == 200
assert response.json()["description"] == "new desc"
def test_update_missing_agent_404(self, agent_client):
response = agent_client.put("/api/agents/ghost-agent", json={"soul": "new"})
assert response.status_code == 404
def test_delete_agent(self, agent_client):
agent_client.post("/api/agents", json={"name": "del-me", "soul": "bye"})
response = agent_client.delete("/api/agents/del-me")
assert response.status_code == 204
# Verify it's gone
response = agent_client.get("/api/agents/del-me")
assert response.status_code == 404
def test_delete_missing_agent_404(self, agent_client):
response = agent_client.delete("/api/agents/does-not-exist")
assert response.status_code == 404
def test_create_agent_with_model_and_tool_groups(self, agent_client):
payload = {
"name": "specialized",
"description": "Specialized agent",
"model": "deepseek-v3",
"tool_groups": ["file:read", "bash"],
"soul": "You are specialized.",
}
response = agent_client.post("/api/agents", json=payload)
assert response.status_code == 201
data = response.json()
assert data["model"] == "deepseek-v3"
assert data["tool_groups"] == ["file:read", "bash"]
def test_create_persists_files_on_disk(self, agent_client, tmp_path):
agent_client.post("/api/agents", json={"name": "disk-check", "soul": "disk soul"})
agent_dir = tmp_path / "agents" / "disk-check"
assert agent_dir.exists()
assert (agent_dir / "config.yaml").exists()
assert (agent_dir / "SOUL.md").exists()
assert (agent_dir / "SOUL.md").read_text() == "disk soul"
def test_delete_removes_files_from_disk(self, agent_client, tmp_path):
agent_client.post("/api/agents", json={"name": "remove-me", "soul": "bye"})
agent_dir = tmp_path / "agents" / "remove-me"
assert agent_dir.exists()
agent_client.delete("/api/agents/remove-me")
assert not agent_dir.exists()
# ===========================================================================
# 9. Gateway API – User Profile endpoints
# ===========================================================================
class TestUserProfileAPI:
def test_get_user_profile_empty(self, agent_client):
response = agent_client.get("/api/user-profile")
assert response.status_code == 200
assert response.json()["content"] is None
def test_put_user_profile(self, agent_client, tmp_path):
content = "# User Profile\n\nI am a developer."
response = agent_client.put("/api/user-profile", json={"content": content})
assert response.status_code == 200
assert response.json()["content"] == content
# File should be written to disk
user_md = tmp_path / "USER.md"
assert user_md.exists()
assert user_md.read_text(encoding="utf-8") == content
def test_get_user_profile_after_put(self, agent_client):
content = "# Profile\n\nI work on data science."
agent_client.put("/api/user-profile", json={"content": content})
response = agent_client.get("/api/user-profile")
assert response.status_code == 200
assert response.json()["content"] == content
def test_put_empty_user_profile_returns_none(self, agent_client):
response = agent_client.put("/api/user-profile", json={"content": ""})
assert response.status_code == 200
assert response.json()["content"] is None
Reference in New Issue View Git Blame Copy Permalink