551 lines
21 KiB
Python
551 lines
21 KiB
Python
"""Tests for QAAgent."""
|
|
|
|
import os
|
|
import subprocess
|
|
from types import SimpleNamespace
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import git as gitmod
|
|
import pytest
|
|
|
|
from app_factory.agents.qa_agent import QAAgent
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _make_review_response(text, input_tokens=10, output_tokens=20):
|
|
"""Build a fake Claude SDK completion response."""
|
|
return SimpleNamespace(text=text, input_tokens=input_tokens, output_tokens=output_tokens)
|
|
|
|
|
|
def _build_agent(repo_path="/fake/repo", **kwargs):
|
|
"""Create a QAAgent with mocked git.Repo and Claude SDK client."""
|
|
with patch("app_factory.agents.qa_agent.git.Repo") as mock_repo_cls, \
|
|
patch("app_factory.agents.qa_agent.ClaudeSDKClient") as mock_sdk_client:
|
|
mock_repo = MagicMock()
|
|
mock_repo_cls.return_value = mock_repo
|
|
mock_client = AsyncMock()
|
|
mock_sdk_client.return_value = mock_client
|
|
agent = QAAgent(repo_path=repo_path, api_key="test-key", **kwargs)
|
|
agent.client = mock_client
|
|
agent.repo = mock_repo
|
|
return agent, mock_repo, mock_client
|
|
|
|
|
|
APPROVED_REVIEW = """\
|
|
APPROVED: true
|
|
ISSUES:
|
|
- [severity: info] Minor style suggestion
|
|
SUMMARY: Code looks good overall."""
|
|
|
|
REJECTED_REVIEW = """\
|
|
APPROVED: false
|
|
ISSUES:
|
|
- [severity: critical] SQL injection in query builder
|
|
- [severity: warning] Missing input validation
|
|
SUMMARY: Critical security issue found."""
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Initialization
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestInitialization:
|
|
def test_no_api_key_uses_default_client(self):
|
|
with patch("app_factory.agents.qa_agent.git.Repo"), \
|
|
patch("app_factory.agents.qa_agent.ClaudeSDKClient") as mock_sdk_client, \
|
|
patch.dict(os.environ, {}, clear=True):
|
|
env = os.environ.copy()
|
|
env.pop("ANTHROPIC_API_KEY", None)
|
|
with patch.dict(os.environ, env, clear=True):
|
|
mock_sdk_client.return_value = AsyncMock()
|
|
agent = QAAgent(repo_path="/fake")
|
|
mock_sdk_client.assert_called_once_with(
|
|
api_key=None,
|
|
auth_token=None,
|
|
enable_debug=False,
|
|
)
|
|
|
|
def test_creates_with_api_key(self):
|
|
agent, mock_repo, _ = _build_agent()
|
|
assert agent.max_retries == 3
|
|
|
|
def test_custom_max_retries(self):
|
|
agent, _, _ = _build_agent(max_retries=5)
|
|
assert agent.max_retries == 5
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Rebase
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestRebaseOntoMain:
|
|
@pytest.mark.asyncio
|
|
async def test_rebase_success(self):
|
|
agent, _, _ = _build_agent()
|
|
mock_wt_repo = MagicMock()
|
|
with patch("app_factory.agents.qa_agent.git.Repo", return_value=mock_wt_repo):
|
|
result = await agent.rebase_onto_main("/worktree/path", "task-1")
|
|
|
|
assert result["success"] is True
|
|
assert result["conflicts"] == []
|
|
mock_wt_repo.git.rebase.assert_called_once_with("main")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_rebase_conflict_unresolvable(self):
|
|
agent, _, _ = _build_agent()
|
|
mock_wt_repo = MagicMock()
|
|
mock_wt_repo.git.rebase.side_effect = gitmod.GitCommandError("rebase", "CONFLICT")
|
|
mock_wt_repo.git.status.return_value = "UU conflicted_file.py"
|
|
|
|
with patch("app_factory.agents.qa_agent.git.Repo", return_value=mock_wt_repo), \
|
|
patch.object(agent, "auto_resolve_conflicts", return_value=False):
|
|
result = await agent.rebase_onto_main("/worktree/path", "task-1")
|
|
|
|
assert result["success"] is False
|
|
assert "conflicted_file.py" in result["conflicts"]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_rebase_conflict_auto_resolved(self):
|
|
agent, _, _ = _build_agent()
|
|
mock_wt_repo = MagicMock()
|
|
mock_wt_repo.git.rebase.side_effect = gitmod.GitCommandError("rebase", "CONFLICT")
|
|
mock_wt_repo.git.status.return_value = "UU file.py"
|
|
|
|
with patch("app_factory.agents.qa_agent.git.Repo", return_value=mock_wt_repo), \
|
|
patch.object(agent, "auto_resolve_conflicts", return_value=True):
|
|
result = await agent.rebase_onto_main("/worktree/path", "task-1")
|
|
|
|
assert result["success"] is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fetch_failure_continues(self):
|
|
"""If fetch fails (no remote), rebase should still be attempted."""
|
|
agent, _, _ = _build_agent()
|
|
mock_wt_repo = MagicMock()
|
|
mock_wt_repo.git.fetch.side_effect = gitmod.GitCommandError("fetch", "No remote")
|
|
|
|
with patch("app_factory.agents.qa_agent.git.Repo", return_value=mock_wt_repo):
|
|
result = await agent.rebase_onto_main("/worktree/path", "task-1")
|
|
|
|
assert result["success"] is True
|
|
mock_wt_repo.git.rebase.assert_called_once_with("main")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Linter
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestRunLinter:
|
|
def test_lint_passes(self):
|
|
agent, _, _ = _build_agent()
|
|
mock_result = subprocess.CompletedProcess(
|
|
args=["ruff", "check", "."],
|
|
returncode=0,
|
|
stdout="All checks passed!\n",
|
|
stderr="",
|
|
)
|
|
with patch("app_factory.agents.qa_agent.subprocess.run", return_value=mock_result):
|
|
result = agent.run_linter("/worktree/path")
|
|
|
|
assert result["passed"] is True
|
|
assert result["errors"] == []
|
|
|
|
def test_lint_fails_with_errors(self):
|
|
agent, _, _ = _build_agent()
|
|
ruff_output = (
|
|
"app/main.py:10:1: E501 Line too long (120 > 88 characters)\n"
|
|
"app/main.py:15:5: F841 Local variable 'x' is assigned but never used\n"
|
|
"Found 2 errors.\n"
|
|
)
|
|
mock_result = subprocess.CompletedProcess(
|
|
args=["ruff", "check", "."],
|
|
returncode=1,
|
|
stdout=ruff_output,
|
|
stderr="",
|
|
)
|
|
with patch("app_factory.agents.qa_agent.subprocess.run", return_value=mock_result):
|
|
result = agent.run_linter("/worktree/path")
|
|
|
|
assert result["passed"] is False
|
|
assert len(result["errors"]) == 2
|
|
|
|
def test_lint_ruff_not_found(self):
|
|
agent, _, _ = _build_agent()
|
|
with patch("app_factory.agents.qa_agent.subprocess.run", side_effect=FileNotFoundError):
|
|
result = agent.run_linter("/worktree/path")
|
|
|
|
assert result["passed"] is True
|
|
assert "ruff not found" in result["warnings"][0]
|
|
|
|
def test_lint_timeout(self):
|
|
agent, _, _ = _build_agent()
|
|
with patch("app_factory.agents.qa_agent.subprocess.run",
|
|
side_effect=subprocess.TimeoutExpired(cmd="ruff", timeout=120)):
|
|
result = agent.run_linter("/worktree/path")
|
|
|
|
assert result["passed"] is False
|
|
assert "timed out" in result["errors"][0]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestRunTests:
|
|
def test_all_tests_pass(self):
|
|
agent, _, _ = _build_agent()
|
|
pytest_output = (
|
|
"tests/test_foo.py::test_one PASSED\n"
|
|
"tests/test_foo.py::test_two PASSED\n"
|
|
"========================= 2 passed =========================\n"
|
|
)
|
|
mock_result = subprocess.CompletedProcess(
|
|
args=["python", "-m", "pytest"],
|
|
returncode=0,
|
|
stdout=pytest_output,
|
|
stderr="",
|
|
)
|
|
with patch("app_factory.agents.qa_agent.subprocess.run", return_value=mock_result):
|
|
result = agent.run_tests("/worktree/path")
|
|
|
|
assert result["passed"] is True
|
|
assert result["total"] == 2
|
|
assert result["failures"] == 0
|
|
assert result["errors"] == 0
|
|
|
|
def test_some_tests_fail(self):
|
|
agent, _, _ = _build_agent()
|
|
pytest_output = (
|
|
"tests/test_foo.py::test_one PASSED\n"
|
|
"tests/test_foo.py::test_two FAILED\n"
|
|
"=================== 1 failed, 1 passed ====================\n"
|
|
)
|
|
mock_result = subprocess.CompletedProcess(
|
|
args=["python", "-m", "pytest"],
|
|
returncode=1,
|
|
stdout=pytest_output,
|
|
stderr="",
|
|
)
|
|
with patch("app_factory.agents.qa_agent.subprocess.run", return_value=mock_result):
|
|
result = agent.run_tests("/worktree/path")
|
|
|
|
assert result["passed"] is False
|
|
assert result["total"] == 2
|
|
assert result["failures"] == 1
|
|
|
|
def test_pytest_not_found(self):
|
|
agent, _, _ = _build_agent()
|
|
with patch("app_factory.agents.qa_agent.subprocess.run", side_effect=FileNotFoundError):
|
|
result = agent.run_tests("/worktree/path")
|
|
|
|
assert result["passed"] is False
|
|
assert "pytest not found" in result["output"]
|
|
|
|
def test_pytest_timeout(self):
|
|
agent, _, _ = _build_agent()
|
|
with patch("app_factory.agents.qa_agent.subprocess.run",
|
|
side_effect=subprocess.TimeoutExpired(cmd="pytest", timeout=300)):
|
|
result = agent.run_tests("/worktree/path")
|
|
|
|
assert result["passed"] is False
|
|
assert "timed out" in result["output"]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Code Review
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestCodeReview:
|
|
@pytest.mark.asyncio
|
|
async def test_review_approved(self):
|
|
agent, _, mock_client = _build_agent()
|
|
mock_client.complete = AsyncMock(
|
|
return_value=_make_review_response(APPROVED_REVIEW)
|
|
)
|
|
|
|
result = await agent.code_review("diff content", task={"id": "1", "title": "Add feature"})
|
|
|
|
assert result["approved"] is True
|
|
assert len(result["issues"]) == 1
|
|
assert result["issues"][0]["severity"] == "info"
|
|
assert result["summary"] != ""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_review_rejected(self):
|
|
agent, _, mock_client = _build_agent()
|
|
mock_client.complete = AsyncMock(
|
|
return_value=_make_review_response(REJECTED_REVIEW)
|
|
)
|
|
|
|
result = await agent.code_review("diff with issues")
|
|
|
|
assert result["approved"] is False
|
|
assert len(result["issues"]) == 2
|
|
assert result["issues"][0]["severity"] == "critical"
|
|
assert "security" in result["summary"].lower()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_review_no_task_context(self):
|
|
agent, _, mock_client = _build_agent()
|
|
mock_client.complete = AsyncMock(
|
|
return_value=_make_review_response(APPROVED_REVIEW)
|
|
)
|
|
|
|
result = await agent.code_review("diff content", task=None)
|
|
|
|
assert result["approved"] is True
|
|
mock_client.complete.assert_awaited_once()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_review_loads_template(self):
|
|
agent, _, mock_client = _build_agent()
|
|
mock_client.complete = AsyncMock(
|
|
return_value=_make_review_response(APPROVED_REVIEW)
|
|
)
|
|
|
|
await agent.code_review("some diff")
|
|
|
|
call_args = mock_client.complete.call_args
|
|
prompt_text = call_args.kwargs["prompt"]
|
|
assert "Review Checklist" in prompt_text
|
|
assert "OWASP" in prompt_text
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Merge to main
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestMergeToMain:
|
|
def test_merge_success(self):
|
|
agent, mock_repo, _ = _build_agent()
|
|
mock_repo.head.commit.hexsha = "abc123def456"
|
|
|
|
result = agent.merge_to_main("/worktree/path", "42")
|
|
|
|
assert result["success"] is True
|
|
assert result["commit_sha"] == "abc123def456"
|
|
mock_repo.git.checkout.assert_called_once_with("main")
|
|
mock_repo.git.merge.assert_called_once_with(
|
|
"--no-ff", "feature/task-42", m="Merge feature/task-42"
|
|
)
|
|
|
|
def test_merge_failure(self):
|
|
agent, mock_repo, _ = _build_agent()
|
|
mock_repo.git.merge.side_effect = gitmod.GitCommandError("merge", "conflict")
|
|
|
|
result = agent.merge_to_main("/worktree/path", "42")
|
|
|
|
assert result["success"] is False
|
|
assert result["commit_sha"] is None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Full pipeline: review_and_merge
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestReviewAndMerge:
|
|
@pytest.mark.asyncio
|
|
async def test_happy_path(self):
|
|
agent, mock_repo, mock_client = _build_agent()
|
|
mock_repo.head.commit.hexsha = "merged123"
|
|
|
|
mock_wt_repo = MagicMock()
|
|
mock_wt_repo.git.diff.return_value = "diff --git a/file.py"
|
|
|
|
with patch.object(agent, "rebase_onto_main", new_callable=AsyncMock,
|
|
return_value={"success": True, "conflicts": []}), \
|
|
patch.object(agent, "run_linter",
|
|
return_value={"passed": True, "errors": [], "warnings": []}), \
|
|
patch.object(agent, "run_tests",
|
|
return_value={"passed": True, "total": 5, "failures": 0, "errors": 0, "output": "ok"}), \
|
|
patch.object(agent, "code_review", new_callable=AsyncMock,
|
|
return_value={"approved": True, "issues": [], "summary": "All good"}), \
|
|
patch.object(agent, "merge_to_main",
|
|
return_value={"success": True, "commit_sha": "merged123"}), \
|
|
patch("app_factory.agents.qa_agent.git.Repo", return_value=mock_wt_repo):
|
|
|
|
result = await agent.review_and_merge("task-1", "/worktree/path")
|
|
|
|
assert result["status"] == "merged"
|
|
assert result["commit_sha"] == "merged123"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_rebase_failure(self):
|
|
agent, _, _ = _build_agent()
|
|
|
|
with patch.object(agent, "rebase_onto_main", new_callable=AsyncMock,
|
|
return_value={"success": False, "conflicts": ["file.py"]}):
|
|
result = await agent.review_and_merge("task-2", "/worktree/path")
|
|
|
|
assert result["status"] == "rebase_failed"
|
|
assert "file.py" in result["conflicts"]
|
|
assert result["retry_count"] == 1
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_lint_failure(self):
|
|
agent, _, _ = _build_agent()
|
|
|
|
with patch.object(agent, "rebase_onto_main", new_callable=AsyncMock,
|
|
return_value={"success": True, "conflicts": []}), \
|
|
patch.object(agent, "run_linter",
|
|
return_value={"passed": False, "errors": ["E501 line too long"], "warnings": []}):
|
|
result = await agent.review_and_merge("task-3", "/worktree/path")
|
|
|
|
assert result["status"] == "lint_failed"
|
|
assert len(result["errors"]) == 1
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_test_failure(self):
|
|
agent, _, _ = _build_agent()
|
|
mock_wt_repo = MagicMock()
|
|
|
|
with patch.object(agent, "rebase_onto_main", new_callable=AsyncMock,
|
|
return_value={"success": True, "conflicts": []}), \
|
|
patch.object(agent, "run_linter",
|
|
return_value={"passed": True, "errors": [], "warnings": []}), \
|
|
patch.object(agent, "run_tests",
|
|
return_value={"passed": False, "total": 3, "failures": 1, "errors": 0, "output": "FAILED"}):
|
|
result = await agent.review_and_merge("task-4", "/worktree/path")
|
|
|
|
assert result["status"] == "tests_failed"
|
|
assert result["failures"] == 1
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_review_failure(self):
|
|
agent, _, _ = _build_agent()
|
|
mock_wt_repo = MagicMock()
|
|
mock_wt_repo.git.diff.return_value = "diff"
|
|
|
|
with patch.object(agent, "rebase_onto_main", new_callable=AsyncMock,
|
|
return_value={"success": True, "conflicts": []}), \
|
|
patch.object(agent, "run_linter",
|
|
return_value={"passed": True, "errors": [], "warnings": []}), \
|
|
patch.object(agent, "run_tests",
|
|
return_value={"passed": True, "total": 3, "failures": 0, "errors": 0, "output": "ok"}), \
|
|
patch.object(agent, "code_review", new_callable=AsyncMock,
|
|
return_value={"approved": False, "issues": [{"severity": "critical", "description": "vuln"}], "summary": "Bad"}), \
|
|
patch("app_factory.agents.qa_agent.git.Repo", return_value=mock_wt_repo):
|
|
result = await agent.review_and_merge("task-5", "/worktree/path")
|
|
|
|
assert result["status"] == "review_failed"
|
|
assert len(result["issues"]) == 1
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Parse test results
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestParseTestResults:
|
|
def test_all_passed(self):
|
|
agent, _, _ = _build_agent()
|
|
output = "========================= 5 passed =========================\n"
|
|
result = agent.parse_test_results(output)
|
|
assert result["passed"] is True
|
|
assert result["total"] == 5
|
|
assert result["failures"] == 0
|
|
assert result["errors"] == 0
|
|
|
|
def test_mixed_results(self):
|
|
agent, _, _ = _build_agent()
|
|
output = "================ 1 failed, 4 passed, 1 error ================\n"
|
|
result = agent.parse_test_results(output)
|
|
assert result["passed"] is False
|
|
assert result["total"] == 6
|
|
assert result["failures"] == 1
|
|
assert result["errors"] == 1
|
|
|
|
def test_all_failed(self):
|
|
agent, _, _ = _build_agent()
|
|
output = "========================= 3 failed =========================\n"
|
|
result = agent.parse_test_results(output)
|
|
assert result["passed"] is False
|
|
assert result["total"] == 3
|
|
assert result["failures"] == 3
|
|
|
|
def test_no_tests(self):
|
|
agent, _, _ = _build_agent()
|
|
output = "no tests ran\n"
|
|
result = agent.parse_test_results(output)
|
|
assert result["passed"] is False
|
|
assert result["total"] == 0
|
|
|
|
def test_errors_only(self):
|
|
agent, _, _ = _build_agent()
|
|
output = "========================= 2 error =========================\n"
|
|
result = agent.parse_test_results(output)
|
|
assert result["passed"] is False
|
|
assert result["errors"] == 2
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Retry counter
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestRetryCounter:
|
|
def test_initial_count_zero(self):
|
|
agent, _, _ = _build_agent()
|
|
assert agent.get_retry_count("task-1") == 0
|
|
|
|
def test_increment_and_get(self):
|
|
agent, _, _ = _build_agent()
|
|
agent._increment_retry("task-1")
|
|
assert agent.get_retry_count("task-1") == 1
|
|
agent._increment_retry("task-1")
|
|
assert agent.get_retry_count("task-1") == 2
|
|
|
|
def test_separate_task_counters(self):
|
|
agent, _, _ = _build_agent()
|
|
agent._increment_retry("task-1")
|
|
agent._increment_retry("task-1")
|
|
agent._increment_retry("task-2")
|
|
assert agent.get_retry_count("task-1") == 2
|
|
assert agent.get_retry_count("task-2") == 1
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_pipeline_failure_increments_counter(self):
|
|
agent, _, _ = _build_agent()
|
|
|
|
with patch.object(agent, "rebase_onto_main", new_callable=AsyncMock,
|
|
return_value={"success": False, "conflicts": ["a.py"]}):
|
|
await agent.review_and_merge("task-99", "/wt")
|
|
|
|
assert agent.get_retry_count("task-99") == 1
|
|
|
|
with patch.object(agent, "rebase_onto_main", new_callable=AsyncMock,
|
|
return_value={"success": False, "conflicts": ["a.py"]}):
|
|
await agent.review_and_merge("task-99", "/wt")
|
|
|
|
assert agent.get_retry_count("task-99") == 2
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Review response parsing
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestParseReviewResponse:
|
|
def test_approved_with_info(self):
|
|
agent, _, _ = _build_agent()
|
|
result = agent._parse_review_response(APPROVED_REVIEW)
|
|
assert result["approved"] is True
|
|
assert len(result["issues"]) == 1
|
|
assert result["issues"][0]["severity"] == "info"
|
|
assert "good" in result["summary"].lower()
|
|
|
|
def test_rejected_with_critical(self):
|
|
agent, _, _ = _build_agent()
|
|
result = agent._parse_review_response(REJECTED_REVIEW)
|
|
assert result["approved"] is False
|
|
assert len(result["issues"]) == 2
|
|
assert result["issues"][0]["severity"] == "critical"
|
|
assert result["issues"][1]["severity"] == "warning"
|
|
|
|
def test_empty_response(self):
|
|
agent, _, _ = _build_agent()
|
|
result = agent._parse_review_response("")
|
|
assert result["approved"] is False
|
|
assert result["issues"] == []
|
|
assert result["summary"] == ""
|