open-swe/tests/github/test_github_checks.py

from __future__ import annotations

from typing import Any
from unittest.mock import AsyncMock, patch

import httpx
import pytest

from agent.middleware.settle_review_check import settle_review_check_on_exit
from agent.review import publish as reviewer_publish
from agent.utils import github_checks


class _FakeResponse:
    def __init__(
        self,
        payload: dict[str, Any] | None = None,
        error: bool = False,
        status_code: int = 200,
    ) -> None:
        self._payload = payload or {}
        self._error = error
        self.status_code = status_code
        self.headers: dict[str, str] = {}

    def raise_for_status(self) -> None:
        if self._error:
            raise httpx.HTTPStatusError("forbidden", request=None, response=None)  # type: ignore[arg-type]

    def json(self) -> dict[str, Any]:
        return self._payload


class _FakeAsyncClient:
    last_post: dict[str, Any] | None = None
    last_patch: dict[str, Any] | None = None
    post_response: _FakeResponse = _FakeResponse({"id": 42})
    patch_response: _FakeResponse = _FakeResponse({})

    def __init__(self, **kwargs: Any) -> None:
        pass

    async def __aenter__(self) -> _FakeAsyncClient:
        return self

    async def __aexit__(self, exc_type: object, exc: object, tb: object) -> None:
        return None

    async def post(self, url: str, **kwargs: Any) -> _FakeResponse:
        type(self).last_post = {"url": url, **kwargs}
        return type(self).post_response

    async def patch(self, url: str, **kwargs: Any) -> _FakeResponse:
        type(self).last_patch = {"url": url, **kwargs}
        return type(self).patch_response


@pytest.fixture(autouse=True)
def _reset_fake_client() -> None:
    _FakeAsyncClient.last_post = None
    _FakeAsyncClient.last_patch = None
    _FakeAsyncClient.post_response = _FakeResponse({"id": 42})
    _FakeAsyncClient.patch_response = _FakeResponse({})


