🧪 单元测试 Agent 核心逻辑开发文档

unit_test_agent/
├── core/
│   ├── __init__.py              # 包初始化
│   ├── unit_test_agent.py       # 核心 Agent 类（主入口）
│   ├── parsers.py               # 代码解析模块（Python/Java/JS AST 解析）
│   ├── analyzer.py              # 测试场景分析引擎
│   ├── generator.py             # AAA 模式测试代码生成器
│   ├── coverage.py              # 覆盖率预估模块
│   └── claude_integration.py    # Claude Code API 集成
├── templates/                   # 测试文件模板（可选）
│   ├── python_test_template.py
│   ├── java_test_template.java
│   └── js_test_template.js
├── tests/                       # Agent 自身测试
│   ├── test_parsers.py
│   ├── test_analyzer.py
│   └── test_generator.py
├── examples/                    # 使用示例
│   ├── example_python.py
│   ├── example_java.py
│   └── example_javascript.js
├── requirements.txt             # 依赖清单
├── README.md                    # 使用说明
└── setup.py                     # 安装脚本

class UnitTestAgent:
    """单元测试自动生成 Agent"""
    
    def __init__(self, config: TestGenerationConfig):
        """初始化 Agent
        
        Args:
            config: 测试生成配置
        """
        pass
    
    async def generate_tests(
        self,
        source_code: str,
        source_file_path: str,
        output_dir: Optional[str] = None
    ) -> TestGenerationResult:
        """生成单元测试（单个文件）
        
        Args:
            source_code: 源代码内容
            source_file_path: 源文件路径
            output_dir: 输出目录（可选）
            
        Returns:
            TestGenerationResult: 生成结果
            
        Example:
            >>> agent = UnitTestAgent(config)
            >>> result = await agent.generate_tests(
            ...     source_code=code,
            ...     source_file_path="user_service.py"
            ... )
            >>> print(f"生成{result.total_tests_generated}个测试")
        """
        pass
    
    async def generate_tests_batch(
        self,
        source_files: List[str],
        output_dir: Optional[str] = None
    ) -> List[TestGenerationResult]:
        """批量生成单元测试
        
        Args:
            source_files: 源文件路径列表
            output_dir: 输出目录（可选）
            
        Returns:
            List[TestGenerationResult]: 结果列表
        """
        pass

async def generate_unit_tests(
    source_code: str,
    source_file_path: str,
    language: Optional[LanguageEnum] = None,
    test_framework: Optional[str] = None,
    output_dir: Optional[str] = None
) -> TestGenerationResult:
    """便捷函数：快速生成单元测试
    
    Args:
        source_code: 源代码
        source_file_path: 源文件路径
        language: 编程语言（自动检测）
        test_framework: 测试框架（默认 pytest）
        output_dir: 输出目录
        
    Returns:
        TestGenerationResult: 生成结果
        
    Example:
        >>> result = await generate_unit_tests(
        ...     source_code=code,
        ...     source_file_path="service.py"
        ... )
    """
    pass

import asyncio
from unit_test_agent.core import (
    UnitTestAgent, 
    TestGenerationConfig,
    LanguageEnum
)

async def main():
    """Python 测试生成示例"""
    
    # 1. 配置 Agent
    config = TestGenerationConfig(
        language=LanguageEnum.PYTHON,
        test_framework="pytest",
        coverage_target=0.95,
        include_normal_cases=True,
        include_exception_cases=True,
        include_boundary_cases=True
    )
    
    agent = UnitTestAgent(config)
    
    # 2. 准备源代码
    source_code = '''
class UserService:
    def __init__(self, user_repository, password_encoder):
        self.user_repository = user_repository
        self.password_encoder = password_encoder
    
    def register_user(self, username: str, email: str, password: str) -> dict:
        if not username or len(username) < 3:
            raise ValueError("用户名长度至少为 3 个字符")
        
        if self.user_repository.exists_by_username(username):
            raise ValueError(f"用户名已存在：{username}")
        
        hashed_password = self.password_encoder.encode(password)
        
        user = {
            "id": 1,
            "username": username,
            "email": email,
            "password_hash": hashed_password
        }
        
        return self.user_repository.save(user)
'''
    
    # 3. 生成测试
    result = await agent.generate_tests(
        source_code=source_code,
        source_file_path="user_service.py",
        output_dir="./tests"
    )
    
    # 4. 查看结果
    print(f"✅ 生成成功：{result.success}")
    print(f"📝 测试用例数：{result.total_tests_generated}")
    print(f"📊 预估覆盖率:")
    print(f"   - 行覆盖率：{result.estimated_line_coverage:.1%}")
    print(f"   - 分支覆盖率：{result.estimated_branch_coverage:.1%}")
    print(f"   - 函数覆盖率：{result.estimated_function_coverage:.1%}")
    print(f"⏱️  生成耗时：{result.generation_time_ms}ms")
    print(f"📁 测试文件：{result.test_file_path}")
    
    # 5. 查看生成的测试用例
    for i, test_case in enumerate(result.test_cases, 1):
        print(f"\n{i}. {test_case.test_name}")
        print(f"   场景：{test_case.scenario_type}")
        print(f"   优先级：{test_case.priority}")
        print(f"   描述：{test_case.description}")

