coding-agent-gitea/tests/test_best_practices.py

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="meeks-ai")
    
    mock_pr = PullRequestModel(
        number=42,
        title="fix bug",
        body="bug details",
        user=UserModel(login="meeks-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]