157 lines
5.2 KiB
Python
157 lines
5.2 KiB
Python
import os
|
|
import json
|
|
import logging
|
|
import pytest
|
|
from unittest.mock import MagicMock, AsyncMock, patch
|
|
from pathlib import Path
|
|
from main import JSONFormatter
|
|
from gitea.tools.coding_tools import CodingTools
|
|
from core.dispatcher import AgentDispatcher
|
|
from core.queue import WorkItem
|
|
from gitea.client import GiteaClient
|
|
from gitea.tools.gitea_tools import GiteaTools
|
|
from gitea.models import IssueModel, PullRequestModel
|
|
|
|
pytestmark = pytest.mark.anyio
|
|
|
|
|
|
def test_json_formatter() -> None:
|
|
formatter = JSONFormatter()
|
|
record = logging.LogRecord(
|
|
name="test-logger",
|
|
level=logging.INFO,
|
|
pathname="test.py",
|
|
lineno=10,
|
|
msg="Test log message",
|
|
args=(),
|
|
exc_info=None
|
|
)
|
|
formatted = formatter.format(record)
|
|
data = json.loads(formatted)
|
|
assert data["level"] == "INFO"
|
|
assert data["logger"] == "test-logger"
|
|
assert data["message"] == "Test log message"
|
|
assert "timestamp" in data
|
|
|
|
|
|
def test_read_file_limit(tmp_path: Path) -> None:
|
|
# Create a large file
|
|
large_file = tmp_path / "large_file.txt"
|
|
lines = [f"Line {i}\n" for i in range(1, 400)]
|
|
large_file.write_text("".join(lines), encoding="utf-8")
|
|
|
|
coding_tools = CodingTools(str(tmp_path))
|
|
# Test read with default limit (250)
|
|
result = coding_tools.read_file("large_file.txt", offset=1)
|
|
result_lines = [line.strip() for line in result.splitlines() if line.strip()]
|
|
assert len(result_lines) == 250
|
|
assert result_lines[0] == "1: Line 1"
|
|
assert result_lines[-1] == "250: Line 250"
|
|
|
|
# Test read with custom limit
|
|
result_custom = coding_tools.read_file("large_file.txt", offset=10, limit=10)
|
|
result_custom_lines = [line.strip() for line in result_custom.splitlines() if line.strip()]
|
|
assert len(result_custom_lines) == 10
|
|
assert result_custom_lines[0] == "10: Line 10"
|
|
assert result_custom_lines[-1] == "19: Line 19"
|
|
|
|
|
|
def test_parse_verification_commands(tmp_path: Path) -> None:
|
|
agents_md = tmp_path / "AGENTS.md"
|
|
agents_md.write_text(
|
|
"# Agent Instructions\n\n"
|
|
"## Verification\n"
|
|
"Please verify your changes with:\n"
|
|
"```bash\n"
|
|
"pytest -v\n"
|
|
"npm run lint\n"
|
|
"```\n\n"
|
|
"## Something Else\n"
|
|
"```bash\n"
|
|
"ignored command\n"
|
|
"```\n",
|
|
encoding="utf-8"
|
|
)
|
|
|
|
coding_tools = CodingTools(str(tmp_path))
|
|
commands = coding_tools._parse_verification_commands()
|
|
assert commands == ["pytest -v", "npm run lint"]
|
|
|
|
|
|
def test_run_verification_failure(tmp_path: Path) -> None:
|
|
agents_md = tmp_path / "AGENTS.md"
|
|
agents_md.write_text(
|
|
"## Verification\n"
|
|
"```bash\n"
|
|
"false\n"
|
|
"```\n",
|
|
encoding="utf-8"
|
|
)
|
|
coding_tools = CodingTools(str(tmp_path))
|
|
success, log_msg = coding_tools.run_verification()
|
|
assert not success
|
|
assert "false" in log_msg
|
|
|
|
|
|
@patch("core.dispatcher.CodingAgent")
|
|
@patch("core.dispatcher.PlanningAgent")
|
|
async def test_dispatch_planning_and_coding_phases(mock_planning_class: MagicMock, mock_coding_class: MagicMock) -> None:
|
|
mock_client: MagicMock = MagicMock(spec=GiteaClient)
|
|
mock_tools: MagicMock = MagicMock(spec=GiteaTools)
|
|
|
|
# Mock no existing PRs
|
|
mock_client.list_repo_pull_requests.return_value = []
|
|
mock_client.get_issue_comments.return_value = []
|
|
|
|
from gitea.models import UserModel
|
|
mock_client.get_authenticated_user.return_value = UserModel(login="unknown-ai")
|
|
|
|
mock_pr = PullRequestModel(
|
|
number=42,
|
|
title="fix bug",
|
|
body="bug details",
|
|
user=UserModel(login="unknown-ai")
|
|
)
|
|
mock_client.get_pull_request.return_value = mock_pr
|
|
mock_client.get_pull_request_diff.return_value = "diff"
|
|
mock_client.get_pull_request_comments.return_value = []
|
|
mock_client.get_pull_request_files.return_value = []
|
|
mock_client.get_pr_reviews.return_value = []
|
|
|
|
# Mock agent instances
|
|
mock_planning_agent = MagicMock()
|
|
mock_planning_agent.run_with_tools = AsyncMock(return_value="Plan: Modify file A")
|
|
mock_planning_class.return_value = mock_planning_agent
|
|
|
|
mock_coding_agent = MagicMock()
|
|
mock_coding_agent.run_with_tools = AsyncMock(return_value="PR #1 Created")
|
|
mock_coding_class.return_value = mock_coding_agent
|
|
|
|
dispatcher = AgentDispatcher(client=mock_client, tools=mock_tools)
|
|
|
|
work_item = WorkItem(
|
|
repo_full_name="meeks/repo1",
|
|
task_type="pr",
|
|
task_number=42,
|
|
task_info=mock_pr,
|
|
priority=0
|
|
)
|
|
|
|
results = await dispatcher.dispatch("meeks/repo1", [work_item])
|
|
|
|
assert len(results) == 1
|
|
assert results[0] == "PR #1 Created"
|
|
|
|
# Verify both agents ran
|
|
assert mock_planning_agent.run_with_tools.call_count == 1
|
|
assert mock_coding_agent.run_with_tools.call_count == 1
|
|
|
|
# Verify planning prompt was passed correct context
|
|
planning_call_args = mock_planning_agent.run_with_tools.call_args[0]
|
|
assert "PHASE 1: PLANNING PHASE" in planning_call_args[0]
|
|
|
|
# Verify coding prompt received the generated plan
|
|
coding_call_args = mock_coding_agent.run_with_tools.call_args[0]
|
|
assert "PHASE 2: EXECUTION/CODING PHASE" in coding_call_args[0]
|
|
assert "Plan: Modify file A" in coding_call_args[0]
|