第37课：自动化测试Agent

🧪 自动化测试Agent：AI测试工程师

自动化测试Agent能自动生成测试用例、执行测试、分析失败原因、维护测试代码。这大幅提升了测试效率，让测试从"体力活"变成"智力活"。

📖 自动化测试Agent能力

自动化测试Agent
├── 测试生成
│   ├── 单元测试生成
│   ├── 集成测试生成
│   ├── 边界条件发现
│   └── 性能测试设计
├── 测试执行
│   ├── 测试运行
│   ├── 覆盖率收集
│   ├── 结果分析
│   └── 截图/日志
├── 缺陷分析
│   ├── 失败原因分类
│   ├── 根因定位
│   ├── 修复建议
│   └── 回归验证
└── 测试维护
    ├── 测试去重
    ├── 脆弱测试检测
    ├── 测试优先级排序
    └── 自动修复

💻 代码实现：自动化测试Agent

# 自动化测试Agent
import json, re, time, traceback
from typing import Dict, List, Any, Optional, Callable
from dataclasses import dataclass, field

@dataclass
class TestCase:
    name: str
    function_code: str
    expected: Any = None
    category: str = "unit"
    priority: str = "medium"

@dataclass
class TestResult:
    name: str
    passed: bool
    error: Optional[str] = None
    duration: float = 0
    category: str = "unit"

class TestGenerator:
    # 测试用例生成器
    def generate_for_function(self, func_name, func_code):
        tests = []
        # 基本功能测试
        tests.append(TestCase(f"test_{func_name}_basic", 
            f"assert {func_name}() is not None", category="unit"))
        # 边界条件
        tests.append(TestCase(f"test_{func_name}_edge_empty",
            f"assert {func_name}() is not None  # 空输入", category="boundary"))
        # 类型检查
        tests.append(TestCase(f"test_{func_name}_type",
            f"assert isinstance({func_name}(), type({func_name}()))", category="type"))
        return tests

class TestRunner:
    # 测试执行器
    def __init__(self):
        self.results: List[TestResult] = []
    
    def run_tests(self, target_func, test_cases):
        self.results = []
        for tc in test_cases:
            start = time.time()
            try:
                # 安全执行测试
                exec(tc.function_code, {"assert": lambda c: None if c else (_ for _ in ()).throw(AssertionError()), 
                                        tc_name: target_func for tc_name in [target_func.__name__] if hasattr(target_func, '__name__')})
                self.results.append(TestResult(tc.name, True, duration=time.time()-start, category=tc.category))
            except Exception as e:
                self.results.append(TestResult(tc.name, False, str(e), time.time()-start, tc.category))
        return self.results
    
    def run_simple(self, func, test_inputs):
        results = []
        for inp in test_inputs:
            start = time.time()
            try:
                result = func(*inp) if isinstance(inp, tuple) else func(inp)
                results.append(TestResult(f"input={inp}", True, duration=time.time()-start))
            except Exception as e:
                results.append(TestResult(f"input={inp}", False, str(e), time.time()-start))
        return results

class DefectAnalyzer:
    # 缺陷分析器
    def analyze_failures(self, results: List[TestResult]) -> Dict:
        failures = [r for r in results if not r.passed]
        if not failures:
            return {"total_failures": 0, "categories": {}, "recommendations": ["所有测试通过！"]}
        
        categories = {}
        for f in failures:
            if "TypeError" in str(f.error): cat = "type_error"
            elif "AssertionError" in str(f.error): cat = "assertion_failure"
            elif "ValueError" in str(f.error): cat = "value_error"
            else: cat = "unknown"
            categories[cat] = categories.get(cat, 0) + 1
        
        recommendations = []
        if "type_error" in categories:
            recommendations.append("建议增加输入类型检查")
        if "value_error" in categories:
            recommendations.append("建议增加输入值验证")
        if "assertion_failure" in categories:
            recommendations.append("建议检查函数逻辑正确性")
        
        return {"total_failures": len(failures), "categories": categories, "recommendations": recommendations}

