diff --git a/tests/test_workflow_push_guard.py b/tests/test_workflow_push_guard.py index 8334a0fc..ac38e1ea 100644 --- a/tests/test_workflow_push_guard.py +++ b/tests/test_workflow_push_guard.py @@ -1,6 +1,7 @@ from __future__ import annotations import json +import os from typing import Any import pytest @@ -347,15 +348,25 @@ async def test_unapproved_workflow_push_blocks_and_posts_slack( assert posted["blocks"][1]["elements"][0]["value"] -async def test_approved_workflow_push_runs_fixed_command( +async def test_approved_workflow_push_elevates_and_restores( monkeypatch: pytest.MonkeyPatch, ) -> None: guard.SANDBOX_BACKENDS["thread-1"] = _Backend() + refreshed: list[dict[str, str]] = [] async def fake_approved(thread_id: str, fingerprint: str) -> bool: return True + async def fake_refresh(thread_id: str | None, *, permissions: dict[str, str]) -> bool: + refreshed.append(dict(permissions)) + return True + + async def fake_find_approval(*args: Any, **kwargs: Any) -> dict[str, Any] | None: + return None + monkeypatch.setattr(guard, "workflow_push_approved", fake_approved) + monkeypatch.setattr(guard, "find_workflow_push_approval", fake_find_approval) + monkeypatch.setattr(guard, "refresh_proxy_token", fake_refresh) pushed_command = "" @@ -372,6 +383,40 @@ async def test_approved_workflow_push_runs_fixed_command( pushed_command == "git -C /repo push origin aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa:refs/heads/feature" ) + assert refreshed[0]["workflows"] == "write" + assert "workflows" not in refreshed[1] + assert refreshed[1]["actions"] == "read" + + +async def test_workflow_push_restoration_falls_back_when_actions_read_unavailable( + monkeypatch: pytest.MonkeyPatch, +) -> None: + guard.SANDBOX_BACKENDS["thread-1"] = _Backend() + refreshed: list[dict[str, str]] = [] + + async def fake_approved(thread_id: str, fingerprint: str) -> bool: + return True + + async def fake_refresh(thread_id: str | None, *, permissions: dict[str, str]) -> bool: + refreshed.append(dict(permissions)) + return "actions" not in permissions + + async def fake_find_approval(*args: Any, **kwargs: Any) -> dict[str, Any] | None: + return None + + monkeypatch.setattr(guard, "workflow_push_approved", fake_approved) + monkeypatch.setattr(guard, "find_workflow_push_approval", fake_find_approval) + monkeypatch.setattr(guard, "refresh_proxy_token", fake_refresh) + + async def handler(_request: Any) -> ToolMessage: + return ToolMessage(content="pushed", tool_call_id="call-1") + + await guard.WorkflowPushGuardMiddleware().awrap_tool_call(_Request(), handler) + + assert refreshed[0]["workflows"] == "write" + assert refreshed[1]["actions"] == "read" + assert refreshed[2] == guard.BASE_RUNTIME_PROXY_TOKEN_PERMISSIONS + assert "actions" not in refreshed[2] async def test_non_workflow_push_runs_without_approval(monkeypatch: pytest.MonkeyPatch) -> None: @@ -665,7 +710,7 @@ async def test_exact_rejection_checked_before_stale_approval( assert payload["workflow_approval_status"] == "rejected" -async def test_approved_workflow_push_passes_through( +async def test_approved_workflow_push_aborts_when_elevation_fails( monkeypatch: pytest.MonkeyPatch, ) -> None: guard.SANDBOX_BACKENDS["thread-1"] = _Backend() @@ -673,7 +718,14 @@ async def test_approved_workflow_push_passes_through( async def fake_approved(thread_id: str, fingerprint: str) -> bool: return True + refresh_calls: list[dict[str, str]] = [] + + async def fake_refresh(*args: Any, **kwargs: Any) -> bool: + refresh_calls.append(dict(kwargs.get("permissions", {}))) + return False + monkeypatch.setattr(guard, "workflow_push_approved", fake_approved) + monkeypatch.setattr(guard, "refresh_proxy_token", fake_refresh) called = False request = _Request() @@ -685,9 +737,51 @@ async def test_approved_workflow_push_passes_through( result = await guard.WorkflowPushGuardMiddleware().awrap_tool_call(request, handler) + assert called is False + assert isinstance(result, ToolMessage) + assert result.tool_call_id == "call-1" + assert result.status == "error" + payload = json.loads(str(result.content)) + assert payload["status"] == "error" + assert payload["error_type"] == "WorkflowPushElevationFailed" + assert "workflows-scoped token" in payload["error"] + assert len(refresh_calls) == 1 + assert refresh_calls[0].get("workflows") == "write" + + +async def test_approved_workflow_push_runs_on_non_langsmith_providers( + monkeypatch: pytest.MonkeyPatch, +) -> None: + guard.SANDBOX_BACKENDS["thread-1"] = _Backend() + + async def fake_approved(thread_id: str, fingerprint: str) -> bool: + return True + + refresh_calls: list[dict[str, str]] = [] + + async def fake_refresh(*args: Any, **kwargs: Any) -> bool: + refresh_calls.append(dict(kwargs.get("permissions", {}))) + return False + + monkeypatch.setattr(guard, "workflow_push_approved", fake_approved) + monkeypatch.setattr(guard, "refresh_proxy_token", fake_refresh) + monkeypatch.setattr(guard, "os", os) + + called = False + + async def handler(request: Any) -> ToolMessage: + nonlocal called + called = True + return ToolMessage(content="pushed", tool_call_id=request.tool_call["id"]) + + with monkeypatch.context() as mp: + mp.setenv("SANDBOX_TYPE", "local") + result = await guard.WorkflowPushGuardMiddleware().awrap_tool_call(_Request(), handler) + assert called is True assert isinstance(result, ToolMessage) assert result.content == "pushed" + assert refresh_calls == [] def test_parse_git_push_guards_wrapped_and_optioned_forms() -> None: