mirror of
https://github.com/Sea-Haven-Industries/open-swe.git
synced 2026-09-30 11:33:14 +00:00
Adam Moussa
ae1f883b4c
Applies the plan's C5 step: git mv every test per the domain-reorg
move-map (movemap-m50.txt) into tests/{agent,analyzer,auth,dashboard,
github,middleware,models,reviewer,sandbox,slack,tools,webhooks}/, plus
the 13 fork-only placements from the scoping report §2c (Atlassian
webhook tests -> tests/webhooks/, test_atlassian_connect.py and
test_auth_error_leak.py -> tests/auth/, jira/confluence util tests ->
tests/tools/, test_repo_binding_isolation.py -> tests/sandbox/,
bot-identity/autofix tests -> tests/github/).
Path-only move: the only content edits are parents[1] -> parents[2]
fixes in test_e2b_integration.py and test_daytona_integration.py,
required because their __file__-relative ROOT path gained one more
directory level in the move.
Monkeypatch retargets for these files were already completed in C4;
none remained outstanding here.
305 lines
10 KiB
Python
305 lines
10 KiB
Python
from __future__ import annotations
|
|
|
|
import logging
|
|
from datetime import UTC, datetime, timedelta
|
|
from typing import Any
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from agent.utils import github_app
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _clear_token_cache() -> Any:
|
|
github_app.clear_app_token_cache()
|
|
yield
|
|
github_app.clear_app_token_cache()
|
|
|
|
|
|
class _FakeResponse:
|
|
def raise_for_status(self) -> None:
|
|
pass
|
|
|
|
def json(self) -> dict[str, str]:
|
|
return {"token": "token", "expires_at": "expires"}
|
|
|
|
|
|
class _FakeAsyncClient:
|
|
last_post: dict[str, Any] | None = None
|
|
|
|
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 _FakeResponse()
|
|
|
|
|
|
def _configure(monkeypatch: pytest.MonkeyPatch, client_cls: type) -> None:
|
|
monkeypatch.setattr(github_app, "GITHUB_APP_ID", "1")
|
|
monkeypatch.setattr(github_app, "GITHUB_APP_PRIVATE_KEY", "key")
|
|
monkeypatch.setattr(github_app, "GITHUB_APP_INSTALLATION_ID", "2")
|
|
monkeypatch.setattr(github_app, "_generate_app_jwt", lambda: "jwt")
|
|
monkeypatch.setattr(github_app.httpx, "AsyncClient", client_cls)
|
|
|
|
|
|
class _CountingResponse:
|
|
def __init__(self, expires_at: str) -> None:
|
|
self._expires_at = expires_at
|
|
|
|
def raise_for_status(self) -> None:
|
|
pass
|
|
|
|
def json(self) -> dict[str, str]:
|
|
return {"token": "tok-123", "expires_at": self._expires_at}
|
|
|
|
|
|
class _CountingClient:
|
|
posts = 0
|
|
expires_at = "2099-01-01T00:00:00Z"
|
|
|
|
def __init__(self, **kwargs: Any) -> None:
|
|
pass
|
|
|
|
async def __aenter__(self) -> _CountingClient:
|
|
return self
|
|
|
|
async def __aexit__(self, exc_type: object, exc: object, tb: object) -> None:
|
|
return None
|
|
|
|
async def post(self, url: str, **kwargs: Any) -> _CountingResponse:
|
|
type(self).posts += 1
|
|
return _CountingResponse(type(self).expires_at)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_token_is_cached_until_near_expiry(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
future = (datetime.now(UTC) + timedelta(hours=1)).isoformat()
|
|
|
|
class Client(_CountingClient):
|
|
posts = 0
|
|
expires_at = future
|
|
|
|
_configure(monkeypatch, Client)
|
|
|
|
t1, _ = await github_app.get_github_app_installation_token_with_expiry()
|
|
t2, _ = await github_app.get_github_app_installation_token_with_expiry()
|
|
|
|
assert t1 == t2 == "tok-123"
|
|
assert Client.posts == 1 # second call served from the in-process cache
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cache_is_scoped_per_repository_set(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
future = (datetime.now(UTC) + timedelta(hours=1)).isoformat()
|
|
|
|
class Client(_CountingClient):
|
|
posts = 0
|
|
expires_at = future
|
|
|
|
_configure(monkeypatch, Client)
|
|
|
|
await github_app.get_github_app_installation_token_with_expiry(repositories=["a"])
|
|
await github_app.get_github_app_installation_token_with_expiry(repositories=["b"])
|
|
await github_app.get_github_app_installation_token_with_expiry(repositories=["a"])
|
|
|
|
assert Client.posts == 2 # distinct scopes mint separately; the repeat is cached
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_near_expiry_token_is_not_cached(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
soon = (datetime.now(UTC) + timedelta(minutes=2)).isoformat()
|
|
|
|
class Client(_CountingClient):
|
|
posts = 0
|
|
expires_at = soon
|
|
|
|
_configure(monkeypatch, Client)
|
|
|
|
await github_app.get_github_app_installation_token_with_expiry()
|
|
await github_app.get_github_app_installation_token_with_expiry()
|
|
|
|
assert Client.posts == 2 # within the safety margin -> re-minted every call
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_installation_token_can_be_scoped_to_repository_ids(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
monkeypatch.setattr(github_app, "GITHUB_APP_ID", "1")
|
|
monkeypatch.setattr(github_app, "GITHUB_APP_PRIVATE_KEY", "key")
|
|
monkeypatch.setattr(github_app, "GITHUB_APP_INSTALLATION_ID", "2")
|
|
monkeypatch.setattr(github_app, "_generate_app_jwt", lambda: "jwt")
|
|
monkeypatch.setattr(github_app.httpx, "AsyncClient", _FakeAsyncClient)
|
|
|
|
token, expires_at = await github_app.get_github_app_installation_token_with_expiry(
|
|
repository_ids=[123]
|
|
)
|
|
|
|
assert token == "token"
|
|
assert expires_at == "expires"
|
|
assert _FakeAsyncClient.last_post is not None
|
|
assert _FakeAsyncClient.last_post["json"] == {"repository_ids": [123]}
|
|
|
|
|
|
def test_runtime_proxy_token_permissions_include_optional_read_only_actions() -> None:
|
|
assert "actions" not in github_app.CORE_RUNTIME_PROXY_TOKEN_PERMISSIONS
|
|
assert github_app.RUNTIME_PROXY_TOKEN_PERMISSIONS["actions"] == "read"
|
|
assert github_app.RUNTIME_PROXY_TOKEN_PERMISSIONS.get("actions") != "write"
|
|
assert "actions" not in github_app.WORKFLOW_RUNTIME_PROXY_TOKEN_PERMISSIONS
|
|
|
|
|
|
def test_workflows_write_is_never_in_the_standing_scope() -> None:
|
|
"""workflows:write is guard-only; the standing token must never carry it, so
|
|
token scope stays a backstop for the workflow-push HITL approval control."""
|
|
assert "workflows" not in github_app.RUNTIME_PROXY_TOKEN_PERMISSIONS
|
|
assert "workflows" not in github_app.CORE_RUNTIME_PROXY_TOKEN_PERMISSIONS
|
|
assert github_app.WORKFLOW_RUNTIME_PROXY_TOKEN_PERMISSIONS["workflows"] == "write"
|
|
assert all("workflows" not in scope for scope in github_app.PROXY_TOKEN_PERMISSION_LADDER)
|
|
|
|
|
|
def test_core_proxy_token_permissions_exclude_optional_grants() -> None:
|
|
"""The terminal ladder rung must only ask for install-time permissions."""
|
|
core = github_app.CORE_RUNTIME_PROXY_TOKEN_PERMISSIONS
|
|
assert "workflows" not in core
|
|
assert "actions" not in core
|
|
assert core["contents"] == "write"
|
|
|
|
|
|
def test_proxy_token_ladder_descends_to_core() -> None:
|
|
"""Ladder goes most→least privileged so a missing grant degrades gracefully."""
|
|
ladder = github_app.PROXY_TOKEN_PERMISSION_LADDER
|
|
assert ladder == (
|
|
github_app.RUNTIME_PROXY_TOKEN_PERMISSIONS,
|
|
github_app.CORE_RUNTIME_PROXY_TOKEN_PERMISSIONS,
|
|
)
|
|
assert [len(scope) for scope in ladder] == sorted(
|
|
(len(scope) for scope in ladder), reverse=True
|
|
)
|
|
|
|
|
|
class _HTTPStatusErrorClient:
|
|
"""Raises an ``HTTPStatusError`` with a configurable status on mint."""
|
|
|
|
status = 500
|
|
|
|
def __init__(self, **kwargs: Any) -> None:
|
|
pass
|
|
|
|
async def __aenter__(self) -> _HTTPStatusErrorClient:
|
|
return self
|
|
|
|
async def __aexit__(self, exc_type: object, exc: object, tb: object) -> None:
|
|
return None
|
|
|
|
async def post(self, url: str, **kwargs: Any) -> Any:
|
|
request = httpx.Request("POST", url)
|
|
response = httpx.Response(type(self).status, request=request)
|
|
raise httpx.HTTPStatusError("mint failed", request=request, response=response)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_missing_grant_422_descends_quietly(
|
|
monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
|
|
) -> None:
|
|
"""A 422 (ungranted permission) is the ladder's expected descend signal, so it
|
|
must not be logged as a transient failure even on a non-terminal rung."""
|
|
|
|
class Client(_HTTPStatusErrorClient):
|
|
status = 422
|
|
|
|
_configure(monkeypatch, Client)
|
|
|
|
with caplog.at_level(logging.DEBUG, logger="agent.utils.github_app"):
|
|
token, _ = await github_app.get_github_app_installation_token_with_expiry(
|
|
permissions={"actions": "read"}, log_errors=False
|
|
)
|
|
|
|
assert token is None
|
|
assert not any(r.levelno >= logging.WARNING for r in caplog.records)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_transient_mint_error_warns_even_when_errors_suppressed(
|
|
monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
|
|
) -> None:
|
|
"""A non-422 failure is not a missing grant; it must surface at WARNING even on
|
|
a non-terminal rung so a blip doesn't silently downscope a whole run."""
|
|
|
|
class Client(_HTTPStatusErrorClient):
|
|
status = 503
|
|
|
|
_configure(monkeypatch, Client)
|
|
|
|
with caplog.at_level(logging.DEBUG, logger="agent.utils.github_app"):
|
|
token, _ = await github_app.get_github_app_installation_token_with_expiry(
|
|
permissions={"actions": "read"}, log_errors=False
|
|
)
|
|
|
|
assert token is None
|
|
assert any(r.levelno == logging.WARNING for r in caplog.records)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_installation_token_includes_permissions(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setattr(github_app, "GITHUB_APP_ID", "1")
|
|
monkeypatch.setattr(github_app, "GITHUB_APP_PRIVATE_KEY", "key")
|
|
monkeypatch.setattr(github_app, "GITHUB_APP_INSTALLATION_ID", "2")
|
|
monkeypatch.setattr(github_app, "_generate_app_jwt", lambda: "jwt")
|
|
monkeypatch.setattr(github_app.httpx, "AsyncClient", _FakeAsyncClient)
|
|
|
|
await github_app.get_github_app_installation_token_with_expiry(
|
|
repositories=["open-swe"], permissions={"workflows": "write", "contents": "write"}
|
|
)
|
|
|
|
assert _FakeAsyncClient.last_post is not None
|
|
assert _FakeAsyncClient.last_post["json"] == {
|
|
"repositories": ["open-swe"],
|
|
"permissions": {"contents": "write", "workflows": "write"},
|
|
}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cache_is_scoped_per_permission_set(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
future = (datetime.now(UTC) + timedelta(hours=1)).isoformat()
|
|
|
|
class Client(_CountingClient):
|
|
posts = 0
|
|
expires_at = future
|
|
|
|
_configure(monkeypatch, Client)
|
|
|
|
await github_app.get_github_app_installation_token_with_expiry(
|
|
permissions={"contents": "write"}
|
|
)
|
|
await github_app.get_github_app_installation_token_with_expiry(
|
|
permissions={"contents": "write", "workflows": "write"}
|
|
)
|
|
await github_app.get_github_app_installation_token_with_expiry(
|
|
permissions={"contents": "write"}
|
|
)
|
|
|
|
assert Client.posts == 2
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_installation_token_omits_scope_for_full_installation(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
monkeypatch.setattr(github_app, "GITHUB_APP_ID", "1")
|
|
monkeypatch.setattr(github_app, "GITHUB_APP_PRIVATE_KEY", "key")
|
|
monkeypatch.setattr(github_app, "GITHUB_APP_INSTALLATION_ID", "2")
|
|
monkeypatch.setattr(github_app, "_generate_app_jwt", lambda: "jwt")
|
|
monkeypatch.setattr(github_app.httpx, "AsyncClient", _FakeAsyncClient)
|
|
|
|
await github_app.get_github_app_installation_token_with_expiry()
|
|
|
|
assert _FakeAsyncClient.last_post is not None
|
|
assert _FakeAsyncClient.last_post["json"] is None
|