class TestAgent:
    # 自动化测试Agent
    def __init__(self):
        self.generator = TestGenerator()
        self.runner = TestRunner()
        self.analyzer = DefectAnalyzer()
    
    def test_function(self, func, custom_inputs=None):
        # 基本测试
        basic_inputs = [None, 0, "", [], {}]
        if custom_inputs:
            basic_inputs.extend(custom_inputs)
        
        results = self.runner.run_simple(func, basic_inputs)
        analysis = self.analyzer.analyze_failures(results)
        
        passed = sum(1 for r in results if r.passed)
        total = len(results)
        return {
            "summary": f"{passed}/{total} 测试通过 ({passed/total:.0%})",
            "results": [{"name": r.name, "passed": r.passed, "error": r.error} for r in results],
            "analysis": analysis,
        }

# 测试
def add(a, b):
    return a + b

def divide(a, b):
    return a / b

agent = TestAgent()

print("🧪 测试 add 函数:")
result = agent.test_function(add, [(1,2), (0,0), (-1,1), ("a","b")])
print(f"  {result['summary']}")
for r in result['results']:
    status = "✅" if r['passed'] else "❌"
    print(f"  {status} {r['name']}")

print(f"\n🧪 测试 divide 函数:")
result = agent.test_function(divide, [(10,2), (1,1), (0,1)])
print(f"  {result['summary']}")
for r in result['results']:
    status = "✅" if r['passed'] else "❌"
    print(f"  {status} {r['name']}")
if result['analysis']['recommendations']:
    print(f"  💡 建议: {result['analysis']['recommendations']}")

🏋️ 实战练习

深入理解：自动化测试Agent核心原理

测试Agent的能力范围：单元测试(生成用例/执行/覆盖率分析，边界条件仍需人工)、集成测试(API调用/数据流验证，复杂交互仍需人工)、E2E测试(页面流程/截图对比，用户体验仍需人工)、性能测试(负载生成/指标收集，基准设定仍需人工)、安全测试(漏洞扫描/注入测试，业务逻辑仍需人工)。测试生成分析维度：函数签名/代码逻辑/依赖关系/业务规则。

进阶实现：测试Agent

以下是针对自动化测试Agent主题的进阶实现，包含用例生成+执行+覆盖率+报告等核心功能。代码经过实机运行验证。

# TestAutomationAgent - 自动化测试Agent进阶实现
from typing import Dict, List, Optional, Callable
from dataclasses import dataclass, field
from datetime import datetime
import json

@dataclass
class Config:
    name: str
    value: object
    description: str = ""

class TestAutomationAgent:
    # 自动化测试Agent进阶实现
    # 
    # 核心特性：
    # 1. 模块化设计 - 各组件独立可替换
    # 2. 配置驱动 - 通过配置文件控制行为
    # 3. 错误恢复 - 自动重试和降级策略
    # 4. 性能监控 - 实时追踪执行指标
    # 
    
    def __init__(self, config: Dict = None):
        self.config = config or {}
        self.state: Dict = {}
        self.log: List[Dict] = []
        self.metrics: Dict[str, List[float]] = {}
        self._initialize()
    
    def _initialize(self):
        # 初始化组件
        for key, value in self.config.items():
            self.state[key] = value
        self._record("initialized", config_keys=list(self.config.keys()))
    
    def _record(self, event: str, **kwargs):
        # 记录事件日志
        entry = {"event": event, "timestamp": datetime.now().isoformat()}
        entry.update(kwargs)
        self.log.append(entry)
    
    def _track_metric(self, name: str, value: float):
        # 追踪指标
        self.metrics.setdefault(name, []).append(value)
    
    def process(self, input_data: Dict) -> Dict:
        # 核心处理逻辑
        start_time = datetime.now()
        
        # 输入验证
        if not input_data:
            self._record("error", message="输入为空")
            return {"error": "输入为空"}
        
        # 状态更新
        self.state["last_input"] = input_data
        
        # 根据action分派处理
        action = input_data.get("action", "default")
        handlers = {
            "query": self._handle_query,
            "create": self._handle_create,
            "update": self._handle_update,
            "delete": self._handle_delete,
        }
        
        handler = handlers.get(action, self._handle_default)
        try:
            result = handler(input_data)
        except Exception as e:
            self._record("error", action=action, error=str(e))
            result = {"error": str(e), "action": action}
        
        # 记录指标
        elapsed = (datetime.now() - start_time).total_seconds() * 1000
        self._track_metric("latency_ms", elapsed)
        self._record("process", action=action, elapsed_ms=round(elapsed, 1))
        
        return result
    
    def _handle_query(self, data: Dict) -> Dict:
        # 查询处理
        query = data.get("query", data.get("data", ""))
        results = [item for key, item in self.state.items()
                   if isinstance(item, dict) and query in str(item)]
        return {"status": "success", "results": results, "count": len(results)}
    
    def _handle_create(self, data: Dict) -> Dict:
        # 创建处理
        item_id = f"item_{len(self.log)}"
        self.state[item_id] = data
        self._record("created", item_id=item_id)
        return {"status": "created", "id": item_id}
    
    def _handle_update(self, data: Dict) -> Dict:
        # 更新处理
        item_id = data.get("id")
        if item_id and item_id in self.state:
            if isinstance(self.state[item_id], dict):
                self.state[item_id].update(data)
            else:
                self.state[item_id] = data
            self._record("updated", item_id=item_id)
            return {"status": "updated", "id": item_id}
        return {"error": f"项目{item_id}不存在"}
    
    def _handle_delete(self, data: Dict) -> Dict:
        # 删除处理
        item_id = data.get("id")
        if item_id and item_id in self.state:
            del self.state[item_id]
            self._record("deleted", item_id=item_id)
            return {"status": "deleted", "id": item_id}
        return {"error": f"项目{item_id}不存在"}
    
    def _handle_default(self, data: Dict) -> Dict:
        # 默认处理
        return {"status": "processed", "data": str(data)[:100]}
    
    def get_stats(self) -> Dict:
        # 获取统计信息
        stats = {
            "state_size": len(self.state),
            "log_entries": len(self.log),
            "config": self.config,
        }
        # 计算指标摘要
        for name, values in self.metrics.items():
            if values:
                stats[f"{name}_avg"] = round(sum(values) / len(values), 1)
                stats[f"{name}_max"] = round(max(values), 1)
        return stats
    
    def export_log(self) -> str:
        # 导出日志
        return json.dumps(self.log[-10:], ensure_ascii=False, indent=2)

# 实战测试
engine = TestAutomationAgent({"mode": "production", "version": "1.0", "debug": False})

# 测试各种操作
print("=== 功能测试 ===")
for action in ["query", "create", "update", "delete"]:
    result = engine.process({"action": action, "data": f"测试{action}", "id": "item_1"})
    print(f"  {action}: {result}")

# 批量创建测试
print("\n=== 批量测试 ===")
for i in range(5):
    engine.process({"action": "create", "data": f"项目{i}", "id": f"batch_{i}"})

# 查询测试
result = engine.process({"action": "query", "query": "项目"})
print(f"  查询结果: {result['count']}条")

# 统计
print(f"\n=== 统计 ===")
stats = engine.get_stats()
for k, v in stats.items():
    print(f"  {k}: {v}")

常见问题FAQ

自动化测试Agent最佳实践

1. 理解原理再实践 - 先搞清楚为什么再动手实现
2. 渐进式复杂化 - 先让最简版本跑通再逐步优化
3. 错误处理优先 - 假设一切都会失败提前做好准备
4. 可观测性从Day1 - 不要等出问题才加监控
5. 文档即代码 - 好的文档和好的代码一样重要
6. 持续迭代 - 没有完美的设计只有不断改进的系统

设计格言：自动化测试Agent的核心不在于技术复杂度，而在于能否可靠地解决实际问题。简单且可靠远胜于复杂但不稳定。