zyc
e9ba36db92
All checks were successful
Build and Deploy Log Center / build-and-deploy (push) Successful in 1m31s
318 lines
13 KiB
Python
318 lines
13 KiB
Python
"""
|
||
Repair Engine 核心逻辑测试 - 测试 3 轮修复重试机制和失败报告上传
|
||
"""
|
||
import pytest
|
||
from unittest.mock import MagicMock, patch, call
|
||
from repair_agent.models import Bug, BugStatus, ErrorInfo, FixResult, BatchFixResult, RepairReport
|
||
from repair_agent.agent.core import RepairEngine
|
||
from repair_agent.agent.test_runner import TestResult
|
||
|
||
|
||
def make_bug(bug_id=1, project_id="rtc_backend"):
|
||
"""创建测试用 Bug 对象"""
|
||
return Bug(
|
||
id=bug_id,
|
||
project_id=project_id,
|
||
environment="production",
|
||
level="ERROR",
|
||
error=ErrorInfo(
|
||
type="ValueError",
|
||
message="test error",
|
||
file_path="app/views.py",
|
||
line_number=42,
|
||
stack_trace=["traceback line 1"],
|
||
),
|
||
status=BugStatus.NEW,
|
||
retry_count=0,
|
||
)
|
||
|
||
|
||
class TestRepairEngineRetryLoop:
|
||
"""测试修复引擎的多轮重试逻辑"""
|
||
|
||
def setup_method(self):
|
||
"""每个测试前初始化"""
|
||
self.engine = RepairEngine.__new__(RepairEngine)
|
||
self.engine.task_manager = MagicMock()
|
||
self.engine.claude_service = MagicMock()
|
||
|
||
@patch("repair_agent.agent.core.settings")
|
||
@patch("repair_agent.agent.core.TestRunner")
|
||
@patch("repair_agent.agent.core.GitManager")
|
||
def test_fix_project_success_first_round(self, MockGitManager, MockTestRunner, mock_settings):
|
||
"""测试第 1 轮就修复成功"""
|
||
mock_settings.get_project_path.return_value = "/tmp/project"
|
||
mock_settings.is_git_enabled.return_value = False
|
||
mock_settings.max_retry_count = 3
|
||
|
||
bug = make_bug()
|
||
self.engine.task_manager.fetch_pending_bugs.return_value = [bug]
|
||
self.engine.claude_service.batch_fix_bugs.return_value = (True, "fixed output")
|
||
|
||
mock_runner = MagicMock()
|
||
mock_runner.run_full_suite.return_value = TestResult(
|
||
success=True, output="OK", failed_tests=[], passed_count=5, failed_count=0
|
||
)
|
||
MockTestRunner.return_value = mock_runner
|
||
|
||
result = self.engine.fix_project("rtc_backend", run_tests=True)
|
||
|
||
assert result.success_count == 1
|
||
assert result.failed_count == 0
|
||
assert "修复成功 (第 1 轮)" in result.results[0].message
|
||
|
||
# Claude 只调用了一次
|
||
self.engine.claude_service.batch_fix_bugs.assert_called_once()
|
||
self.engine.claude_service.retry_fix_bugs.assert_not_called()
|
||
|
||
# 状态更新：FIXING → FIXED
|
||
self.engine.task_manager.update_status.assert_any_call(bug.id, BugStatus.FIXING)
|
||
self.engine.task_manager.update_status.assert_any_call(bug.id, BugStatus.FIXED)
|
||
|
||
# 上传了成功报告
|
||
self.engine.task_manager.upload_report.assert_called_once()
|
||
uploaded = self.engine.task_manager.upload_report.call_args[0][0]
|
||
assert uploaded.repair_round == 1
|
||
assert uploaded.failure_reason is None
|
||
assert uploaded.status == BugStatus.FIXED
|
||
|
||
@patch("repair_agent.agent.core.settings")
|
||
@patch("repair_agent.agent.core.TestRunner")
|
||
@patch("repair_agent.agent.core.GitManager")
|
||
def test_fix_project_success_second_round(self, MockGitManager, MockTestRunner, mock_settings):
|
||
"""测试第 1 轮失败，第 2 轮修复成功"""
|
||
mock_settings.get_project_path.return_value = "/tmp/project"
|
||
mock_settings.is_git_enabled.return_value = False
|
||
mock_settings.max_retry_count = 3
|
||
|
||
bug = make_bug()
|
||
self.engine.task_manager.fetch_pending_bugs.return_value = [bug]
|
||
self.engine.claude_service.batch_fix_bugs.return_value = (True, "round 1 output")
|
||
self.engine.claude_service.retry_fix_bugs.return_value = (True, "round 2 output")
|
||
|
||
fail_result = TestResult(
|
||
success=False, output="FAILED: test_foo",
|
||
failed_tests=["test_foo"], passed_count=4, failed_count=1,
|
||
)
|
||
pass_result = TestResult(
|
||
success=True, output="OK", failed_tests=[], passed_count=5, failed_count=0,
|
||
)
|
||
|
||
mock_runner = MagicMock()
|
||
mock_runner.run_full_suite.side_effect = [fail_result, pass_result]
|
||
MockTestRunner.return_value = mock_runner
|
||
|
||
result = self.engine.fix_project("rtc_backend", run_tests=True)
|
||
|
||
assert result.success_count == 1
|
||
assert result.failed_count == 0
|
||
assert "修复成功 (第 2 轮)" in result.results[0].message
|
||
|
||
# Claude: batch_fix_bugs 第1轮, retry_fix_bugs 第2轮
|
||
self.engine.claude_service.batch_fix_bugs.assert_called_once()
|
||
self.engine.claude_service.retry_fix_bugs.assert_called_once()
|
||
retry_args = self.engine.claude_service.retry_fix_bugs.call_args
|
||
assert retry_args.kwargs["round_num"] == 2
|
||
assert "FAILED: test_foo" in retry_args.kwargs["previous_test_output"]
|
||
|
||
# 上传了 2 次报告：第1轮失败 + 第2轮成功
|
||
assert self.engine.task_manager.upload_report.call_count == 2
|
||
reports = [c[0][0] for c in self.engine.task_manager.upload_report.call_args_list]
|
||
assert reports[0].repair_round == 1
|
||
assert reports[0].failure_reason is not None
|
||
assert reports[0].status == BugStatus.FIXING # 中间轮
|
||
assert reports[1].repair_round == 2
|
||
assert reports[1].failure_reason is None
|
||
assert reports[1].status == BugStatus.FIXED
|
||
|
||
@patch("repair_agent.agent.core.settings")
|
||
@patch("repair_agent.agent.core.TestRunner")
|
||
@patch("repair_agent.agent.core.GitManager")
|
||
def test_fix_project_all_rounds_fail(self, MockGitManager, MockTestRunner, mock_settings):
|
||
"""测试 3 轮都失败"""
|
||
mock_settings.get_project_path.return_value = "/tmp/project"
|
||
mock_settings.is_git_enabled.return_value = False
|
||
mock_settings.max_retry_count = 3
|
||
|
||
bug = make_bug()
|
||
self.engine.task_manager.fetch_pending_bugs.return_value = [bug]
|
||
self.engine.claude_service.batch_fix_bugs.return_value = (True, "round 1")
|
||
self.engine.claude_service.retry_fix_bugs.side_effect = [
|
||
(True, "round 2"),
|
||
(True, "round 3"),
|
||
]
|
||
|
||
fail_result = TestResult(
|
||
success=False, output="FAILED",
|
||
failed_tests=["test_foo"], passed_count=0, failed_count=1,
|
||
)
|
||
mock_runner = MagicMock()
|
||
mock_runner.run_full_suite.return_value = fail_result
|
||
MockTestRunner.return_value = mock_runner
|
||
|
||
result = self.engine.fix_project("rtc_backend", run_tests=True)
|
||
|
||
assert result.success_count == 0
|
||
assert result.failed_count == 1
|
||
assert "3 轮修复仍未通过测试" in result.results[0].message
|
||
|
||
# 上传了 3 次报告
|
||
assert self.engine.task_manager.upload_report.call_count == 3
|
||
reports = [c[0][0] for c in self.engine.task_manager.upload_report.call_args_list]
|
||
|
||
assert reports[0].repair_round == 1
|
||
assert reports[0].status == BugStatus.FIXING
|
||
assert reports[1].repair_round == 2
|
||
assert reports[1].status == BugStatus.FIXING
|
||
assert reports[2].repair_round == 3
|
||
assert reports[2].status == BugStatus.FIX_FAILED
|
||
|
||
# 最终状态为 FIX_FAILED
|
||
final_call = self.engine.task_manager.update_status.call_args_list[-1]
|
||
assert final_call[0][1] == BugStatus.FIX_FAILED
|
||
|
||
@patch("repair_agent.agent.core.settings")
|
||
@patch("repair_agent.agent.core.GitManager")
|
||
def test_fix_project_claude_cli_failure(self, MockGitManager, mock_settings):
|
||
"""测试 Claude CLI 执行失败不重试"""
|
||
mock_settings.get_project_path.return_value = "/tmp/project"
|
||
mock_settings.is_git_enabled.return_value = False
|
||
mock_settings.max_retry_count = 3
|
||
|
||
bug = make_bug()
|
||
self.engine.task_manager.fetch_pending_bugs.return_value = [bug]
|
||
self.engine.claude_service.batch_fix_bugs.return_value = (False, "timeout error")
|
||
|
||
result = self.engine.fix_project("rtc_backend", run_tests=True)
|
||
|
||
assert result.success_count == 0
|
||
assert result.failed_count == 1
|
||
assert "Claude CLI 执行失败" in result.results[0].message
|
||
|
||
# 只调用了第1轮，没有重试
|
||
self.engine.claude_service.batch_fix_bugs.assert_called_once()
|
||
self.engine.claude_service.retry_fix_bugs.assert_not_called()
|
||
|
||
# 上传了失败报告
|
||
self.engine.task_manager.upload_report.assert_called_once()
|
||
uploaded = self.engine.task_manager.upload_report.call_args[0][0]
|
||
assert uploaded.failure_reason.startswith("Claude CLI 执行失败")
|
||
assert uploaded.repair_round == 1
|
||
|
||
|
||
class TestFixSingleBugRetry:
|
||
"""测试单个 Bug 修复的多轮重试"""
|
||
|
||
def setup_method(self):
|
||
self.engine = RepairEngine.__new__(RepairEngine)
|
||
self.engine.task_manager = MagicMock()
|
||
self.engine.claude_service = MagicMock()
|
||
|
||
@patch("repair_agent.agent.core.settings")
|
||
@patch("repair_agent.agent.core.TestRunner")
|
||
@patch("repair_agent.agent.core.GitManager")
|
||
def test_single_bug_success_third_round(self, MockGitManager, MockTestRunner, mock_settings):
|
||
"""测试单个 Bug 第 3 轮修复成功"""
|
||
mock_settings.get_project_path.return_value = "/tmp/project"
|
||
mock_settings.max_retry_count = 3
|
||
|
||
bug = make_bug()
|
||
self.engine.task_manager.get_bug_detail.return_value = bug
|
||
|
||
mock_git = MagicMock()
|
||
mock_git.get_modified_files.return_value = ["file.py"]
|
||
mock_git.get_diff.return_value = "+fix"
|
||
MockGitManager.return_value = mock_git
|
||
|
||
self.engine.claude_service.batch_fix_bugs.return_value = (True, "r1")
|
||
self.engine.claude_service.retry_fix_bugs.side_effect = [
|
||
(True, "r2"),
|
||
(True, "r3"),
|
||
]
|
||
|
||
fail_result = TestResult(success=False, output="FAIL", failed_tests=["t1"], passed_count=0, failed_count=1)
|
||
pass_result = TestResult(success=True, output="OK", failed_tests=[], passed_count=5, failed_count=0)
|
||
|
||
mock_runner = MagicMock()
|
||
mock_runner.run_full_suite.side_effect = [fail_result, fail_result, pass_result]
|
||
MockTestRunner.return_value = mock_runner
|
||
|
||
result = self.engine.fix_single_bug(1, run_tests=True)
|
||
|
||
assert result.success is True
|
||
assert "修复成功 (第 3 轮)" in result.message
|
||
|
||
# 3 次报告上传
|
||
assert self.engine.task_manager.upload_report.call_count == 3
|
||
|
||
@patch("repair_agent.agent.core.settings")
|
||
@patch("repair_agent.agent.core.TestRunner")
|
||
@patch("repair_agent.agent.core.GitManager")
|
||
def test_single_bug_all_fail(self, MockGitManager, MockTestRunner, mock_settings):
|
||
"""测试单个 Bug 3 轮全部失败"""
|
||
mock_settings.get_project_path.return_value = "/tmp/project"
|
||
mock_settings.max_retry_count = 3
|
||
|
||
bug = make_bug()
|
||
self.engine.task_manager.get_bug_detail.return_value = bug
|
||
|
||
mock_git = MagicMock()
|
||
mock_git.get_modified_files.return_value = ["file.py"]
|
||
mock_git.get_diff.return_value = "+fix"
|
||
MockGitManager.return_value = mock_git
|
||
|
||
self.engine.claude_service.batch_fix_bugs.return_value = (True, "r1")
|
||
self.engine.claude_service.retry_fix_bugs.return_value = (True, "retry output")
|
||
|
||
fail_result = TestResult(success=False, output="FAIL", failed_tests=["t1"], passed_count=0, failed_count=1)
|
||
mock_runner = MagicMock()
|
||
mock_runner.run_full_suite.return_value = fail_result
|
||
MockTestRunner.return_value = mock_runner
|
||
|
||
result = self.engine.fix_single_bug(1, run_tests=True)
|
||
|
||
assert result.success is False
|
||
assert "3 轮修复仍未通过测试" in result.message
|
||
assert self.engine.task_manager.upload_report.call_count == 3
|
||
|
||
|
||
class TestRepairReportModel:
|
||
"""测试 RepairReport 模型新字段"""
|
||
|
||
def test_repair_report_with_failure(self):
|
||
"""测试带失败原因的修复报告"""
|
||
report = RepairReport(
|
||
error_log_id=1,
|
||
status=BugStatus.FIX_FAILED,
|
||
project_id="rtc_backend",
|
||
ai_analysis="analysis",
|
||
fix_plan="plan",
|
||
code_diff="diff",
|
||
modified_files=["a.py"],
|
||
test_output="FAILED",
|
||
test_passed=False,
|
||
repair_round=2,
|
||
failure_reason="测试未通过",
|
||
)
|
||
data = report.model_dump()
|
||
assert data["repair_round"] == 2
|
||
assert data["failure_reason"] == "测试未通过"
|
||
assert data["test_passed"] is False
|
||
|
||
def test_repair_report_success_defaults(self):
|
||
"""测试成功报告默认值"""
|
||
report = RepairReport(
|
||
error_log_id=1,
|
||
status=BugStatus.FIXED,
|
||
project_id="rtc_backend",
|
||
ai_analysis="fixed",
|
||
fix_plan="plan",
|
||
code_diff="diff",
|
||
modified_files=[],
|
||
test_output="OK",
|
||
test_passed=True,
|
||
)
|
||
data = report.model_dump()
|
||
assert data["repair_round"] == 1 # 默认
|
||
assert data["failure_reason"] is None # 默认
|