第十七章　测试策略：pytest 与智能体系统的 Mock 艺术

# tests/conftest.py（简化）

def _create_sdk_mock():
    """创建 SDK 模块的全面 mock。"""
    mock = MagicMock()
    mock.ClaudeAgentOptions = MagicMock
    mock.ClaudeSDKClient = MagicMock
    mock.HookMatcher = MagicMock
    return mock

# 在任何其他 import 之前注入 mock
if 'claude_agent_sdk' not in sys.modules:
    sys.modules['claude_agent_sdk'] = _create_sdk_mock()
    sys.modules['claude_agent_sdk.types'] = MagicMock()

if 'claude_code_sdk' not in sys.modules:
    sys.modules['claude_code_sdk'] = _create_sdk_mock()
    sys.modules['claude_code_sdk.types'] = MagicMock()

# 连 dotenv 也要 mock：防止 cli.utils 在 import 时调用 sys.exit()
if 'dotenv' not in sys.modules:
    sys.modules['dotenv'] = MagicMock()
    sys.modules['dotenv'].load_dotenv = MagicMock()

# 可能被某些测试 mock 的模块列表
_POTENTIALLY_MOCKED_MODULES = [
    'claude_code_sdk', 'claude_code_sdk.types',
    'claude_agent_sdk', 'claude_agent_sdk.types',
    'dotenv', 'ui', 'progress', 'task_logger',
    # ...更多模块
]

# 在 import 时保存模块的原始状态
_original_module_state = {}
for _name in _POTENTIALLY_MOCKED_MODULES:
    if _name in sys.modules:
        _original_module_state[_name] = sys.modules[_name]

def _cleanup_mocked_modules():
    """移除 sys.modules 中所有 MagicMock 模块。"""
    for name in _POTENTIALLY_MOCKED_MODULES:
        if name in sys.modules:
            module = sys.modules[name]
            if isinstance(module, MagicMock):
                if name in _original_module_state:
                    # 恢复原始模块（如果原来就有）
                    sys.modules[name] = _original_module_state[name]
                else:
                    # 删除 mock（原来没有这个模块）
                    del sys.modules[name]

# 不同测试模块需要不同的 mock 集
module_mocks = {
    'test_qa_loop': {'claude_code_sdk', 'claude_code_sdk.types',
                     'claude_agent_sdk', 'claude_agent_sdk.types'},

    'test_spec_pipeline': {'claude_code_sdk', 'claude_code_sdk.types',
                           'init', 'client', 'review', 'task_logger', 'ui', 'validate_spec'},

    'test_qa_fixer': {'claude_agent_sdk', 'ui', 'progress', 'task_logger',
                      'agents.memory_manager', 'agents.base',
                      'core.error_utils', 'security.tool_input_validator'},
}

def pytest_runtest_setup(item):
    """每个测试运行前清理：只保留当前测试需要的 mock。"""
    module_name = item.module.__name__
    preserved_mocks = module_mocks.get(module_name, set())

    for name in _POTENTIALLY_MOCKED_MODULES:
        if name in preserved_mocks:
            continue  # 当前测试需要这个 mock，保留
        if name in sys.modules and isinstance(sys.modules[name], MagicMock):
            # 不需要的 mock 清理掉
            del sys.modules[name]

@pytest.fixture
def temp_dir() -> Generator[Path, None, None]:
    """提供隔离的临时目录，测试结束自动清理。"""
    with tempfile.TemporaryDirectory() as tmp:
        yield Path(tmp)

@pytest.fixture
def git_repo(temp_dir: Path) -> Generator[Path, None, None]:
    """创建一个初始化好的 Git 仓库（有初始 commit）。"""
    subprocess.run(
        ['git', 'init', '--initial-branch=main', str(temp_dir)],
        check=True, capture_output=True
    )
    subprocess.run(
        ['git', '-C', str(temp_dir), 'config', 'user.email', 'test@test.com'],
        check=True, capture_output=True
    )
    subprocess.run(
        ['git', '-C', str(temp_dir), 'config', 'user.name', 'Test User'],
        check=True, capture_output=True
    )

    # 创建初始 commit（很多 Git 操作需要至少一个 commit）
    readme = temp_dir / "README.md"
    readme.write_text("# Test Project\n")
    subprocess.run(['git', '-C', str(temp_dir), 'add', '.'], check=True)
    subprocess.run(
        ['git', '-C', str(temp_dir), 'commit', '-m', 'Initial commit'],
        check=True, capture_output=True
    )
    yield temp_dir

# tests/test_security.py

def test_rm_validator_blocks_recursive():
    """测试 rm -rf 被正确拦截。"""
    from security.validator import validate_command
    result = validate_command('rm', ['-rf', '/home/user/project'])
    assert result.allowed is False
    assert 'recursive' in result.reason.lower()

def test_rm_validator_allows_single_file():
    """测试删除单文件被允许。"""
    from security.validator import validate_command
    result = validate_command('rm', ['/tmp/test.txt'])
    assert result.allowed is True

# tests/test_worktree.py（概念化）

def test_worktree_create(git_repo):
    """测试在真实 Git 仓库里创建 worktree。"""
    from core.worktree import WorktreeManager

    manager = WorktreeManager(git_repo)
    worktree_path = manager.create_worktree('feature-branch')

    # 验证：worktree 目录真实存在
    assert worktree_path.exists()
    # 验证：分支被创建
    result = subprocess.run(
        ['git', '-C', str(git_repo), 'branch'],
        capture_output=True, text=True
    )
    assert 'feature-branch' in result.stdout

