集成测试

单元测试验证了各个 Agent 的功能，集成测试则验证它们协作是否正常。

测试金字塔

Agent 系统的测试分三层：

        /\
       /  \  端到端测试（少量）
      /----\
     /      \  集成测试（适量）
    /--------\
   /          \  单元测试（大量）
  --------------

层级	测试内容	数量	执行速度
单元测试	单个 Agent 功能	多	快
集成测试	Agent 间协作	中	中
端到端测试	完整流程	少	慢

集成测试策略

1. 节点间数据传递

验证状态在节点间正确传递：

# tests/test_integration.py
import pytest
from graph.workflow import create_workflow_with_routing
from graph.state import ResearchState

class TestNodeIntegration:
    def test_search_to_read(self):
        """测试搜索结果能传递给阅读节点"""
        app = create_workflow_with_routing()

        initial = {
            "topic": "Python 异步编程",
            "research_type": "quick",
            "queries": [], "urls": [], "summaries": [],
            "analysis": None, "report": None,
            "current_step": "init", "errors": [], "progress": 0
        }

        # 只执行到 read 节点
        events = list(app.stream(initial))

        # 验证搜索结果
        search_output = events[0].get("search", {})
        assert len(search_output.get("urls", [])) > 0

        # 验证阅读结果
        read_output = events[1].get("read", {})
        assert len(read_output.get("summaries", [])) > 0

2. 条件路由测试

验证路由逻辑正确：

# tests/test_integration.py
class TestRouting:
    def test_quick_research_skips_analysis(self):
        """快速研究应跳过分析步骤"""
        app = create_workflow_with_routing()

        initial = {
            "topic": "测试主题",
            "research_type": "quick",  # 快速模式
            "queries": [], "urls": [], "summaries": [],
            "analysis": None, "report": None,
            "current_step": "init", "errors": [], "progress": 0
        }

        events = list(app.stream(initial))
        node_names = [list(e.keys())[0] for e in events]

        # 快速模式不应包含 analyze
        assert "analyze" not in node_names
        assert "summarize" in node_names

    def test_full_research_includes_analysis(self):
        """完整研究应包含分析步骤"""
        app = create_workflow_with_routing()

        initial = {
            "topic": "测试主题",
            "research_type": "full",  # 完整模式
            "queries": [], "urls": [], "summaries": [],
            "analysis": None, "report": None,
            "current_step": "init", "errors": [], "progress": 0
        }

        events = list(app.stream(initial))
        node_names = [list(e.keys())[0] for e in events]

        assert "analyze" in node_names

3. 错误传播测试

验证错误能正确传播和处理：

# tests/test_integration.py
class TestErrorHandling:
    def test_search_error_propagates(self):
        """搜索错误应记录到状态"""
        app = create_workflow_with_routing()

        # 使用无效主题触发错误
        initial = {
            "topic": "",  # 空主题
            "research_type": "quick",
            "queries": [], "urls": [], "summaries": [],
            "analysis": None, "report": None,
            "current_step": "init", "errors": [], "progress": 0
        }

        result = app.invoke(initial)
        # 应该有错误记录或优雅降级
        assert result.get("errors") or result.get("report")

    def test_partial_failure_continues(self):
        """部分 URL 失败不应中断流程"""
        # 模拟部分 URL 不可访问的情况
        pass  # 需要 mock

Mock 测试

使用 Mock 隔离外部依赖：

# tests/test_integration.py
from unittest.mock import Mock, patch

class TestWithMock:
    @patch('agents.search.TavilySearchResults')
    def test_search_with_mock(self, mock_tavily):
        """使用 Mock 测试搜索"""
        # 设置 Mock 返回值
        mock_tavily.return_value.invoke.return_value = [
            {"url": "https://example.com/1", "content": "Test 1"},
            {"url": "https://example.com/2", "content": "Test 2"}
        ]

        from agents.search import SearchAgent
        agent = SearchAgent()
        result = agent.search("测试")

        assert len(result["results"]) == 2

    @patch('agents.reader.WebBaseLoader')
    def test_reader_with_mock(self, mock_loader):
        """使用 Mock 测试阅读"""
        mock_doc = Mock()
        mock_doc.page_content = "这是测试内容" * 100
        mock_loader.return_value.load.return_value = [mock_doc]

        from agents.reader import ReaderAgent
        agent = ReaderAgent()
        content = agent.load_url("https://example.com")

        assert content is not None
        assert "测试内容" in content

性能测试

验证系统在负载下的表现：

# tests/test_performance.py
import time
import pytest

class TestPerformance:
    def test_research_completes_in_time(self):
        """研究应在合理时间内完成"""
        app = create_workflow_with_routing()

        initial = {
            "topic": "Python 基础",
            "research_type": "quick",
            "queries": [], "urls": [], "summaries": [],
            "analysis": None, "report": None,
            "current_step": "init", "errors": [], "progress": 0
        }

        start = time.time()
        result = app.invoke(initial)
        duration = time.time() - start

        # 快速研究应在 60 秒内完成
        assert duration < 60
        assert result.get("report") is not None

    def test_concurrent_requests(self):
        """测试并发请求"""
        import asyncio

        async def run_research():
            app = create_workflow_with_routing()
            initial = {
                "topic": "测试",
                "research_type": "quick",
                "queries": [], "urls": [], "summaries": [],
                "analysis": None, "report": None,
                "current_step": "init", "errors": [], "progress": 0
            }
            return app.invoke(initial)

        async def test_concurrent():
            tasks = [run_research() for _ in range(3)]
            results = await asyncio.gather(*tasks)
            return results

        results = asyncio.run(test_concurrent())
        assert len(results) == 3

端到端测试

完整流程测试：

# tests/test_e2e.py
class TestEndToEnd:
    def test_full_research_flow(self):
        """完整研究流程测试"""
        app = create_workflow_with_routing()

        initial = {
            "topic": "LangGraph 入门教程",
            "research_type": "full",
            "queries": [], "urls": [], "summaries": [],
            "analysis": None, "report": None,
            "current_step": "init", "errors": [], "progress": 0
        }

        result = app.invoke(initial)

        # 验证最终输出
        assert result["progress"] == 100
        assert result["report"] is not None
        assert len(result["report"]) > 500
        assert "LangGraph" in result["report"]

        # 验证中间状态
        assert len(result["urls"]) > 0
        assert len(result["summaries"]) > 0
        assert result["analysis"] is not None

测试配置

# pytest.ini
[pytest]
testpaths = tests
python_files = test_*.py
python_functions = test_*
addopts = -v --tb=short
markers =
    slow: 标记慢速测试
    integration: 集成测试
    e2e: 端到端测试

运行测试

# 运行所有测试
pytest

# 只运行单元测试
pytest tests/test_agents.py

# 只运行集成测试
pytest -m integration

# 跳过慢速测试
pytest -m "not slow"

# 生成覆盖率报告
pytest --cov=agents --cov=graph --cov-report=html

验收标准

完成集成测试后，你应该有：

• [ ] 节点间数据传递测试通过
• [ ] 条件路由测试通过
• [ ] 错误处理测试通过
• [ ] 端到端测试通过
• [ ] 测试覆盖率 > 70%

下一步

测试通过，准备部署上线。

继续：部署监控 →

---

← 返回前端界面 | 返回项目二