async def test_create_review_check_run_posts_in_progress(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    monkeypatch.setattr(github_checks.httpx, "AsyncClient", _FakeAsyncClient)

    check_run_id = await github_checks.create_review_check_run(
        owner="acme",
        repo="widgets",
        head_sha="abc123",
        token="tok",
        details_url="https://example.com/thread",
    )

    assert check_run_id == 42
    assert _FakeAsyncClient.last_post is not None
    assert _FakeAsyncClient.last_post["url"].endswith("/repos/acme/widgets/check-runs")
    body = _FakeAsyncClient.last_post["json"]
    assert body["name"] == github_checks.REVIEW_CHECK_RUN_NAME
    assert body["head_sha"] == "abc123"
    assert body["status"] == "in_progress"
    assert body["details_url"] == "https://example.com/thread"


async def test_create_review_check_run_returns_none_on_http_error(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    monkeypatch.setattr(github_checks.httpx, "AsyncClient", _FakeAsyncClient)
    _FakeAsyncClient.post_response = _FakeResponse(error=True)

    check_run_id = await github_checks.create_review_check_run(
        owner="acme", repo="widgets", head_sha="abc123", token="tok"
    )

    assert check_run_id is None


async def test_complete_review_check_run_patches_completed(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    monkeypatch.setattr(github_checks.httpx, "AsyncClient", _FakeAsyncClient)

    ok = await github_checks.complete_review_check_run(
        owner="acme",
        repo="widgets",
        check_run_id=42,
        token="tok",
        conclusion="neutral",
        title="Found 2 potential issues",
        summary="…",
    )

    assert ok is True
    assert _FakeAsyncClient.last_patch is not None
    assert _FakeAsyncClient.last_patch["url"].endswith("/repos/acme/widgets/check-runs/42")
    body = _FakeAsyncClient.last_patch["json"]
    assert body["status"] == "completed"
    assert body["conclusion"] == "neutral"
    assert body["output"]["title"] == "Found 2 potential issues"


async def test_post_autofix_status_check_completes_neutral(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    monkeypatch.setattr(github_checks.httpx, "AsyncClient", _FakeAsyncClient)

    ok = await github_checks.post_autofix_status_check(
        owner="acme",
        repo="widgets",
        head_sha="abc123",
        token="tok",
        title="Auto-fixing 1 failing check(s)",
        summary="working on it",
        details_url="https://example.com/thread",
    )

    assert ok is True
    assert _FakeAsyncClient.last_post is not None
    assert _FakeAsyncClient.last_post["url"].endswith("/repos/acme/widgets/check-runs")
    body = _FakeAsyncClient.last_post["json"]
    assert body["name"] == github_checks.AUTOFIX_CHECK_RUN_NAME
    assert body["status"] == "completed"
    assert body["conclusion"] == "neutral"
    assert body["details_url"] == "https://example.com/thread"


@pytest.mark.parametrize(
    ("outcome", "expected_conclusion", "expected_title"),
    [
        (
            {
                "verdict_submitted": True,
                "verdict_event": "APPROVE",
                "blocking_finding_count": 2,
                "verdict_authorization": "requested",
            },
            "success",
            "Review approved",
        ),
        (
            {
                "verdict_submitted": True,
                "verdict_event": "REQUEST_CHANGES",
                "blocking_finding_count": 0,
                "verdict_authorization": "requested",
            },
            "failure",
            "Changes requested",
        ),
        (
            {"blocking_finding_count": 0, "verdict_authorization": "none"},
            "success",
            "No issues found",
        ),
        (
            {"blocking_finding_count": 2, "verdict_authorization": "consistent"},
            "failure",
            "Found 2 blocking issues",
        ),
        (
            {
                "blocking_finding_count": 2,
                "verdict_authorization": "consistent",
                "verdict_ignored_reason": "approve_with_open_findings",
            },
            "failure",
            "Found 2 blocking issues",
        ),
        (
            {
                "blocking_finding_count": 1,
                "verdict_authorization": "requested",
                "verdict_ignored_reason": "self_review",
            },
            "failure",
            "Found 1 blocking issue",
        ),
        (
            {
                "blocking_finding_count": 0,
                "verdict_authorization": "consistent",
                "verdict_ignored_reason": "self_review",
            },
            "success",
            "No issues found",
        ),
        (
            {
                "blocking_finding_count": 1,
                "verdict_authorization": "consistent",
                "verdict_ignored_reason": "head_moved",
            },
            "neutral",
            "Verdict withheld",
        ),
        (
            {
                "blocking_finding_count": 1,
                "verdict_authorization": "consistent",
                "verdict_ignored_reason": "author_unknown",
            },
            "neutral",
            "Verdict withheld",
        ),
        (
            {
                "blocking_finding_count": 1,
                "verdict_authorization": "consistent",
                "verdict_ignored_reason": "head_check_failed",
            },
            "neutral",
            "Verdict withheld",
        ),
        (
            {
                "blocking_finding_count": 3,
                "surfaced_count": 3,
                "verdict_authorization": "none",
            },
            "neutral",
            "Found 3 potential issues",
        ),
        ({}, "neutral", "Review completed without verdict"),
    ],
)
def test_review_check_conclusion_mapping(
    outcome: dict[str, object],
    expected_conclusion: str,
    expected_title: str,
) -> None:
    conclusion, title, _ = github_checks.review_check_conclusion(outcome)
    assert conclusion == expected_conclusion
    assert title == expected_title


async def test_settle_review_check_run_noop_without_tracked_id(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    async def fake_get_thread_metadata(thread_id: str) -> dict[str, Any]:
        return {}

    completed: list[dict[str, Any]] = []

    async def fake_complete(**kwargs: Any) -> bool:
        completed.append(kwargs)
        return True

    monkeypatch.setattr(reviewer_publish, "get_thread_metadata", fake_get_thread_metadata)
    monkeypatch.setattr(reviewer_publish, "complete_review_check_run", fake_complete)

    await reviewer_publish.settle_review_check_run(
        thread_id="t1",
        owner="acme",
        repo="widgets",
        token="tok",
        conclusion="success",
        title="t",
        summary="s",
    )

    assert completed == []


async def test_settle_review_check_run_completes_and_clears(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    async def fake_get_thread_metadata(thread_id: str) -> dict[str, Any]:
        return {"review_check_run_id": 42}

    completed: list[dict[str, Any]] = []
    metadata_writes: list[dict[str, Any]] = []

    async def fake_complete(**kwargs: Any) -> bool:
        completed.append(kwargs)
        return True

    async def fake_set_metadata(thread_id: str, **kwargs: Any) -> None:
        metadata_writes.append({"thread_id": thread_id, **kwargs})

    monkeypatch.setattr(reviewer_publish, "get_thread_metadata", fake_get_thread_metadata)
    monkeypatch.setattr(reviewer_publish, "complete_review_check_run", fake_complete)
    monkeypatch.setattr(reviewer_publish, "set_reviewer_thread_metadata", fake_set_metadata)

    await reviewer_publish.settle_review_check_run(
        thread_id="t1",
        owner="acme",
        repo="widgets",
        token="tok",
        conclusion="neutral",
        title="t",
        summary="s",
    )

    assert len(completed) == 1
    assert completed[0]["check_run_id"] == 42
    assert completed[0]["conclusion"] == "neutral"
    assert metadata_writes == [
        {
            "thread_id": "t1",
            "extra": {"review_check_run_id": None, "review_check_pending_result": None},
        }
    ]


async def test_settle_review_check_run_keeps_id_on_patch_failure(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    async def fake_get_thread_metadata(thread_id: str) -> dict[str, Any]:
        return {"review_check_run_id": 42}

    metadata_writes: list[dict[str, Any]] = []

    async def fake_complete(**kwargs: Any) -> bool:
        return False

    async def fake_set_metadata(thread_id: str, **kwargs: Any) -> None:
        metadata_writes.append({"thread_id": thread_id, **kwargs})

    monkeypatch.setattr(reviewer_publish, "get_thread_metadata", fake_get_thread_metadata)
    monkeypatch.setattr(reviewer_publish, "complete_review_check_run", fake_complete)
    monkeypatch.setattr(reviewer_publish, "set_reviewer_thread_metadata", fake_set_metadata)

    await reviewer_publish.settle_review_check_run(
        thread_id="t1",
        owner="acme",
        repo="widgets",
        token="tok",
        conclusion="success",
        title="t",
        summary="s",
    )

    assert metadata_writes == [
        {
            "thread_id": "t1",
            "extra": {
                "review_check_pending_result": {
                    "conclusion": "success",
                    "title": "t",
                    "summary": "s",
                }
            },
        }
    ]


async def test_settle_review_check_on_exit_without_publish_is_neutral() -> None:
    settle = AsyncMock()
    with (
        patch(
            "agent.middleware.settle_review_check.get_config",
            return_value={
                "configurable": {
                    "thread_id": "t1",
                    "repo": {"owner": "acme", "name": "widgets"},
                }
            },
        ),
        patch(
            "agent.middleware.settle_review_check.get_thread_metadata",
            AsyncMock(return_value={"review_check_run_id": 42}),
        ),
        patch("agent.middleware.settle_review_check.get_github_token", return_value="tok"),
        patch("agent.middleware.settle_review_check.settle_review_check_run", settle),
    ):
        await settle_review_check_on_exit.aafter_agent({}, None)

    settle.assert_awaited_once()
    assert settle.await_args.kwargs["conclusion"] == "neutral"
    assert settle.await_args.kwargs["title"] == "Review did not complete"


@pytest.mark.parametrize("conclusion", ["success", "neutral", "failure"])
async def test_settle_review_check_on_exit_preserves_pending_conclusion(
    conclusion: str,
) -> None:
    settle = AsyncMock()
    with (
        patch(
            "agent.middleware.settle_review_check.get_config",
            return_value={
                "configurable": {
                    "thread_id": "t1",
                    "repo": {"owner": "acme", "name": "widgets"},
                }
            },
        ),
        patch(
            "agent.middleware.settle_review_check.get_thread_metadata",
            AsyncMock(
                return_value={
                    "review_check_run_id": 42,
                    "review_check_pending_result": {
                        "conclusion": conclusion,
                        "title": "Published result",
                        "summary": "Authoritative outcome",
                    },
                }
            ),
        ),
        patch("agent.middleware.settle_review_check.get_github_token", return_value="tok"),
        patch("agent.middleware.settle_review_check.settle_review_check_run", settle),
    ):
        await settle_review_check_on_exit.aafter_agent({}, None)

    assert settle.await_args.kwargs["conclusion"] == conclusion
    assert settle.await_args.kwargs["title"] == "Published result"