if __name__ == "__main__":
    asyncio.run(main())

import asyncio
from unit_test_agent.core import UnitTestAgent
from pathlib import Path

async def batch_generate():
    """批量生成整个项目的测试"""
    
    agent = UnitTestAgent()
    
    # 查找所有 Python 源文件
    source_files = list(Path("./src").rglob("*.py"))
    
    # 批量生成测试
    results = await agent.generate_tests_batch(
        source_files=[str(f) for f in source_files],
        output_dir="./tests"
    )
    
    # 统计结果
    total_tests = sum(r.total_tests_generated for r in results if r.success)
    avg_coverage = sum(r.estimated_line_coverage for r in results if r.success) / len(results)
    
    print(f"\n📊 批量生成完成:")
    print(f"   - 处理文件：{len(results)}个")
    print(f"   - 生成测试：{total_tests}个")
    print(f"   - 平均覆盖率：{avg_coverage:.1%}")

asyncio.run(batch_generate())

def estimate_coverage(self, class_info, test_cases):
    """覆盖率预估算法"""
    
    # 1. 函数覆盖率
    total_methods = len([m for m in class_info.methods if m.visibility == 'public'])
    covered_methods = len(set([tc.test_name.split('_')[1] for tc in test_cases]))
    function_coverage = covered_methods / max(total_methods, 1)
    
    # 2. 行覆盖率估算
    normal_tests = len([tc for tc in test_cases if tc.scenario_type == 'normal'])
    exception_tests = len([tc for tc in test_cases if tc.scenario_type == 'exception'])
    boundary_tests = len([tc for tc in test_cases if tc.scenario_type == 'boundary'])
    
    line_coverage = min(0.95, (
        normal_tests * 0.3 +      # 正常测试覆盖 30% 代码
        exception_tests * 0.2 +   # 异常测试覆盖 20% 代码
        boundary_tests * 0.15     # 边界测试覆盖 15% 代码
    ) / max(total_methods, 1))
    
    # 3. 分支覆盖率估算
    branch_coverage = min(0.90, line_coverage * 0.9)
    
    return {
        'line': line_coverage,
        'branch': branch_coverage,
        'function': function_coverage
    }

def _build_prompt(self, class_info, method_name, scenario, language, test_framework):
    """构建生成测试的 Prompt"""
    
    prompt = f"""你是一个专业的软件测试工程师。请为以下代码生成单元测试。

【要求】
1. 遵循 AAA (Arrange-Act-Assert) 模式
2. 使用 {test_framework} 测试框架
3. 测试场景：{scenario.get('description', '正常测试')}
4. 优先级：{scenario.get('priority', 'P1')}
5. 必须包含清晰的注释

【类信息】
类名：{class_info.name}
包名：{class_info.package}

【方法信息】
方法名：{method_name}
参数：{class_info.methods[0].parameters if class_info.methods else []}
返回值：{class_info.methods[0].return_type if class_info.methods else 'void'}

【测试场景详情】
类型：{scenario.get('type', 'normal')}
{f'输入条件：{scenario.get("input_condition", "N/A")}' if scenario.get('input_condition') else ''}
{f'期望异常：{scenario.get("expected_exception", "N/A")}' if scenario.get('expected_exception') else ''}

请生成完整的测试代码（只返回代码，不要解释）："""
    
    return prompt

# 1. 克隆代码
git clone https://github.com/your-org/unit-test-agent.git
cd unit-test-agent

# 2. 创建虚拟环境
python -m venv venv
source venv/bin/activate  # Linux/Mac
# 或
venv\Scripts\activate  # Windows

# 3. 安装依赖
pip install -r requirements.txt

# 4. 安装为本地包（可选）
pip install -e .

# .env 配置文件
CLAUDE_API_KEY=sk-ant-api03-xxxxxxxxxxxxx
CLAUDE_MODEL=claude-sonnet-4-5-20250929

# 测试生成配置
DEFAULT_LANGUAGE=python
DEFAULT_FRAMEWORK=pytest
COVERAGE_TARGET=0.95
OUTPUT_DIR=./tests

# 功能开关
INCLUDE_NORMAL_CASES=true
INCLUDE_EXCEPTION_CASES=true
INCLUDE_BOUNDARY_CASES=true
INCLUDE_EDGE_CASES=false

# 性能配置
MAX_CONCURRENT_REQUESTS=5
TIMEOUT_SECONDS=30

# Jenkins Pipeline 示例
pipeline {
    agent any
    
    stages {
        stage('Generate Unit Tests') {
            steps {
                script {
                    sh '''
                        python -m unit_test_agent.cli \
                            --src-dir ./src \
                            --output-dir ./tests \
                            --language python \
                            --framework pytest
                    '''
                }
            }
        }
        
        stage('Run Tests') {
            steps {
                sh 'pytest tests/ --cov=src --cov-report=html'
            }
        }
        
        stage('Quality Gate') {
            steps {
                script {
                    def coverage = sh(script: 'pytest --cov=src --cov-report=term-missing', returnStdout: true)
                    if (coverage.contains('TOTAL') && !coverage.contains('FAIL')) {
                        echo '✅ 测试通过'
                    } else {
                        error '❌ 测试失败'
                    }
                }
            }
        }
    }
}