fix: align workflow-push-guard tests with dev's transient-elevation impl

The dev merge auto-combined dev's elevation tests with the stale passthrough
tests inherited from the durable-dispatch branch; the passthrough tests
contradict dev's restored _run_with_workflow_token impl. Take dev's test file.
This commit is contained in:
Adam Moussa 2026-07-09 17:20:16 -04:00
parent 3c25fac52b
commit a094ab099b
No known key found for this signature in database

View file

@ -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: