基于 OpenClaw + Claude Code 的端到端研发自动化系统核心模块
本系统是《基于 OpenClaw + Claude Code 的端到端研发自动化系统》的核心组成部分, 专注于单元测试用例的在线补充与人机协同优化。系统融合了先进的 AI 代码生成技术 与人类专家的审核决策能力,实现了从测试用例自动生成、人工审核、执行验证到覆盖率优化的完整闭环。
系统采用模块化设计,包含以下核心组件:
class AITestGenerator:
"""AI 测试用例生成器 - 模拟 Claude Code/OpenClaw 的测试生成能力"""
def analyze_source_code(self, source_code: str, file_path: str) -> Dict:
"""分析源代码结构,识别可测试的函数和类"""
# 使用 AST 解析 Python 源代码
# 提取函数名、参数、装饰器、文档字符串等信息
def generate_test_template(self, func_info: Dict) -> str:
"""为函数生成测试模板"""
# 根据函数签名自动生成测试框架
# 包含 Arrange-Act-Assert 三段式结构
def generate_tests_for_file(self, source_code: str) -> List[TestAction]:
"""为整个文件生成测试用例列表"""
# 批量生成测试并封装为 TestAction 对象
class CoverageAnalyzer:
"""代码覆盖率分析器"""
def analyze_coverage(self, test_file: str, source_file: str) -> Dict:
"""运行测试并分析覆盖率"""
# 使用 coverage.py 库进行覆盖率收集
# 生成覆盖率报告和缺失行信息
def identify_gaps(self, source_code: str) -> List[CoverageGap]:
"""识别代码覆盖率缺口"""
# 分析 AST 识别多分支路径
# 标记未覆盖的代码区域
class HumanInTheLoopMiddleware:
"""人机协同中间件 - 实现 approve/edit/reject 决策机制"""
def create_interrupt_request(self, test_actions: List[TestAction]) -> HITLRequest:
"""创建中断请求,等待人工审核"""
# 生成唯一的 request_id
# 配置每个动作的允许决策类型
def process_decision(self, request_id: str, decisions: List[Dict]) -> HITLResponse:
"""处理人工决策"""
# 验证决策数量与动作数量匹配
# 更新测试动作状态
# 应用编辑操作(如果有)
class TestExecutionEngine:
"""测试执行引擎"""
def execute_test(self, test_action: TestAction, workspace_dir: str) -> Dict:
"""执行单个测试"""
# 创建临时测试文件
# 调用 pytest 运行测试
# 捕获输出和结果
def execute_batch(self, test_actions: List[TestAction]) -> List[Dict]:
"""批量执行测试"""
# 过滤已批准的测试
# 并行/串行执行
# 聚合结果
自动分析源代码
生成测试用例
Approve/Edit/Reject
人机协同决策
运行 pytest
收集执行结果
分析覆盖情况
识别缺口
针对缺口生成
补充测试用例
系统使用 AST 解析源代码,识别所有公共函数和类方法,为每个可测试单元生成符合 pytest 规范的测试模板。 测试模板遵循 Arrange-Act-Assert模式,包含合理的默认输入值和断言框架。
生成的测试用例会触发人机协同中断请求。审核人员可以对每个测试用例做出三种决策: Approve(批准)直接执行、Edit(编辑)修改后执行、 Reject(拒绝)并提供反馈理由。所有决策会被记录并用于后续优化。
系统将所有状态为 approved 或 edited 的测试用例写入临时文件,使用 pytest 框架批量执行。 执行结果包括成功/失败状态、输出日志、执行时间等详细信息。
利用 coverage.py 库分析测试执行后的代码覆盖情况,识别未被覆盖的行和分支。 系统会进一步分析 AST,找出包含多分支逻辑但测试不足的函数,标记为覆盖率缺口。
针对识别出的覆盖率缺口,系统自动生成针对性的补充测试用例,重点关注边界条件、异常情况和边缘案例。 补充测试同样需要经过人工审核流程,形成闭环优化。
基于 AST 的源代码解析,自动识别函数、类、方法及其参数、装饰器、文档字符串等元数据
根据参数类型推断合适的测试数据,支持 int、str、list、dict 等多种数据类型的智能填充
Approve/Edit/Reject三种决策模式,支持批量审批和个性化反馈
每个测试用例都有完整的生命周期追踪:pending → approved/rejected → executed → passed/failed
基于覆盖率分析结果自动生成补充测试,持续迭代直到达到目标准确率
原生支持 pytest 测试框架,兼容现有测试基础设施和 CI/CD 流程
支持将完整系统状态导出为 JSON 格式,便于审计、复现和版本控制
AI 科幻量子炫彩风格的 Web 界面,实时展示测试状态、覆盖率和执行结果
人机协同(Human-in-the-Loop, HITL)是本系统的核心创新点。通过引入人工审核环节, 系统能够在保持 AI 高效生成的同时,确保测试用例的质量和适用性。
| 决策类型 | 图标 | 描述 | 适用场景 |
|---|---|---|---|
approve |
✅ | 批准执行 | 测试用例完全符合预期,无需修改 |
edit |
✏️ | 编辑后执行 | 测试逻辑正确但需要调整输入值、断言条件或添加注释 |
reject |
❌ | 拒绝并提供反馈 | 测试用例存在严重问题或当前优先级较低 |
@dataclass
class HITLRequest:
"""人机协同请求"""
request_id: str # 唯一请求标识
action_requests: List[TestAction] # 待审核的测试动作列表
review_configs: List[Dict] # 每个动作的审核配置
created_at: str # 创建时间戳
@dataclass
class HITLResponse:
"""人机协同响应"""
request_id: str # 对应的请求 ID
decisions: List[Dict] # 决策列表
submitted_at: str # 提交时间戳
代码覆盖率是衡量测试质量的重要指标。本系统使用业界标准的 coverage.py 库进行覆盖率收集和分析。
| 指标名称 | 说明 | 计算公式 |
|---|---|---|
| 语句覆盖率 | 已执行的语句占总语句的比例 | 执行语句数 / 总语句数 × 100% |
| 分支覆盖率 | 已覆盖的分支占全部分支的比例 | 覆盖分支数 / 总分支数 × 100% |
| 函数覆盖率 | 被调用的函数占总函数的比例 | 被调用函数数 / 总函数数 × 100% |
| 行覆盖率 | 已执行的代码行占总代码行的比例 | 执行行数 / 总行数 × 100% |
def identify_gaps(self, source_code: str, file_path: str) -> List[CoverageGap]:
"""识别代码覆盖率缺口"""
tree = ast.parse(source_code)
gaps = []
for node in ast.walk(tree):
if isinstance(node, ast.FunctionDef):
# 检查是否有多个分支路径
has_multiple_paths = False
for child in ast.walk(node):
if isinstance(child, (ast.If, ast.For, ast.While, ast.Try)):
has_multiple_paths = True
break
if has_multiple_paths:
gaps.append(CoverageGap(
file_path=file_path,
line_number=node.lineno,
function_name=node.name,
missing_code=ast.unparse(node),
suggested_test=f"Suggest adding edge case tests for {node.name}"
))
return gaps
以下是一个完整的系统运行示例,展示了从源代码到测试生成的全流程:
"""Sample module for unit test generation demo"""
def add(a, b):
"""Add two numbers"""
return a + b
def divide(a, b):
"""Divide a by b with error handling"""
if b == 0:
raise ValueError("Cannot divide by zero")
return a / b
def process_list(items):
"""Process a list of items"""
if not items:
return []
result = []
for item in items:
if isinstance(item, (int, float)):
result.append(item * 2)
elif isinstance(item, str):
result.append(item.upper())
else:
result.append(str(item))
return result
class Calculator:
"""Simple calculator class"""
def __init__(self, initial_value=0):
self.value = initial_value
def add(self, amount):
"""Add amount to current value"""
self.value += amount
return self.value
def subtract(self, amount):
"""Subtract amount from current value"""
self.value -= amount
return self.value
def multiply(self, factor):
"""Multiply current value by factor"""
self.value *= factor
return self.value
def reset(self):
"""Reset calculator to zero"""
self.value = 0
return self.value
# 测试用例 1: test_add
def test_add():
"""Test case for add
Docstring: Add two numbers
"""
# Arrange - Setup test data
result = add(1, 2)
# Assert - Verify expected behavior
assert result == 3, "1 + 2 should equal 3"
print("✓ test_add executed successfully")
# 测试用例 2: test_divide_by_zero (补充测试)
def test_divide_by_zero():
"""Test edge case: division by zero"""
try:
divide(10, 0)
assert False, "Should raise ValueError"
except ValueError as e:
assert str(e) == "Cannot divide by zero"
print("✓ test_divide_by_zero executed successfully")
# 测试用例 3: test_Calculator_add
def test_Calculator_add():
"""Test Calculator.add method"""
# Arrange
calc = Calculator(initial_value=10)
# Act
result = calc.add(5)
# Assert
assert result == 15
assert calc.value == 15
print("✓ test_Calculator_add executed successfully")
============================================================
🤖 Step 1: AI Generating Initial Test Cases
============================================================
✓ Generated 11 test cases
⏳ Created HITL Request: hitl_20260314_035330_0
Waiting for human review...
============================================================
👤 Step 2: Human Review & Decision
============================================================
✓ Review completed:
Approved: 6
Edited: 4
Rejected: 1
============================================================
▶️ Step 3: Executing Approved Tests
============================================================
============================================================
📊 Step 4: Coverage Analysis & Gap Identification
============================================================
✓ Identified 2 coverage gaps
Gap 1: divide at line 8
Gap 2: process_list at line 14
============================================================
🤖 Step 5: Generating Supplementary Tests for Gaps
============================================================
✓ Generated 2 supplementary tests
============================================================
✅ Workflow Completed!
============================================================
Total tests generated: 13
Pending review: 1
Completed reviews: 1
实现了从需求→PRD→技术方案→API 设计→AI Coding→Unit Test→集成测试→CI/CD→部署→UI 自动化验收的全流程自动化中的单元测试环节
不是简单的 AI 替代人类,而是 AI 与人类专家的优势互补,AI 负责重复性工作,人类负责关键决策
基于覆盖率数据的反馈循环,系统能够自我完善,不断补充遗漏的测试场景
模块化架构设计,易于集成到现有的 Jenkins + Docker + K8S (KubeSphere) DevOps 工具链
集成更强大的语言模型(如 Claude 3.5、GPT-4o),提升测试用例的智能程度和准确性; 增加对异步代码、并发测试、性能测试的支持。
支持多人协同审核、评论、批注;建立测试用例知识库,积累最佳实践; 与项目管理工具(Jira、TAPD)深度集成。
基于机器学习预测哪些代码变更最容易引入 bug,优先生成相关测试; 实现测试用例的自动优先级排序和执行调度。