123 lines
4.4 KiB
Python
123 lines
4.4 KiB
Python
from __future__ import annotations
|
|
|
|
import pytest
|
|
import yaml
|
|
|
|
from app.gateway.routers.agents import _load_config_extras, _write_agent_config
|
|
from app.gateway.services import build_run_config
|
|
from deerflow.config import paths as paths_module
|
|
from deerflow.persistence.agents import MemoryAgentStore
|
|
|
|
|
|
@pytest.fixture
|
|
def isolated_paths(tmp_path, monkeypatch):
|
|
"""Redirect get_paths() at a fresh temp directory for the duration of one test."""
|
|
monkeypatch.setenv("DEER_FLOW_HOME", str(tmp_path))
|
|
monkeypatch.setattr(paths_module, "_paths", None)
|
|
yield tmp_path
|
|
monkeypatch.setattr(paths_module, "_paths", None)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_agent_store_allows_duplicate_chinese_names_and_filters_visibility() -> None:
|
|
store = MemoryAgentStore()
|
|
chinese_name = "\u4e2d\u6587\u52a9\u624b"
|
|
|
|
await store.ensure_builtin({"id": "builtin-agent", "name": chinese_name, "description": "built in"})
|
|
await store.create({"id": "private-a", "name": chinese_name, "description": "private", "user_id": "user-a"})
|
|
await store.create({"id": "published-b", "name": chinese_name, "description": "published", "user_id": "user-b", "published": True})
|
|
|
|
user_a_ids = {agent["id"] for agent in await store.list_visible("user-a")}
|
|
assert user_a_ids == {"builtin-agent", "private-a", "published-b"}
|
|
|
|
user_c_ids = {agent["id"] for agent in await store.list_visible("user-c")}
|
|
assert user_c_ids == {"builtin-agent", "published-b"}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_agent_store_only_owner_can_update_or_delete() -> None:
|
|
store = MemoryAgentStore()
|
|
|
|
await store.ensure_builtin({"id": "builtin-agent", "name": "Built in"})
|
|
await store.create({"id": "private-a", "name": "Owned", "user_id": "user-a"})
|
|
|
|
assert await store.update("builtin-agent", "user-a", {"name": "Nope"}) is None
|
|
assert await store.update("private-a", "user-b", {"name": "Nope"}) is None
|
|
|
|
updated = await store.update("private-a", "user-a", {"name": "Renamed"})
|
|
assert updated is not None
|
|
assert updated["name"] == "Renamed"
|
|
|
|
assert await store.delete("private-a", "user-b") is False
|
|
assert await store.delete("private-a", "user-a") is True
|
|
|
|
|
|
def test_build_run_config_injects_custom_assistant_as_agent_id() -> None:
|
|
config = build_run_config("thread-1", None, None, assistant_id="agent-123")
|
|
|
|
assert config["configurable"]["thread_id"] == "thread-1"
|
|
assert config["configurable"]["agent_id"] == "agent-123"
|
|
assert "agent_name" not in config["configurable"]
|
|
|
|
|
|
def test_write_agent_config_persists_skills_into_config_yaml(isolated_paths) -> None:
|
|
agent_id = "agent-001"
|
|
_write_agent_config(
|
|
agent_id,
|
|
name="中文助手",
|
|
description="desc",
|
|
skills=["s1", "s2"],
|
|
model="gpt-x",
|
|
tool_groups=["tg-a"],
|
|
)
|
|
|
|
config_file = paths_module.get_paths().agent_dir(agent_id) / "config.yaml"
|
|
raw = yaml.safe_load(config_file.read_text(encoding="utf-8"))
|
|
|
|
assert raw["id"] == agent_id
|
|
assert raw["name"] == "中文助手"
|
|
assert raw["description"] == "desc"
|
|
assert raw["skills"] == ["s1", "s2"]
|
|
assert raw["model"] == "gpt-x"
|
|
assert raw["tool_groups"] == ["tg-a"]
|
|
|
|
|
|
def test_load_config_extras_returns_values_written_to_yaml(isolated_paths) -> None:
|
|
agent_id = "agent-002"
|
|
_write_agent_config(
|
|
agent_id,
|
|
name="agent-002",
|
|
description="",
|
|
skills=["s1"],
|
|
model=None,
|
|
tool_groups=None,
|
|
)
|
|
|
|
extras = _load_config_extras(agent_id)
|
|
assert extras == {"skills": ["s1"], "model": None, "tool_groups": None}
|
|
|
|
|
|
def test_load_config_extras_missing_directory_returns_none_triple(isolated_paths) -> None:
|
|
extras = _load_config_extras("does-not-exist")
|
|
assert extras == {"skills": None, "model": None, "tool_groups": None}
|
|
|
|
|
|
def test_write_agent_config_explicit_empty_skills_is_persisted(isolated_paths) -> None:
|
|
agent_id = "agent-003"
|
|
_write_agent_config(agent_id, name="x", description="", skills=[])
|
|
|
|
extras = _load_config_extras(agent_id)
|
|
assert extras["skills"] == []
|
|
|
|
|
|
def test_write_agent_config_omits_skills_when_none(isolated_paths) -> None:
|
|
agent_id = "agent-004"
|
|
_write_agent_config(agent_id, name="x", description="")
|
|
|
|
config_file = paths_module.get_paths().agent_dir(agent_id) / "config.yaml"
|
|
raw = yaml.safe_load(config_file.read_text(encoding="utf-8"))
|
|
|
|
assert "skills" not in raw
|
|
assert "model" not in raw
|
|
assert "tool_groups" not in raw
|