mirror of
https://github.com/Sea-Haven-Industries/open-swe.git
synced 2026-09-30 20:53:15 +00:00
Adam Moussa
8d8d5bbbbf
Co-authored-by: open-swe[bot] <open-swe@users.noreply.github.com> (cherry picked from commit 1ea0e600dcc234fa5a333c6f4b80b90e2e6679d3) Co-authored-by: Johannes du Plessis <johannes@langchain.dev>
269 lines
9.5 KiB
Python
269 lines
9.5 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
|
|
import pytest
|
|
|
|
from agent.utils import auth
|
|
|
|
|
|
def test_leave_failure_comment_posts_generic_token_free_slack_notice(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
"""Slack auth failures post a generic notice, never the (possibly sensitive) message."""
|
|
monkeypatch.setenv("DASHBOARD_BASE_URL", "https://app.example.com")
|
|
thread_called: dict[str, str] = {}
|
|
|
|
async def fake_post_slack_thread_reply(channel_id: str, thread_ts: str, message: str) -> bool:
|
|
thread_called["channel_id"] = channel_id
|
|
thread_called["thread_ts"] = thread_ts
|
|
thread_called["message"] = message
|
|
return True
|
|
|
|
monkeypatch.setattr(auth, "post_slack_thread_reply", fake_post_slack_thread_reply)
|
|
monkeypatch.setattr(
|
|
auth,
|
|
"get_config",
|
|
lambda: {
|
|
"configurable": {
|
|
"slack_thread": {
|
|
"channel_id": "C123",
|
|
"thread_ts": "1.2",
|
|
"triggering_user_id": "U123",
|
|
}
|
|
}
|
|
},
|
|
)
|
|
|
|
# Pass a message that embeds a per-user auth URL; it must NOT be echoed publicly.
|
|
asyncio.run(auth.leave_failure_comment("slack", "Click https://auth.example/secret-token"))
|
|
|
|
assert thread_called["channel_id"] == "C123"
|
|
assert thread_called["thread_ts"] == "1.2"
|
|
assert "secret-token" not in thread_called["message"]
|
|
assert "https://app.example.com/my-settings" in thread_called["message"]
|
|
|
|
|
|
def _slack_config(github_login: str | None = "mason-gh") -> dict:
|
|
configurable: dict = {
|
|
"source": "slack",
|
|
"user_email": "mason@example.com",
|
|
"thread_id": "t1",
|
|
}
|
|
if github_login is not None:
|
|
configurable["github_login"] = github_login
|
|
return {"configurable": configurable}
|
|
|
|
|
|
def _stub_dashboard_store(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
*,
|
|
token: str | None,
|
|
expires_at: str | None = "2099-01-01T00:00:00Z",
|
|
cached: tuple[str | None, str | None] = (None, None),
|
|
) -> None:
|
|
from agent.dashboard import profiles
|
|
|
|
async def fake_get_from_thread(thread_id: str, **_kwargs):
|
|
return cached
|
|
|
|
async def fake_get_valid(login: str):
|
|
return token
|
|
|
|
async def fake_get_value(namespace, key):
|
|
return {"token_expires_at": expires_at}
|
|
|
|
monkeypatch.setattr(auth, "get_github_token_from_thread", fake_get_from_thread)
|
|
monkeypatch.setattr(profiles, "get_valid_access_token", fake_get_valid)
|
|
monkeypatch.setattr(profiles, "_get_value", fake_get_value)
|
|
|
|
|
|
def _set_profile(monkeypatch: pytest.MonkeyPatch, *, author_prs_as_user: bool) -> None:
|
|
"""Mock the per-user profile lookup that gates per-user attribution.
|
|
|
|
``author_prs_as_user=False`` → no opt-in (default: author as the app bot).
|
|
"""
|
|
from agent.dashboard import agent_overrides
|
|
|
|
profile = {"author_prs_as_user": True} if author_prs_as_user else None
|
|
|
|
async def fake_load_profile(login: str):
|
|
return profile
|
|
|
|
monkeypatch.setattr(agent_overrides, "load_profile", fake_load_profile)
|
|
|
|
|
|
def test_resolve_github_token_slack_defaults_to_bot(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
# DEFAULT (no author_prs_as_user opt-in): slack runs author as the app bot,
|
|
# even when a valid per-user token exists — so PRs come in as the app.
|
|
_stub_dashboard_store(monkeypatch, token="user-tok")
|
|
_set_profile(monkeypatch, author_prs_as_user=False)
|
|
monkeypatch.setattr(auth, "is_bot_token_only_mode", lambda: False)
|
|
|
|
async def fake_bot(thread_id: str):
|
|
return ("bot-tok", None)
|
|
|
|
monkeypatch.setattr(auth, "_resolve_bot_installation_token", fake_bot)
|
|
|
|
token, _ = asyncio.run(auth.resolve_github_token(_slack_config(), "t1"))
|
|
assert token == "bot-tok"
|
|
|
|
|
|
def test_resolve_github_token_slack_optin_uses_dashboard_store(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
_stub_dashboard_store(monkeypatch, token="user-tok")
|
|
_set_profile(monkeypatch, author_prs_as_user=True)
|
|
monkeypatch.setattr(auth, "is_bot_token_only_mode", lambda: False)
|
|
|
|
token, expires_at = asyncio.run(auth.resolve_github_token(_slack_config(), "t1"))
|
|
|
|
assert token == "user-tok"
|
|
assert expires_at == "2099-01-01T00:00:00Z"
|
|
|
|
|
|
def test_resolve_github_token_slack_optin_ignores_stale_thread_cache(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
# Slack thread ids are shared, so a prior user's cached token must NOT be
|
|
# returned. Resolution always goes by github_login via the dashboard store.
|
|
_stub_dashboard_store(
|
|
monkeypatch,
|
|
token="bob-token",
|
|
cached=("alice-token", "2099-01-01T00:00:00Z"),
|
|
)
|
|
_set_profile(monkeypatch, author_prs_as_user=True)
|
|
monkeypatch.setattr(auth, "is_bot_token_only_mode", lambda: False)
|
|
|
|
token, _ = asyncio.run(auth.resolve_github_token(_slack_config(), "t1"))
|
|
|
|
assert token == "bob-token"
|
|
|
|
|
|
def test_resolve_github_token_slack_optin_no_token_raises(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
_stub_dashboard_store(monkeypatch, token=None)
|
|
_set_profile(monkeypatch, author_prs_as_user=True)
|
|
monkeypatch.setattr(auth, "is_bot_token_only_mode", lambda: False)
|
|
|
|
with pytest.raises(auth.GitHubUserAuthRequired):
|
|
asyncio.run(auth.resolve_github_token(_slack_config(), "t1"))
|
|
|
|
|
|
def test_resolve_github_token_optin_per_user_wins_over_bot_only_mode(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
_stub_dashboard_store(monkeypatch, token="user-tok")
|
|
_set_profile(monkeypatch, author_prs_as_user=True)
|
|
monkeypatch.setattr(auth, "is_bot_token_only_mode", lambda: True)
|
|
|
|
async def fail_bot(thread_id: str):
|
|
raise AssertionError("bot token must not be used when the opt-in user token exists")
|
|
|
|
monkeypatch.setattr(auth, "_resolve_bot_installation_token", fail_bot)
|
|
|
|
token, _ = asyncio.run(auth.resolve_github_token(_slack_config(), "t1"))
|
|
assert token == "user-tok"
|
|
|
|
|
|
def test_resolve_github_token_slack_optin_no_token_falls_back_to_bot_in_bot_only_mode(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
_stub_dashboard_store(monkeypatch, token=None)
|
|
_set_profile(monkeypatch, author_prs_as_user=True)
|
|
monkeypatch.setattr(auth, "is_bot_token_only_mode", lambda: True)
|
|
|
|
async def fake_bot(thread_id: str):
|
|
return ("bot-tok", None)
|
|
|
|
monkeypatch.setattr(auth, "_resolve_bot_installation_token", fake_bot)
|
|
|
|
token, expires_at = asyncio.run(auth.resolve_github_token(_slack_config(), "t1"))
|
|
assert (token, expires_at) == ("bot-tok", None)
|
|
|
|
|
|
def _linear_config(github_login: str | None = "mason-gh") -> dict:
|
|
configurable: dict = {
|
|
"source": "linear",
|
|
"user_email": "mason@example.com",
|
|
"thread_id": "t1",
|
|
}
|
|
if github_login is not None:
|
|
configurable["github_login"] = github_login
|
|
return {"configurable": configurable}
|
|
|
|
|
|
def test_resolve_github_token_linear_optin_uses_dashboard_store(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
_stub_dashboard_store(monkeypatch, token="user-tok")
|
|
_set_profile(monkeypatch, author_prs_as_user=True)
|
|
monkeypatch.setattr(auth, "is_bot_token_only_mode", lambda: False)
|
|
|
|
token, expires_at = asyncio.run(auth.resolve_github_token(_linear_config(), "t1"))
|
|
|
|
assert token == "user-tok"
|
|
assert expires_at == "2099-01-01T00:00:00Z"
|
|
|
|
|
|
def test_resolve_github_token_linear_optin_no_token_raises(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
_stub_dashboard_store(monkeypatch, token=None)
|
|
_set_profile(monkeypatch, author_prs_as_user=True)
|
|
monkeypatch.setattr(auth, "is_bot_token_only_mode", lambda: False)
|
|
|
|
with pytest.raises(auth.GitHubUserAuthRequired):
|
|
asyncio.run(auth.resolve_github_token(_linear_config(), "t1"))
|
|
|
|
|
|
def test_resolve_github_token_linear_optin_no_token_falls_back_to_bot_in_bot_only_mode(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
_stub_dashboard_store(monkeypatch, token=None)
|
|
_set_profile(monkeypatch, author_prs_as_user=True)
|
|
monkeypatch.setattr(auth, "is_bot_token_only_mode", lambda: True)
|
|
|
|
async def fake_bot(thread_id: str):
|
|
return ("bot-tok", None)
|
|
|
|
monkeypatch.setattr(auth, "_resolve_bot_installation_token", fake_bot)
|
|
|
|
token, expires_at = asyncio.run(auth.resolve_github_token(_linear_config(), "t1"))
|
|
assert (token, expires_at) == ("bot-tok", None)
|
|
|
|
|
|
def test_resolve_github_token_linear_defaults_to_bot(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
# DEFAULT (no author_prs_as_user opt-in): linear runs author as the app bot,
|
|
# even when a valid per-user token and mapped login exist — so PRs can never
|
|
# be opened as a non-actor (creator/assignee).
|
|
_stub_dashboard_store(monkeypatch, token="user-tok")
|
|
_set_profile(monkeypatch, author_prs_as_user=False)
|
|
monkeypatch.setattr(auth, "is_bot_token_only_mode", lambda: False)
|
|
|
|
async def fake_bot(thread_id: str):
|
|
return ("bot-tok", None)
|
|
|
|
monkeypatch.setattr(auth, "_resolve_bot_installation_token", fake_bot)
|
|
|
|
token, _ = asyncio.run(auth.resolve_github_token(_linear_config(), "t1"))
|
|
assert token == "bot-tok"
|
|
|
|
|
|
@pytest.mark.parametrize("source", ["github"])
|
|
def test_resolve_github_token_bot_only_mode_non_slack_uses_bot(
|
|
monkeypatch: pytest.MonkeyPatch, source: str
|
|
) -> None:
|
|
monkeypatch.setattr(auth, "is_bot_token_only_mode", lambda: True)
|
|
|
|
async def fake_bot(thread_id: str):
|
|
return ("bot-tok", None)
|
|
|
|
monkeypatch.setattr(auth, "_resolve_bot_installation_token", fake_bot)
|
|
|
|
config = {"configurable": {"source": source, "github_login": "octo", "thread_id": "t1"}}
|
|
token, _ = asyncio.run(auth.resolve_github_token(config, "t1"))
|
|
assert token == "bot-tok"
|