mirror of
https://github.com/Sea-Haven-Industries/open-swe.git
synced 2026-09-30 08:03:15 +00:00
417 lines
13 KiB
Python
417 lines
13 KiB
Python
"""Tests for the public-repo org-membership gate on GitHub webhooks."""
|
|
|
|
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.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, secret: str = _TEST_WEBHOOK_SECRET) -> str:
|
|
sig = hmac.new(secret.encode(), body, hashlib.sha256).hexdigest()
|
|
return f"sha256={sig}"
|
|
|
|
|
|
def _post_github_webhook(client: TestClient, event_type: str, payload: dict) -> 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 _install_membership_stub(monkeypatch, members: set[str]) -> dict[str, list[str]]:
|
|
"""Stub the org-membership lookup to return True only for ``members``."""
|
|
seen: dict[str, list[str]] = {"calls": []}
|
|
|
|
async def fake_is_user_active_org_member(username: str, org: str) -> bool:
|
|
seen["calls"].append(username)
|
|
return username in members
|
|
|
|
monkeypatch.setattr(webhook_common, "is_user_active_org_member", fake_is_user_active_org_member)
|
|
return seen
|
|
|
|
|
|
def _common_setup(monkeypatch, *, gate: str = "langchain-ai") -> None:
|
|
monkeypatch.setattr(webhook_common, "GITHUB_WEBHOOK_SECRET", _TEST_WEBHOOK_SECRET)
|
|
monkeypatch.setattr(webhook_common, "PUBLIC_REPO_ORG_GATE", gate)
|
|
monkeypatch.setattr(webhook_common, "ALLOWED_GITHUB_ORGS", frozenset())
|
|
|
|
|
|
def test_gate_blocks_non_member_on_public_pr_comment(monkeypatch) -> None:
|
|
_common_setup(monkeypatch)
|
|
seen = _install_membership_stub(monkeypatch, members={"insider"})
|
|
|
|
async def fake_process_github_review_command(*_args, **_kwargs) -> None:
|
|
raise AssertionError("should not be called")
|
|
|
|
monkeypatch.setattr(
|
|
github_webhooks,
|
|
"process_github_review_command",
|
|
fake_process_github_review_command,
|
|
)
|
|
|
|
client = TestClient(api_app.app)
|
|
response = _post_github_webhook(
|
|
client,
|
|
"issue_comment",
|
|
{
|
|
"action": "created",
|
|
"issue": {
|
|
"id": 1,
|
|
"number": 7,
|
|
"title": "PR title",
|
|
"pull_request": {
|
|
"url": "https://api.github.com/repos/langchain-ai/open-swe/pulls/7"
|
|
},
|
|
},
|
|
"comment": {"body": "@open-swe review"},
|
|
"repository": {
|
|
"owner": {"login": "langchain-ai"},
|
|
"name": "open-swe",
|
|
"private": False,
|
|
},
|
|
"sender": {"login": "stranger", "type": "User"},
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
body = response.json()
|
|
assert body["status"] == "ignored"
|
|
assert "not a member" in body["reason"]
|
|
assert seen["calls"] == ["stranger"]
|
|
|
|
|
|
def test_gate_allows_org_member_on_public_pr_comment(monkeypatch) -> None:
|
|
_common_setup(monkeypatch)
|
|
_install_membership_stub(monkeypatch, members={"insider"})
|
|
|
|
called: dict[str, object] = {}
|
|
|
|
async def fake_process_github_review_command(payload, event_type, *, instructions: str) -> None:
|
|
called["event"] = event_type
|
|
called["instructions"] = instructions
|
|
|
|
monkeypatch.setattr(
|
|
github_webhooks,
|
|
"process_github_review_command",
|
|
fake_process_github_review_command,
|
|
)
|
|
|
|
client = TestClient(api_app.app)
|
|
response = _post_github_webhook(
|
|
client,
|
|
"issue_comment",
|
|
{
|
|
"action": "created",
|
|
"issue": {
|
|
"id": 1,
|
|
"number": 7,
|
|
"title": "PR title",
|
|
"pull_request": {
|
|
"url": "https://api.github.com/repos/langchain-ai/open-swe/pulls/7"
|
|
},
|
|
},
|
|
"comment": {"body": "@open-swe review"},
|
|
"repository": {
|
|
"owner": {"login": "langchain-ai"},
|
|
"name": "open-swe",
|
|
"private": False,
|
|
},
|
|
"sender": {"login": "insider", "type": "User"},
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert response.json()["status"] == "accepted"
|
|
assert called["event"] == "issue_comment"
|
|
assert called["instructions"] == ""
|
|
|
|
|
|
def test_gate_skipped_on_private_repo(monkeypatch) -> None:
|
|
_common_setup(monkeypatch)
|
|
seen = _install_membership_stub(monkeypatch, members=set())
|
|
|
|
called: dict[str, object] = {}
|
|
|
|
async def fake_process_github_pr_comment(payload, event_type) -> None:
|
|
called["event"] = event_type
|
|
|
|
monkeypatch.setattr(
|
|
github_webhooks, "process_github_pr_comment", fake_process_github_pr_comment
|
|
)
|
|
|
|
client = TestClient(api_app.app)
|
|
response = _post_github_webhook(
|
|
client,
|
|
"issue_comment",
|
|
{
|
|
"action": "created",
|
|
"issue": {
|
|
"id": 1,
|
|
"number": 7,
|
|
"title": "PR title",
|
|
"pull_request": {
|
|
"url": "https://api.github.com/repos/langchain-ai/open-swe/pulls/7"
|
|
},
|
|
},
|
|
"comment": {"body": "@open-swe please look at this"},
|
|
"repository": {
|
|
"owner": {"login": "langchain-ai"},
|
|
"name": "open-swe",
|
|
"private": True,
|
|
},
|
|
"sender": {"login": "stranger", "type": "User"},
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert response.json()["status"] == "accepted"
|
|
assert called["event"] == "issue_comment"
|
|
assert seen["calls"] == []
|
|
|
|
|
|
def test_gate_disabled_when_env_unset(monkeypatch) -> None:
|
|
_common_setup(monkeypatch, gate="")
|
|
seen = _install_membership_stub(monkeypatch, members=set())
|
|
|
|
called: dict[str, object] = {}
|
|
|
|
async def fake_process_github_pr_comment(payload, event_type) -> None:
|
|
called["event"] = event_type
|
|
|
|
monkeypatch.setattr(
|
|
github_webhooks, "process_github_pr_comment", fake_process_github_pr_comment
|
|
)
|
|
|
|
client = TestClient(api_app.app)
|
|
response = _post_github_webhook(
|
|
client,
|
|
"issue_comment",
|
|
{
|
|
"action": "created",
|
|
"issue": {
|
|
"id": 1,
|
|
"number": 7,
|
|
"title": "PR title",
|
|
"pull_request": {
|
|
"url": "https://api.github.com/repos/langchain-ai/open-swe/pulls/7"
|
|
},
|
|
},
|
|
"comment": {"body": "@open-swe please look at this"},
|
|
"repository": {
|
|
"owner": {"login": "langchain-ai"},
|
|
"name": "open-swe",
|
|
"private": False,
|
|
},
|
|
"sender": {"login": "stranger", "type": "User"},
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert response.json()["status"] == "accepted"
|
|
assert called["event"] == "issue_comment"
|
|
assert seen["calls"] == []
|
|
|
|
|
|
def test_gate_blocks_non_member_on_public_issue(monkeypatch) -> None:
|
|
_common_setup(monkeypatch)
|
|
_install_membership_stub(monkeypatch, members={"insider"})
|
|
|
|
async def fake_process_github_issue(*_args, **_kwargs) -> None:
|
|
raise AssertionError("should not be called")
|
|
|
|
monkeypatch.setattr(github_webhooks, "process_github_issue", fake_process_github_issue)
|
|
|
|
client = TestClient(api_app.app)
|
|
response = _post_github_webhook(
|
|
client,
|
|
"issues",
|
|
{
|
|
"action": "opened",
|
|
"issue": {
|
|
"id": 1,
|
|
"number": 7,
|
|
"title": "@openswe please help",
|
|
"body": "x",
|
|
},
|
|
"repository": {
|
|
"owner": {"login": "langchain-ai"},
|
|
"name": "open-swe",
|
|
"private": False,
|
|
},
|
|
"sender": {"login": "stranger", "type": "User"},
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
body = response.json()
|
|
assert body["status"] == "ignored"
|
|
assert "not a member" in body["reason"]
|
|
|
|
|
|
def test_review_requested_is_unsupported_before_public_repo_gate(monkeypatch) -> None:
|
|
_common_setup(monkeypatch)
|
|
seen = _install_membership_stub(monkeypatch, members={"insider"})
|
|
|
|
client = TestClient(api_app.app)
|
|
response = _post_github_webhook(
|
|
client,
|
|
"pull_request",
|
|
{
|
|
"action": "review_requested",
|
|
"requested_reviewer": {"login": "open-swe[bot]"},
|
|
"pull_request": {
|
|
"number": 1244,
|
|
"html_url": "https://github.com/langchain-ai/open-swe/pull/1244",
|
|
"base": {"sha": "base-sha"},
|
|
"head": {"sha": "head-sha", "ref": "feature-branch"},
|
|
},
|
|
"repository": {
|
|
"owner": {"login": "langchain-ai"},
|
|
"name": "open-swe",
|
|
"private": False,
|
|
},
|
|
"sender": {"login": "stranger", "type": "User"},
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert response.json() == {
|
|
"status": "ignored",
|
|
"reason": "Unsupported GitHub pull_request action: review_requested",
|
|
}
|
|
assert seen["calls"] == []
|
|
|
|
|
|
def test_gate_allows_internal_bot_sender(monkeypatch) -> None:
|
|
_common_setup(monkeypatch)
|
|
seen = _install_membership_stub(monkeypatch, members=set())
|
|
|
|
called: dict[str, object] = {}
|
|
|
|
async def fake_process_github_pr_comment(payload, event_type) -> None:
|
|
called["event"] = event_type
|
|
|
|
monkeypatch.setattr(
|
|
github_webhooks, "process_github_pr_comment", fake_process_github_pr_comment
|
|
)
|
|
|
|
client = TestClient(api_app.app)
|
|
response = _post_github_webhook(
|
|
client,
|
|
"issue_comment",
|
|
{
|
|
"action": "created",
|
|
"issue": {
|
|
"id": 1,
|
|
"number": 7,
|
|
"title": "PR title",
|
|
"pull_request": {
|
|
"url": "https://api.github.com/repos/langchain-ai/open-swe/pulls/7"
|
|
},
|
|
},
|
|
"comment": {"body": "@open-swe please look at this"},
|
|
"repository": {
|
|
"owner": {"login": "langchain-ai"},
|
|
"name": "open-swe",
|
|
"private": False,
|
|
},
|
|
"sender": {"login": "open-swe[bot]", "type": "Bot"},
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert response.json()["status"] == "accepted"
|
|
assert called["event"] == "issue_comment"
|
|
assert seen["calls"] == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_is_user_active_org_member_returns_false_when_no_token(monkeypatch) -> None:
|
|
from agent.utils import github_org_membership
|
|
|
|
async def fake_token() -> None:
|
|
return None
|
|
|
|
monkeypatch.setattr(github_org_membership, "get_github_app_installation_token", fake_token)
|
|
|
|
assert await github_org_membership.is_user_active_org_member("alice", "langchain-ai") is False
|
|
|
|
|
|
class _FakeAsyncClient:
|
|
def __init__(self, response) -> None:
|
|
self._response = response
|
|
|
|
async def __aenter__(self) -> _FakeAsyncClient:
|
|
return self
|
|
|
|
async def __aexit__(self, *_args) -> None:
|
|
return None
|
|
|
|
async def get(self, *_args, **_kwargs):
|
|
return self._response
|
|
|
|
|
|
class _FakeResponse:
|
|
def __init__(self, status_code: int, payload: dict | None = None) -> None:
|
|
self.status_code = status_code
|
|
self._payload = payload or {}
|
|
|
|
def json(self) -> dict:
|
|
return self._payload
|
|
|
|
|
|
def _patch_membership_http(monkeypatch, response: _FakeResponse) -> None:
|
|
from agent.utils import github_org_membership
|
|
|
|
async def fake_token() -> str:
|
|
return "x"
|
|
|
|
monkeypatch.setattr(github_org_membership, "get_github_app_installation_token", fake_token)
|
|
|
|
def factory(*_args, **_kwargs) -> _FakeAsyncClient:
|
|
return _FakeAsyncClient(response)
|
|
|
|
monkeypatch.setattr(github_org_membership.httpx, "AsyncClient", factory)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_is_user_active_org_member_handles_404(monkeypatch) -> None:
|
|
from agent.utils import github_org_membership
|
|
|
|
_patch_membership_http(monkeypatch, _FakeResponse(404))
|
|
|
|
assert await github_org_membership.is_user_active_org_member("alice", "langchain-ai") is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_is_user_active_org_member_active(monkeypatch) -> None:
|
|
from agent.utils import github_org_membership
|
|
|
|
_patch_membership_http(monkeypatch, _FakeResponse(200, {"state": "active"}))
|
|
|
|
assert await github_org_membership.is_user_active_org_member("alice", "langchain-ai") is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_is_user_active_org_member_pending_returns_false(monkeypatch) -> None:
|
|
from agent.utils import github_org_membership
|
|
|
|
_patch_membership_http(monkeypatch, _FakeResponse(200, {"state": "pending"}))
|
|
|
|
assert await github_org_membership.is_user_active_org_member("alice", "langchain-ai") is False
|