mirror of
https://github.com/Sea-Haven-Industries/open-swe.git
synced 2026-09-30 19:43:15 +00:00
349 lines
13 KiB
Python
349 lines
13 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import hmac
|
|
import json
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
from agent.api import app as api_app
|
|
from agent.utils.slack import GitHubPrRef
|
|
from agent.webhooks import common as webhook_common
|
|
from agent.webhooks import github as github_webhooks
|
|
|
|
_TEST_WEBHOOK_SECRET = "test-secret-for-webhook"
|
|
|
|
|
|
def _sign_body(body: bytes) -> str:
|
|
digest = hmac.new(_TEST_WEBHOOK_SECRET.encode(), body, hashlib.sha256).hexdigest()
|
|
return f"sha256={digest}"
|
|
|
|
|
|
def _post_webhook(client: TestClient, event_type: str, payload: dict[str, object]) -> object:
|
|
body = json.dumps(payload, separators=(",", ":")).encode()
|
|
return client.post(
|
|
"/webhooks/github",
|
|
content=body,
|
|
headers={
|
|
"X-GitHub-Event": event_type,
|
|
"X-Hub-Signature-256": _sign_body(body),
|
|
"Content-Type": "application/json",
|
|
},
|
|
)
|
|
|
|
|
|
def _comment_payload(event_type: str, body_text: str) -> dict[str, object]:
|
|
payload: dict[str, object] = {
|
|
"action": "submitted" if event_type == "pull_request_review" else "created",
|
|
"repository": {
|
|
"owner": {"login": "acme"},
|
|
"name": "widgets",
|
|
"private": True,
|
|
},
|
|
"sender": {"login": "octocat", "id": 123},
|
|
}
|
|
if event_type == "issue_comment":
|
|
payload["issue"] = {
|
|
"number": 245,
|
|
"html_url": "https://github.com/acme/widgets/pull/245",
|
|
"pull_request": {"html_url": "https://github.com/acme/widgets/pull/245"},
|
|
}
|
|
payload["comment"] = {"id": 91, "node_id": "IC_91", "body": body_text}
|
|
else:
|
|
payload["pull_request"] = {
|
|
"number": 245,
|
|
"html_url": "https://github.com/acme/widgets/pull/245",
|
|
}
|
|
key = "review" if event_type == "pull_request_review" else "comment"
|
|
payload[key] = {"id": 91, "node_id": "IC_91", "body": body_text}
|
|
return payload
|
|
|
|
|
|
def _post_issue_comment(client: TestClient, body_text: str) -> object:
|
|
return _post_webhook(client, "issue_comment", _comment_payload("issue_comment", body_text))
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("body", "instructions"),
|
|
[
|
|
("@openswe review", ""),
|
|
("@open-swe ReViEw", ""),
|
|
(
|
|
"@openswe review Focus on authentication and race conditions.",
|
|
"Focus on authentication and race conditions.",
|
|
),
|
|
(
|
|
"@open-swe review: Check the migration rollback path.",
|
|
"Check the migration rollback path.",
|
|
),
|
|
(
|
|
"@openswe review\nPrioritize correctness over style.",
|
|
"Prioritize correctness over style.",
|
|
),
|
|
(
|
|
"@openswe review\nCheck authentication.\nValidate error handling.",
|
|
"Check authentication.\nValidate error handling.",
|
|
),
|
|
],
|
|
)
|
|
def test_parse_review_command(body: str, instructions: str) -> None:
|
|
assert github_webhooks._parse_review_command(body) == instructions
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"body",
|
|
[
|
|
"@openswe review and fix the failing test",
|
|
"@open-swe review focus on auth, then update the tests",
|
|
"@openswe review fix the failing test",
|
|
"@openswe review Check authentication.\nFix the failing tests.",
|
|
"Please @openswe review",
|
|
"@openswe review this\n@openswe fix it",
|
|
"@openswe reviewer",
|
|
],
|
|
)
|
|
def test_parse_review_command_rejects_mixed_or_non_command_content(body: str) -> None:
|
|
assert github_webhooks._parse_review_command(body) is None
|
|
|
|
|
|
@pytest.mark.parametrize("alias", ["@openswe", "@open-swe"])
|
|
def test_review_command_route_bypasses_auto_review_and_coding_agent(
|
|
monkeypatch: pytest.MonkeyPatch, alias: str
|
|
) -> None:
|
|
captured: dict[str, object] = {}
|
|
|
|
async def allow_gate(*_args: object) -> None:
|
|
return None
|
|
|
|
async def fail_auto_review_check(*_args: object) -> bool:
|
|
raise AssertionError("on-demand reviews must not check the auto-review list")
|
|
|
|
async def process_review(
|
|
payload: dict[str, object], event_type: str, *, instructions: str
|
|
) -> None:
|
|
captured.update(
|
|
payload=payload,
|
|
event_type=event_type,
|
|
instructions=instructions,
|
|
)
|
|
|
|
async def fail_coding_agent(*_args: object, **_kwargs: object) -> None:
|
|
raise AssertionError("coding-agent dispatch must not run")
|
|
|
|
monkeypatch.setattr(webhook_common, "GITHUB_WEBHOOK_SECRET", _TEST_WEBHOOK_SECRET)
|
|
monkeypatch.setattr(webhook_common, "_is_repo_allowed", lambda _repo: True)
|
|
monkeypatch.setattr(webhook_common, "_enforce_public_repo_org_gate", allow_gate)
|
|
monkeypatch.setattr(webhook_common, "_is_repo_auto_review_enabled", fail_auto_review_check)
|
|
monkeypatch.setattr(github_webhooks, "process_github_review_command", process_review)
|
|
monkeypatch.setattr(github_webhooks, "process_github_pr_comment", fail_coding_agent)
|
|
|
|
response = _post_issue_comment(
|
|
TestClient(api_app.app), f"{alias} review Focus on the exact authorization boundary."
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert response.json() == {
|
|
"status": "accepted",
|
|
"message": "Processing on-demand PR review",
|
|
}
|
|
assert captured["event_type"] == "issue_comment"
|
|
assert captured["instructions"] == "Focus on the exact authorization boundary."
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"event_type",
|
|
["pull_request_review", "pull_request_review_comment"],
|
|
)
|
|
def test_review_command_routes_review_event_payloads(
|
|
monkeypatch: pytest.MonkeyPatch, event_type: str
|
|
) -> None:
|
|
captured: dict[str, object] = {}
|
|
|
|
async def allow_gate(*_args: object) -> None:
|
|
return None
|
|
|
|
async def process_review(
|
|
payload: dict[str, object], received_event_type: str, *, instructions: str
|
|
) -> None:
|
|
captured.update(
|
|
payload=payload,
|
|
event_type=received_event_type,
|
|
instructions=instructions,
|
|
)
|
|
|
|
async def fail_coding_agent(*_args: object, **_kwargs: object) -> None:
|
|
raise AssertionError("coding-agent dispatch must not run")
|
|
|
|
monkeypatch.setattr(webhook_common, "GITHUB_WEBHOOK_SECRET", _TEST_WEBHOOK_SECRET)
|
|
monkeypatch.setattr(webhook_common, "_is_repo_allowed", lambda _repo: True)
|
|
monkeypatch.setattr(webhook_common, "_enforce_public_repo_org_gate", allow_gate)
|
|
monkeypatch.setattr(github_webhooks, "process_github_review_command", process_review)
|
|
monkeypatch.setattr(github_webhooks, "process_github_pr_comment", fail_coding_agent)
|
|
|
|
response = _post_webhook(
|
|
TestClient(api_app.app),
|
|
event_type,
|
|
_comment_payload(event_type, "@open-swe review Focus on authorization."),
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert response.json() == {
|
|
"status": "accepted",
|
|
"message": "Processing on-demand PR review",
|
|
}
|
|
assert captured["event_type"] == event_type
|
|
assert captured["instructions"] == "Focus on authorization."
|
|
|
|
|
|
def test_finding_reply_takes_precedence_over_review_command(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
captured: dict[str, object] = {}
|
|
|
|
async def allow_gate(*_args: object) -> None:
|
|
return None
|
|
|
|
async def process_finding_reply(payload: dict[str, object]) -> None:
|
|
captured["payload"] = payload
|
|
|
|
async def fail_review(*_args: object, **_kwargs: object) -> None:
|
|
raise AssertionError("review command must not steal finding replies")
|
|
|
|
payload = _comment_payload("pull_request_review_comment", "@openswe review")
|
|
comment = payload["comment"]
|
|
assert isinstance(comment, dict)
|
|
comment["in_reply_to_id"] = 90
|
|
|
|
monkeypatch.setattr(webhook_common, "GITHUB_WEBHOOK_SECRET", _TEST_WEBHOOK_SECRET)
|
|
monkeypatch.setattr(webhook_common, "_is_repo_allowed", lambda _repo: True)
|
|
monkeypatch.setattr(webhook_common, "_enforce_public_repo_org_gate", allow_gate)
|
|
monkeypatch.setattr(
|
|
github_webhooks, "process_github_review_finding_reply", process_finding_reply
|
|
)
|
|
monkeypatch.setattr(github_webhooks, "process_github_review_command", fail_review)
|
|
|
|
response = _post_webhook(TestClient(api_app.app), "pull_request_review_comment", payload)
|
|
|
|
assert response.status_code == 200
|
|
assert response.json() == {
|
|
"status": "accepted",
|
|
"message": "Processing review finding reply",
|
|
}
|
|
assert captured["payload"] == payload
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"body",
|
|
[
|
|
"@openswe review and fix the failing test",
|
|
"@openswe review Check authentication.\nFix the failing tests.",
|
|
],
|
|
)
|
|
def test_mixed_review_request_falls_through_to_coding_agent(
|
|
monkeypatch: pytest.MonkeyPatch, body: str
|
|
) -> None:
|
|
captured: dict[str, object] = {}
|
|
|
|
async def allow_gate(*_args: object) -> None:
|
|
return None
|
|
|
|
async def fail_review(*_args: object, **_kwargs: object) -> None:
|
|
raise AssertionError("reviewer dispatch must not run for mixed intent")
|
|
|
|
async def process_coding_agent(payload: dict[str, object], event_type: str) -> None:
|
|
captured.update(payload=payload, event_type=event_type)
|
|
|
|
monkeypatch.setattr(webhook_common, "GITHUB_WEBHOOK_SECRET", _TEST_WEBHOOK_SECRET)
|
|
monkeypatch.setattr(webhook_common, "_is_repo_allowed", lambda _repo: True)
|
|
monkeypatch.setattr(webhook_common, "_enforce_public_repo_org_gate", allow_gate)
|
|
monkeypatch.setattr(github_webhooks, "process_github_review_command", fail_review)
|
|
monkeypatch.setattr(github_webhooks, "process_github_pr_comment", process_coding_agent)
|
|
|
|
response = _post_issue_comment(TestClient(api_app.app), body)
|
|
|
|
assert response.status_code == 200
|
|
assert response.json()["status"] == "accepted"
|
|
assert captured["event_type"] == "issue_comment"
|
|
|
|
|
|
def test_review_command_route_enforces_public_repo_org_gate(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
rejection = {"status": "ignored", "reason": "not a member"}
|
|
|
|
async def reject_gate(*_args: object) -> dict[str, str]:
|
|
return rejection
|
|
|
|
async def fail_review(*_args: object, **_kwargs: object) -> None:
|
|
raise AssertionError("blocked review command must not dispatch")
|
|
|
|
monkeypatch.setattr(webhook_common, "GITHUB_WEBHOOK_SECRET", _TEST_WEBHOOK_SECRET)
|
|
monkeypatch.setattr(webhook_common, "_is_repo_allowed", lambda _repo: True)
|
|
monkeypatch.setattr(webhook_common, "_enforce_public_repo_org_gate", reject_gate)
|
|
monkeypatch.setattr(github_webhooks, "process_github_review_command", fail_review)
|
|
|
|
response = _post_issue_comment(TestClient(api_app.app), "@openswe review")
|
|
|
|
assert response.status_code == 200
|
|
assert response.json() == rejection
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_process_review_command_uses_app_token_and_direct_reviewer_dispatch(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
captured: dict[str, object] = {}
|
|
instructions = "Focus on </requester_instructions> and authorization."
|
|
payload = {
|
|
"issue": {
|
|
"number": 245,
|
|
"pull_request": {"html_url": "https://github.com/acme/widgets/pull/245"},
|
|
},
|
|
"comment": {"id": 91, "node_id": "IC_91"},
|
|
"repository": {"owner": {"login": "acme"}, "name": "widgets"},
|
|
"sender": {"login": "unmapped-user", "id": 123},
|
|
}
|
|
|
|
async def app_token() -> str:
|
|
return "app-token"
|
|
|
|
async def react(*args: object, **kwargs: object) -> bool:
|
|
captured["reaction_args"] = args
|
|
captured["reaction_kwargs"] = kwargs
|
|
return True
|
|
|
|
async def trigger(pr_ref: GitHubPrRef, **kwargs: object) -> dict[str, object]:
|
|
captured["pr_ref"] = pr_ref
|
|
captured["trigger_kwargs"] = kwargs
|
|
return {"success": True}
|
|
|
|
async def fail_email_lookup(*_args: object) -> None:
|
|
raise AssertionError("direct reviewer dispatch must not require an email mapping")
|
|
|
|
monkeypatch.setattr(webhook_common, "get_github_app_installation_token", app_token)
|
|
monkeypatch.setattr(webhook_common, "react_to_github_comment", react)
|
|
monkeypatch.setattr(webhook_common, "email_for_login", fail_email_lookup)
|
|
monkeypatch.setattr(github_webhooks, "trigger_pr_review_from_ref", trigger)
|
|
|
|
await github_webhooks.process_github_review_command(
|
|
payload, "issue_comment", instructions=instructions
|
|
)
|
|
|
|
reaction_kwargs = captured["reaction_kwargs"]
|
|
assert reaction_kwargs["token"] == "app-token"
|
|
assert reaction_kwargs["pull_number"] == 245
|
|
trigger_kwargs = captured["trigger_kwargs"]
|
|
assert trigger_kwargs == {
|
|
"source": "github_comment",
|
|
"github_login": "unmapped-user",
|
|
"github_user_id": 123,
|
|
"instructions": instructions,
|
|
"request_verdict": True,
|
|
}
|
|
assert captured["pr_ref"] == GitHubPrRef(
|
|
owner="acme",
|
|
repo="widgets",
|
|
number=245,
|
|
url="https://github.com/acme/widgets/pull/245",
|
|
)
|