From 133ae3927e78fa3ecb003942d0d985888072ac89 Mon Sep 17 00:00:00 2001 From: chiguyong Date: Fri, 5 Jun 2026 17:25:51 +0800 Subject: [PATCH] test(u8): add GEO business integration tests for YAML configs and agent creation 41 tests covering: - 8 YAML config file loading and validation - ConfigDrivenAgent creation from YAML (llm_generate/tool_call/custom) - Custom handler routing (citation/monitor/schema) - Tool registration completeness - Adapter compatibility (unique names/types, no task overlap) --- tests/unit/test_u8_geo_integration.py | 314 ++++++++++++++++++++++++++ 1 file changed, 314 insertions(+) create mode 100644 tests/unit/test_u8_geo_integration.py diff --git a/tests/unit/test_u8_geo_integration.py b/tests/unit/test_u8_geo_integration.py new file mode 100644 index 0000000..921342a --- /dev/null +++ b/tests/unit/test_u8_geo_integration.py @@ -0,0 +1,314 @@ +"""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"