open-swe/tests/test_autofix_state.py
Johannes du Plessis 7397ff93ba
feat: CI auto-fix and PR babysitting for agent PRs (#1530)
* feat: CI auto-fix and PR babysitting for agent PRs

Watch CI failures and review feedback on PRs Open SWE opened, then dispatch
confidence-gated fix runs on the originating agent thread. Adds CI webhook
ingestion (check_run/check_suite/workflow_run/status), a per-PR @open-swe
autofix on|off toggle, auto-response to review comments, and a polling
ci_monitor graph that also flags merge conflicts. Gated by the existing
autofix_mode/trigger_mode settings, the enabled-repos opt-in, base-branch and
human-commit skip rules, dedupe, and a per-PR attempt cap.

Co-authored-by: open-swe[bot] <open-swe@users.noreply.github.com>

* fix: address review feedback on CI auto-fix

- Security: gate the no-mention review-feedback path on author trust —
  require a trusted author_association (OWNER/MEMBER/COLLABORATOR) plus a
  GitHub write/maintain/admin permission check before dispatching a
  write-capable agent run, preventing privilege escalation from
  read/triage/outside reviewers.
- Auth: reuse the originating PR thread's source + login/email when
  dispatching fix runs so the GitHub-token resolver authenticates them in
  non-bot-token deployments (bespoke github_ci source failed to resolve).
- Docs: document the Commit statuses: Read-only permission required for the
  Status webhook event.

Co-authored-by: open-swe[bot] <open-swe@users.noreply.github.com>

---------

Co-authored-by: open-swe[bot] <open-swe@users.noreply.github.com>
2026-06-15 13:53:50 -07:00

70 lines
2.3 KiB
Python

"""Unit tests for per-PR auto-fix opt-out state and team settings accessor."""
from __future__ import annotations
from typing import Any
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from agent.dashboard import autofix_state, team_settings
@pytest.mark.asyncio
async def test_set_and_check_pr_disabled() -> None:
store: dict[tuple[Any, ...], Any] = {}
client = MagicMock()
async def put_item(ns: list[str], key: str, value: dict[str, Any]) -> None:
store[(tuple(ns), key)] = value
async def get_item(ns: list[str], key: str) -> dict[str, Any] | None:
value = store.get((tuple(ns), key))
return {"value": value} if value is not None else None
client.store.put_item = AsyncMock(side_effect=put_item)
client.store.get_item = AsyncMock(side_effect=get_item)
with patch.object(autofix_state, "get_client", return_value=client):
assert await autofix_state.is_pr_autofix_disabled("O", "R", 5) is False
await autofix_state.set_pr_autofix_disabled("O", "R", 5, True)
assert await autofix_state.is_pr_autofix_disabled("o", "r", 5) is True
await autofix_state.set_pr_autofix_disabled("o", "r", 5, False)
assert await autofix_state.is_pr_autofix_disabled("o", "r", 5) is False
@pytest.mark.asyncio
async def test_get_autofix_settings_normalizes() -> None:
with patch.object(
team_settings,
"get_team_settings",
AsyncMock(
return_value={
"autofix_mode": "bogus",
"autofix_severity_threshold": "high",
"trigger_mode": "weird",
}
),
):
settings = await team_settings.get_autofix_settings()
assert settings == {
"autofix_mode": "off",
"autofix_severity_threshold": "high",
"trigger_mode": "every_push",
}
@pytest.mark.asyncio
async def test_is_autofix_enabled() -> None:
with patch.object(
team_settings,
"get_team_settings",
AsyncMock(return_value={"autofix_mode": "high"}),
):
assert await team_settings.is_autofix_enabled() is True
with patch.object(
team_settings,
"get_team_settings",
AsyncMock(return_value={"autofix_mode": "off"}),
):
assert await team_settings.is_autofix_enabled() is False