315 lines
12 KiB
Python
315 lines
12 KiB
Python
"""U8 GEO 适配层集成测试
|
|
|
|
验证 YAML 配置文件、ConfigDrivenAgent 创建、Custom Handler 路由等。
|
|
测试在 fischer-agentkit 环境中运行，不依赖 GEO 业务代码。
|
|
"""
|
|
|
|
import pytest
|
|
import yaml
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
from agentkit.core.config_driven import AgentConfig, ConfigDrivenAgent
|
|
from agentkit.core.protocol import TaskMessage, TaskStatus
|
|
from agentkit.tools.function_tool import FunctionTool
|
|
from agentkit.tools.registry import ToolRegistry
|
|
|
|
|
|
CONFIGS_DIR = Path(__file__).parent.parent.parent.parent / "geo" / "backend" / "app" / "agent_framework" / "agents" / "configs"
|
|
|
|
|
|
class TestYAMLConfigLoading:
|
|
"""测试 YAML 配置文件加载"""
|
|
|
|
@pytest.mark.parametrize("yaml_file", [
|
|
"citation_detector.yaml",
|
|
"content_generator.yaml",
|
|
"deai_agent.yaml",
|
|
"geo_optimizer.yaml",
|
|
"monitor.yaml",
|
|
"schema_advisor.yaml",
|
|
"competitor_analyzer.yaml",
|
|
"trend_agent.yaml",
|
|
])
|
|
def test_yaml_file_exists(self, yaml_file):
|
|
path = CONFIGS_DIR / yaml_file
|
|
assert path.exists(), f"Config file {yaml_file} not found at {path}"
|
|
|
|
@pytest.mark.parametrize("yaml_file", [
|
|
"citation_detector.yaml",
|
|
"content_generator.yaml",
|
|
"deai_agent.yaml",
|
|
"geo_optimizer.yaml",
|
|
"monitor.yaml",
|
|
"schema_advisor.yaml",
|
|
"competitor_analyzer.yaml",
|
|
"trend_agent.yaml",
|
|
])
|
|
def test_yaml_valid_structure(self, yaml_file):
|
|
path = CONFIGS_DIR / yaml_file
|
|
with open(path) as f:
|
|
data = yaml.safe_load(f)
|
|
assert isinstance(data, dict)
|
|
assert "name" in data
|
|
assert "agent_type" in data
|
|
assert "task_mode" in data
|
|
assert "supported_tasks" in data
|
|
assert data["task_mode"] in {"llm_generate", "tool_call", "custom"}
|
|
|
|
@pytest.mark.parametrize("yaml_file", [
|
|
"citation_detector.yaml",
|
|
"content_generator.yaml",
|
|
"deai_agent.yaml",
|
|
"geo_optimizer.yaml",
|
|
"monitor.yaml",
|
|
"schema_advisor.yaml",
|
|
"competitor_analyzer.yaml",
|
|
"trend_agent.yaml",
|
|
])
|
|
def test_yaml_to_agent_config(self, yaml_file):
|
|
path = CONFIGS_DIR / yaml_file
|
|
config = AgentConfig.from_yaml(str(path))
|
|
assert config.name
|
|
assert config.agent_type
|
|
assert config.task_mode
|
|
assert len(config.supported_tasks) > 0
|
|
|
|
def test_llm_generate_agents_have_prompt(self):
|
|
llm_agents = ["content_generator.yaml", "deai_agent.yaml", "geo_optimizer.yaml"]
|
|
for yaml_file in llm_agents:
|
|
path = CONFIGS_DIR / yaml_file
|
|
config = AgentConfig.from_yaml(str(path))
|
|
assert config.prompt, f"{yaml_file}: llm_generate mode requires prompt"
|
|
assert "identity" in config.prompt
|
|
|
|
def test_custom_agents_have_handler(self):
|
|
custom_agents = ["citation_detector.yaml", "monitor.yaml", "schema_advisor.yaml"]
|
|
for yaml_file in custom_agents:
|
|
path = CONFIGS_DIR / yaml_file
|
|
config = AgentConfig.from_yaml(str(path))
|
|
assert config.custom_handler, f"{yaml_file}: custom mode requires custom_handler"
|
|
|
|
def test_tool_call_agents_have_tools(self):
|
|
tool_agents = ["competitor_analyzer.yaml", "trend_agent.yaml"]
|
|
for yaml_file in tool_agents:
|
|
path = CONFIGS_DIR / yaml_file
|
|
config = AgentConfig.from_yaml(str(path))
|
|
assert config.tools, f"{yaml_file}: tool_call mode requires tools list"
|
|
|
|
|
|
class TestConfigDrivenAgentCreation:
|
|
"""测试从 YAML 创建 ConfigDrivenAgent"""
|
|
|
|
def test_create_llm_generate_agent(self):
|
|
config = AgentConfig.from_yaml(str(CONFIGS_DIR / "content_generator.yaml"))
|
|
tool_registry = ToolRegistry()
|
|
agent = ConfigDrivenAgent(config=config, tool_registry=tool_registry)
|
|
assert agent.name == "content_generator"
|
|
assert agent.agent_type == "content_generation"
|
|
assert agent.prompt_template is not None
|
|
|
|
def test_create_tool_call_agent(self):
|
|
config = AgentConfig.from_yaml(str(CONFIGS_DIR / "competitor_analyzer.yaml"))
|
|
tool_registry = ToolRegistry()
|
|
|
|
async def mock_analyze(**kwargs):
|
|
return {"result": "mock"}
|
|
|
|
tool_registry.register(
|
|
FunctionTool(name="competitor_analyze", description="mock", func=mock_analyze)
|
|
)
|
|
tool_registry.register(
|
|
FunctionTool(name="competitor_gap_analysis", description="mock", func=mock_analyze)
|
|
)
|
|
|
|
agent = ConfigDrivenAgent(config=config, tool_registry=tool_registry)
|
|
assert agent.name == "competitor_analyzer"
|
|
assert len(agent._tools) == 2
|
|
|
|
def test_create_custom_agent(self):
|
|
config = AgentConfig.from_yaml(str(CONFIGS_DIR / "citation_detector.yaml"))
|
|
|
|
async def mock_handler(task):
|
|
return {"mock": True}
|
|
|
|
custom_handlers = {
|
|
"app.agent_framework.agents.custom_handlers.citation_handler.handle_citation_task": mock_handler,
|
|
}
|
|
|
|
agent = ConfigDrivenAgent(config=config, custom_handlers=custom_handlers)
|
|
assert agent.name == "citation_detector"
|
|
|
|
def test_create_all_8_agents(self):
|
|
"""验证所有 8 个 Agent 都能成功创建"""
|
|
for yaml_file in CONFIGS_DIR.glob("*.yaml"):
|
|
config = AgentConfig.from_yaml(str(yaml_file))
|
|
tool_registry = ToolRegistry()
|
|
|
|
# 为 tool_call 模式注册 mock 工具
|
|
for tool_name in config.tools:
|
|
async def mock_func(**kwargs):
|
|
return {"mock": True}
|
|
tool_registry.register(
|
|
FunctionTool(name=tool_name, description=f"Mock {tool_name}", func=mock_func)
|
|
)
|
|
|
|
# 为 custom 模式提供 mock handler
|
|
custom_handlers = {}
|
|
if config.custom_handler:
|
|
async def mock_handler(task):
|
|
return {"mock": True}
|
|
custom_handlers[config.custom_handler] = mock_handler
|
|
|
|
agent = ConfigDrivenAgent(
|
|
config=config,
|
|
tool_registry=tool_registry,
|
|
custom_handlers=custom_handlers,
|
|
)
|
|
assert agent.name == config.name
|
|
|
|
|
|
class TestCustomHandlerRouting:
|
|
"""测试 Custom Handler 路由"""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_citation_handler_routing(self):
|
|
config = AgentConfig.from_yaml(str(CONFIGS_DIR / "citation_detector.yaml"))
|
|
|
|
call_log = []
|
|
|
|
async def mock_handler(task):
|
|
call_log.append(task.task_type)
|
|
return {"mock": True, "task_type": task.task_type}
|
|
|
|
custom_handlers = {
|
|
"app.agent_framework.agents.custom_handlers.citation_handler.handle_citation_task": mock_handler,
|
|
}
|
|
|
|
agent = ConfigDrivenAgent(config=config, custom_handlers=custom_handlers)
|
|
task = TaskMessage(
|
|
task_id="test-1",
|
|
agent_name="citation_detector",
|
|
task_type="citation_detect",
|
|
priority=0,
|
|
input_data={"query_id": "test-qid"},
|
|
callback_url=None,
|
|
created_at=datetime.now(timezone.utc),
|
|
)
|
|
result = await agent.execute(task)
|
|
assert result.status == TaskStatus.COMPLETED
|
|
assert "citation_detect" in call_log
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_monitor_handler_routing(self):
|
|
config = AgentConfig.from_yaml(str(CONFIGS_DIR / "monitor.yaml"))
|
|
|
|
async def mock_handler(task):
|
|
return {"brand_id": task.input_data.get("brand_id"), "reports": []}
|
|
|
|
custom_handlers = {
|
|
"app.agent_framework.agents.custom_handlers.monitor_handler.handle_monitor_task": mock_handler,
|
|
}
|
|
|
|
agent = ConfigDrivenAgent(config=config, custom_handlers=custom_handlers)
|
|
task = TaskMessage(
|
|
task_id="test-2",
|
|
agent_name="monitor",
|
|
task_type="monitor_track",
|
|
priority=0,
|
|
input_data={"brand_id": "test-brand-id"},
|
|
callback_url=None,
|
|
created_at=datetime.now(timezone.utc),
|
|
)
|
|
result = await agent.execute(task)
|
|
assert result.status == TaskStatus.COMPLETED
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_schema_handler_routing(self):
|
|
config = AgentConfig.from_yaml(str(CONFIGS_DIR / "schema_advisor.yaml"))
|
|
|
|
async def mock_handler(task):
|
|
return {"brand_id": task.input_data.get("brand_id"), "suggestions": [], "total": 0}
|
|
|
|
custom_handlers = {
|
|
"app.agent_framework.agents.custom_handlers.schema_handler.handle_schema_task": mock_handler,
|
|
}
|
|
|
|
agent = ConfigDrivenAgent(config=config, custom_handlers=custom_handlers)
|
|
task = TaskMessage(
|
|
task_id="test-3",
|
|
agent_name="schema_advisor",
|
|
task_type="schema_advise",
|
|
priority=0,
|
|
input_data={"brand_id": "test-brand-id"},
|
|
callback_url=None,
|
|
created_at=datetime.now(timezone.utc),
|
|
)
|
|
result = await agent.execute(task)
|
|
assert result.status == TaskStatus.COMPLETED
|
|
|
|
|
|
class TestToolRegistration:
|
|
"""测试 Tool 注册完整性"""
|
|
|
|
def test_all_yaml_referenced_tools_registered(self):
|
|
registry = ToolRegistry()
|
|
all_tool_names = set()
|
|
for yaml_file in CONFIGS_DIR.glob("*.yaml"):
|
|
config = AgentConfig.from_yaml(str(yaml_file))
|
|
all_tool_names.update(config.tools)
|
|
|
|
for tool_name in all_tool_names:
|
|
async def mock_func(**kwargs):
|
|
return {"mock": True}
|
|
registry.register(
|
|
FunctionTool(name=tool_name, description=f"Mock {tool_name}", func=mock_func)
|
|
)
|
|
|
|
for tool_name in all_tool_names:
|
|
assert registry.has_tool(tool_name), f"Tool '{tool_name}' not registered"
|
|
|
|
|
|
class TestAdapterCompatibility:
|
|
"""测试适配层兼容性"""
|
|
|
|
def test_yaml_configs_count(self):
|
|
yaml_files = list(CONFIGS_DIR.glob("*.yaml"))
|
|
assert len(yaml_files) == 8, f"Expected 8 YAML configs, found {len(yaml_files)}"
|
|
|
|
def test_all_agent_names_unique(self):
|
|
names = []
|
|
for yaml_file in CONFIGS_DIR.glob("*.yaml"):
|
|
config = AgentConfig.from_yaml(str(yaml_file))
|
|
names.append(config.name)
|
|
assert len(names) == len(set(names)), f"Duplicate agent names: {names}"
|
|
|
|
def test_all_agent_types_unique(self):
|
|
types = []
|
|
for yaml_file in CONFIGS_DIR.glob("*.yaml"):
|
|
config = AgentConfig.from_yaml(str(yaml_file))
|
|
types.append(config.agent_type)
|
|
assert len(types) == len(set(types)), f"Duplicate agent types: {types}"
|
|
|
|
def test_supported_tasks_no_overlap(self):
|
|
all_tasks = {}
|
|
for yaml_file in CONFIGS_DIR.glob("*.yaml"):
|
|
config = AgentConfig.from_yaml(str(yaml_file))
|
|
for task in config.supported_tasks:
|
|
if task in all_tasks:
|
|
assert False, f"Task '{task}' defined in both '{all_tasks[task]}' and '{config.name}'"
|
|
all_tasks[task] = config.name
|
|
|
|
def test_migration_script_exists(self):
|
|
migration_path = (
|
|
Path(__file__).parent.parent.parent.parent
|
|
/ "geo" / "backend" / "alembic" / "versions" / "b001_agentkit_extension.py"
|
|
)
|
|
assert migration_path.exists(), "Migration script not found"
|
|
|
|
def test_adapter_module_exists(self):
|
|
adapter_path = (
|
|
Path(__file__).parent.parent.parent.parent
|
|
/ "geo" / "backend" / "app" / "agent_framework" / "adapter.py"
|
|
)
|
|
assert adapter_path.exists(), "adapter.py not found"
|