open-swe/tests/github/test_pr_verdict_guard.py

from __future__ import annotations

import json
from typing import Any

from langchain_core.messages import ToolMessage

from agent.middleware.pr_verdict_guard import (
    PullRequestVerdictGuardMiddleware,
    is_pr_verdict_fallback_command,
)


class _Request:
    def __init__(self, command: str, *, tool: str = "execute") -> None:
        self.tool_call = {
            "name": tool,
            "args": {"command": command},
            "id": "call-1",
        }


async def _handler(_request: Any) -> ToolMessage:
    return ToolMessage(content="allowed", tool_call_id="call-1")


def test_detects_gh_pr_review_verdict_flags() -> None:
    assert is_pr_verdict_fallback_command("gh pr review 12 --approve")
    assert is_pr_verdict_fallback_command("gh pr review -a")
    assert is_pr_verdict_fallback_command("gh pr review 12 -r -b 'needs work'")
    assert is_pr_verdict_fallback_command("gh pr review 12 --request-changes")
    assert is_pr_verdict_fallback_command("gh pr review --approve=true")
    assert is_pr_verdict_fallback_command("gh pr review --request-changes='fix it'")
    assert is_pr_verdict_fallback_command("GH_TOKEN=dummy gh pr review 5 --approve")
    assert is_pr_verdict_fallback_command("git fetch && gh pr review 3 -a")


def test_detects_gh_api_review_verdicts() -> None:
    assert is_pr_verdict_fallback_command(
        "gh api repos/langchain-ai/open-swe/pulls/5/reviews -f event=APPROVE"
    )
    assert is_pr_verdict_fallback_command(
        "gh api repos/o/r/pulls/5/reviews -X POST -f event=REQUEST_CHANGES"
    )
    assert is_pr_verdict_fallback_command(
        "gh api repos/o/r/pulls/5/reviews/9/events -f event=approve"
    )
    assert is_pr_verdict_fallback_command(
        "gh api https://api.github.com/repos/o/r/pulls/5/reviews -f event=APPROVE"
    )


def test_detects_curl_review_verdicts() -> None:
    assert is_pr_verdict_fallback_command(
        'curl -X POST https://api.github.com/repos/o/r/pulls/5/reviews -d \'{"event": "APPROVE"}\''
    )
    assert is_pr_verdict_fallback_command(
        "curl https://api.github.com/repos/o/r/pulls/5/reviews/9/events "
        '--json \'{"event":"REQUEST_CHANGES"}\''
    )


def test_detects_nested_and_normalized_verdict_commands() -> None:
    assert is_pr_verdict_fallback_command("/usr/bin/gh pr review 12 --approve")
    assert is_pr_verdict_fallback_command(
        "/usr/bin/curl -X POST https://api.github.com/repos/o/r/pulls/5/reviews "
        '-d \'{"event": "APPROVE"}\''
    )
    assert is_pr_verdict_fallback_command("bash -c 'gh pr review 12 --approve'")
    assert is_pr_verdict_fallback_command("bash -c'gh pr review 12 --approve'")
    assert is_pr_verdict_fallback_command("zsh -lc'gh pr review 12 --approve'")
    assert is_pr_verdict_fallback_command("GH_TOKEN=dummy sh -c 'gh pr review -a'")
    assert is_pr_verdict_fallback_command(
        "zsh -lc 'gh api repos/o/r/pulls/5/reviews -X POST -f event=REQUEST_CHANGES'"
    )
    assert is_pr_verdict_fallback_command(
        'bash -c "curl -X POST https://api.github.com/repos/o/r/pulls/5/reviews '
        '-d \'{\\"event\\": \\"APPROVE\\"}\'"'
    )
    assert is_pr_verdict_fallback_command(
        'bash -c \'sh -c "dash -c \\"zsh -c \\\\\\"gh pr review 12 --approve\\\\\\"\\""\''
    )


def test_allows_safe_review_commands() -> None:
    assert not is_pr_verdict_fallback_command("gh pr review 12 --comment -b 'looks good'")
    assert not is_pr_verdict_fallback_command("gh pr review -c")
    assert not is_pr_verdict_fallback_command("gh pr review")
    assert not is_pr_verdict_fallback_command("gh pr diff 12")
    assert not is_pr_verdict_fallback_command("gh pr view 12 --json reviews")
    assert not is_pr_verdict_fallback_command("gh api repos/o/r/pulls/5/reviews")
    assert not is_pr_verdict_fallback_command("gh pr create --draft")
    assert not is_pr_verdict_fallback_command("curl https://api.github.com/repos/o/r/pulls/5")
    assert not is_pr_verdict_fallback_command("bash -c 'gh pr review 12 --comment -b ok'")
    assert not is_pr_verdict_fallback_command("/usr/bin/gh pr view 12 --json reviews")


async def test_middleware_blocks_execute_verdict_fallbacks() -> None:
    for command in (
        "gh pr review 12 --approve",
        "gh api repos/o/r/pulls/5/reviews -f event=REQUEST_CHANGES",
        'curl -X POST https://api.github.com/repos/o/r/pulls/5/reviews -d \'{"event": "APPROVE"}\'',
    ):
        result = await PullRequestVerdictGuardMiddleware().awrap_tool_call(
            _Request(command), _handler
        )

        assert isinstance(result, ToolMessage)
        assert result.status == "error"
        payload = json.loads(str(result.content))
        assert payload["code"] == "pr_verdict_fallback_blocked"
        assert payload["error_type"] == "PullRequestVerdictFallbackBlocked"
        assert payload["recoverable_by_agent"] is False
        assert "publish_review" in payload["error"]
        assert payload["blocked_command"] == command


async def test_middleware_allows_comment_review() -> None:
    result = await PullRequestVerdictGuardMiddleware().awrap_tool_call(
        _Request("gh pr review 12 --comment -b 'nit: rename'"), _handler
    )

    assert isinstance(result, ToolMessage)
    assert result.content == "allowed"


async def test_middleware_ignores_other_tools() -> None:
    result = await PullRequestVerdictGuardMiddleware().awrap_tool_call(
        _Request("gh pr review 12 --approve", tool="read_file"), _handler
    )

    assert isinstance(result, ToolMessage)
    assert result.content == "allowed"