|
2 | 2 |
|
3 | 3 | import pytest |
4 | 4 |
|
5 | | -from acp.contrib.permissions import PermissionBroker, default_permission_options |
| 5 | +from acp.contrib.permissions import MissingPermissionOptionsError, PermissionBroker, default_permission_options |
6 | 6 | from acp.contrib.tool_calls import ToolCallTracker |
7 | 7 | from acp.schema import ( |
8 | 8 | AllowedOutcome, |
@@ -58,6 +58,60 @@ async def requester(request: RequestPermissionRequest): |
58 | 58 | assert recorded == ["allow"] |
59 | 59 |
|
60 | 60 |
|
| 61 | +@pytest.mark.asyncio |
| 62 | +async def test_permission_broker_none_uses_standard_options(): |
| 63 | + tracker = ToolCallTracker(id_factory=lambda: "standard") |
| 64 | + tracker.start("external", title="Standard options") |
| 65 | + recorded: list[list[str]] = [] |
| 66 | + |
| 67 | + async def requester(request: RequestPermissionRequest): |
| 68 | + recorded.append([option.option_id for option in request.options]) |
| 69 | + return RequestPermissionResponse(outcome=AllowedOutcome(option_id="approve", outcome="selected")) |
| 70 | + |
| 71 | + broker = PermissionBroker("session", requester, tracker=tracker, default_options=None) |
| 72 | + await broker.request_for("external", options=None) |
| 73 | + assert recorded == [["approve", "approve_for_session", "reject"]] |
| 74 | + |
| 75 | + |
| 76 | +@pytest.mark.asyncio |
| 77 | +async def test_permission_broker_custom_default_and_override(): |
| 78 | + tracker = ToolCallTracker(id_factory=lambda: "reject-only") |
| 79 | + tracker.start("external", title="Reject only") |
| 80 | + reject = PermissionOption(option_id="reject", name="Reject", kind="reject_once") |
| 81 | + allow = PermissionOption(option_id="allow", name="Allow", kind="allow_once") |
| 82 | + recorded: list[list[str]] = [] |
| 83 | + |
| 84 | + async def requester(request: RequestPermissionRequest): |
| 85 | + recorded.append([option.option_id for option in request.options]) |
| 86 | + return RequestPermissionResponse( |
| 87 | + outcome=AllowedOutcome(option_id=request.options[0].option_id, outcome="selected") |
| 88 | + ) |
| 89 | + |
| 90 | + broker = PermissionBroker("session", requester, tracker=tracker, default_options=[reject]) |
| 91 | + await broker.request_for("external") |
| 92 | + await broker.request_for("external", options=[allow]) |
| 93 | + assert recorded == [["reject"], ["allow"]] |
| 94 | + |
| 95 | + |
| 96 | +@pytest.mark.asyncio |
| 97 | +@pytest.mark.parametrize(("default_options", "options"), [(None, []), ([], None)]) |
| 98 | +async def test_permission_broker_rejects_empty_option_sources(default_options, options): |
| 99 | + tracker = ToolCallTracker(id_factory=lambda: "empty") |
| 100 | + tracker.start("external", title="No options") |
| 101 | + recorded: list[RequestPermissionRequest] = [] |
| 102 | + |
| 103 | + async def requester(request: RequestPermissionRequest): |
| 104 | + recorded.append(request) |
| 105 | + return RequestPermissionResponse( |
| 106 | + outcome=AllowedOutcome(option_id=request.options[0].option_id, outcome="selected") |
| 107 | + ) |
| 108 | + |
| 109 | + broker = PermissionBroker("session", requester, tracker=tracker, default_options=default_options) |
| 110 | + with pytest.raises(MissingPermissionOptionsError, match="requires at least one permission option"): |
| 111 | + await broker.request_for("external", options=options) |
| 112 | + assert recorded == [] |
| 113 | + |
| 114 | + |
61 | 115 | def test_default_permission_options_shape(): |
62 | 116 | options = default_permission_options() |
63 | 117 | assert len(options) == 3 |
|
0 commit comments