def test_project_analyzer_detects_python(temp_dir):
    """测试 Python 项目被正确识别。"""
    (temp_dir / 'requirements.txt').write_text('flask==2.0')
    (temp_dir / 'main.py').write_text('print("hello")')

    from project.analyzer import ProjectAnalyzer
    analyzer = ProjectAnalyzer(str(temp_dir))
    stack = analyzer.analyze()

    assert 'python' in stack.languages

# tests/test_qa_loop.py（简化）

@pytest.fixture
def mock_qa_agent():
    """模拟 QA Agent 的输出序列。"""
    # 第一次：发现问题；第二次：验证通过
    responses = [
        {"status": "FAIL", "issues": ["Missing error handling"]},
        {"status": "PASS", "issues": []},
    ]
    call_count = 0

    async def fake_run(*args, **kwargs):
        nonlocal call_count
        response = responses[min(call_count, len(responses) - 1)]
        call_count += 1
        return response

    with patch('qa.loop.run_qa_reviewer', side_effect=fake_run):
        yield

async def test_qa_loop_retries_on_failure(git_repo, mock_qa_agent):
    """QA Loop 在失败后重试，第二次通过。"""
    from qa.loop import run_qa_validation_loop

    result = await run_qa_validation_loop(
        project_path=str(git_repo),
        spec_id='test-001',
        max_retries=3
    )

    assert result.status == 'PASSED'
    assert result.iterations == 2

// apps/frontend/src/renderer/stores/__tests__/task-store.test.ts（概念化）

import { renderHook, act } from '@testing-library/react';
import { useTaskStore } from '../task-store';

describe('TaskStore', () => {
  beforeEach(() => {
    // 重置 store 状态（Zustand 支持）
    useTaskStore.setState({ tasks: [], selectedTaskId: null });
  });

  it('should enforce MAX_LOG_ENTRIES limit', () => {
    const { result } = renderHook(() => useTaskStore());

    act(() => {
      result.current.addTask({ id: 'task-1', status: 'running', logs: [] });
    });

    // 追加 5100 条日志
    const logs = Array.from({ length: 5100 }, (_, i) => `log-${i}`);
    act(() => {
      result.current.batchAppendLogs('task-1', logs);
    });

    const task = result.current.tasks.find(t => t.id === 'task-1');
    expect(task?.logs?.length).toBe(5000); // MAX_LOG_ENTRIES
  });
});

# lab17/conftest.py
import sys
from unittest.mock import MagicMock

def _create_ai_sdk_mock():
    """创建 AI SDK 的基本 mock。"""
    mock = MagicMock()
    # TODO Part A: 给 mock 添加常用的 SDK 属性
    # mock.SomeClass = MagicMock
    # mock.some_constant = "value"
    return mock

# TODO Part A: 在 conftest 顶部注入 mock（在其他 import 之前）
# 如果 'fake_ai_sdk' 未在 sys.modules 中，就注入 mock

# lab17/conftest.py（续）
import subprocess
import tempfile
from pathlib import Path
from typing import Generator
import pytest

@pytest.fixture
def temp_dir() -> Generator[Path, None, None]:
    """临时目录 fixture（自动清理）。"""
    # TODO Part B: 使用 tempfile.TemporaryDirectory 创建临时目录
    # 记得 yield 而不是 return
    pass

@pytest.fixture
def git_repo(temp_dir: Path) -> Generator[Path, None, None]:
    """初始化好的 Git 仓库 fixture。"""
    # TODO Part B:
    # 1. git init --initial-branch=main {temp_dir}
    # 2. git config user.email / user.name
    # 3. 创建 README.md，git add，git commit -m "Initial commit"
    # 4. yield temp_dir
    pass

# lab17/test_agent_mock.py
from unittest.mock import patch, AsyncMock, MagicMock
import pytest

def make_sequence_mock(responses: list):
    """
    创建一个按顺序返回响应的 mock。
    第 N 次调用返回 responses[N]，超出范围时返回最后一个。
    """
    # TODO Part C: 实现响应序列 mock
    # 提示：用 AsyncMock(side_effect=...) 或用列表的迭代器

    call_count = 0
    async def mock_func(*args, **kwargs):
        nonlocal call_count
        # TODO: 返回对应的响应，处理越界情况
        pass

    return AsyncMock(side_effect=mock_func)

@pytest.mark.asyncio
async def test_retry_on_failure():
    """测试：第一次失败，第二次成功。"""
    responses = [
        {"status": "FAIL", "issues": ["Bug found"]},
        {"status": "PASS", "issues": []},
    ]
    mock_agent = make_sequence_mock(responses)

    # 模拟一个会重试的函数
    async def run_with_retry(agent, max_retries=3):
        for attempt in range(max_retries):
            result = await agent()
            if result["status"] == "PASS":
                return result, attempt + 1
        return None, max_retries

    result, attempts = await run_with_retry(mock_agent)

    # TODO Part C: 验证
    # assert result["status"] == "PASS"
    # assert attempts == 2  # 第二次才成功
    # assert mock_agent.call_count == 2