""" Tests for permission callback functionality """ import asyncio import pytest from copilot.rpc import ( PermissionDecisionApproveOnce, PermissionDecisionReject, PermissionDecisionUserNotAvailable, ) from copilot.session import PermissionHandler, PermissionNoResult, PermissionRequestResult from copilot.session_events import ( PermissionRequest, SessionIdleData, ToolExecutionCompleteData, ) from .testharness import E2ETestContext from .testharness.helper import read_file, write_file pytestmark = pytest.mark.asyncio(loop_scope="module") class TestPermissions: async def test_should_invoke_permission_handler_for_write_operations(self, ctx: E2ETestContext): """Test that permission handler is invoked for write operations""" permission_requests = [] def on_permission_request( request: PermissionRequest, invocation: dict ) -> PermissionRequestResult: permission_requests.append(request) assert invocation["session_id"] == session.session_id return PermissionDecisionApproveOnce() session = await ctx.client.create_session(on_permission_request=on_permission_request) write_file(ctx.work_dir, "test.txt", "original content") await session.send_and_wait("Edit test.txt and replace 'original' with 'modified'") # Should have received at least one permission request assert len(permission_requests) > 0 # Should include write permission request write_requests = [req for req in permission_requests if req.kind == "write"] assert len(write_requests) > 0 await session.disconnect() async def test_should_deny_permission_when_handler_returns_denied(self, ctx: E2ETestContext): """Test denying permissions""" def on_permission_request( request: PermissionRequest, invocation: dict ) -> PermissionRequestResult: return PermissionDecisionReject() session = await ctx.client.create_session(on_permission_request=on_permission_request) # Regression check for https://github.com/github/copilot-sdk/issues/1194: # the reject decision must round-trip through the CLI with its discriminator # intact so the agent surfaces the user-rejected error to the model. The # CLI emits a kind-specific error message ("The user rejected this tool call.") # for the reject decision, which lets us assert the decision was honored # — not merely that the operation didn't happen. user_rejected_events = [] def on_event(event): match event.data: case ToolExecutionCompleteData(success=False) as data: error = data.error msg = ( error if isinstance(error, str) else (getattr(error, "message", None) if error is not None else None) ) if msg and "user rejected" in msg.lower(): user_rejected_events.append(event) session.on(on_event) original_content = "protected content" write_file(ctx.work_dir, "protected.txt", original_content) await session.send_and_wait("Edit protected.txt and replace 'protected' with 'hacked'.") assert len(user_rejected_events) > 0 # Verify the file was NOT modified content = read_file(ctx.work_dir, "protected.txt") assert content == original_content await session.disconnect() async def test_should_deny_tool_operations_when_handler_explicitly_denies( self, ctx: E2ETestContext ): """Test that tool operations are denied when handler explicitly denies""" def deny_all(request, invocation): return PermissionDecisionUserNotAvailable() session = await ctx.client.create_session(on_permission_request=deny_all) denied_events = [] done_event = asyncio.Event() def on_event(event): match event.data: case ToolExecutionCompleteData(success=False) as data: error = data.error msg = ( error if isinstance(error, str) else (getattr(error, "message", None) if error is not None else None) ) if msg and "Permission denied" in msg: denied_events.append(event) case SessionIdleData(): done_event.set() session.on(on_event) await session.send("Run 'node --version'") await asyncio.wait_for(done_event.wait(), timeout=60) assert len(denied_events) > 0 await session.disconnect() async def test_should_deny_tool_operations_when_handler_explicitly_denies_after_resume( self, ctx: E2ETestContext ): """Test that tool operations are denied after resume when handler explicitly denies""" session1 = await ctx.client.create_session( on_permission_request=PermissionHandler.approve_all ) session_id = session1.session_id await session1.send_and_wait("What is 1+1?") def deny_all(request, invocation): return PermissionDecisionUserNotAvailable() session2 = await ctx.client.resume_session(session_id, on_permission_request=deny_all) denied_events = [] done_event = asyncio.Event() def on_event(event): match event.data: case ToolExecutionCompleteData(success=False) as data: error = data.error msg = ( error if isinstance(error, str) else (getattr(error, "message", None) if error is not None else None) ) if msg and "Permission denied" in msg: denied_events.append(event) case SessionIdleData(): done_event.set() session2.on(on_event) await session2.send("Run 'node --version'") await asyncio.wait_for(done_event.wait(), timeout=60) assert len(denied_events) > 0 await session2.disconnect() async def test_should_work_with_approve_all_permission_handler(self, ctx: E2ETestContext): """Test that sessions work with approve-all permission handler""" session = await ctx.client.create_session( on_permission_request=PermissionHandler.approve_all ) message = await session.send_and_wait("What is 2+2?") assert message is not None assert "4" in message.data.content await session.disconnect() async def test_should_handle_async_permission_handler(self, ctx: E2ETestContext): """Test async permission handler""" permission_requests = [] async def on_permission_request( request: PermissionRequest, invocation: dict ) -> PermissionRequestResult: permission_requests.append(request) await asyncio.sleep(0) return PermissionDecisionApproveOnce() session = await ctx.client.create_session(on_permission_request=on_permission_request) await session.send_and_wait("Run 'echo test' and tell me what happens") assert len(permission_requests) > 0 await session.disconnect() async def test_should_resume_session_with_permission_handler(self, ctx: E2ETestContext): """Test resuming session with permission handler""" permission_requests = [] # Create initial session session1 = await ctx.client.create_session( on_permission_request=PermissionHandler.approve_all ) session_id = session1.session_id await session1.send_and_wait("What is 1+1?") # Resume with permission handler def on_permission_request( request: PermissionRequest, invocation: dict ) -> PermissionRequestResult: permission_requests.append(request) return PermissionDecisionApproveOnce() session2 = await ctx.client.resume_session( session_id, on_permission_request=on_permission_request ) await session2.send_and_wait("Run 'echo resumed' for me") # Should have permission requests from resumed session assert len(permission_requests) > 0 await session2.disconnect() async def test_should_handle_permission_handler_errors_gracefully(self, ctx: E2ETestContext): """Test that permission handler errors are handled gracefully""" permission_request_received = asyncio.get_running_loop().create_future() def on_permission_request( request: PermissionRequest, invocation: dict ) -> PermissionRequestResult: if not permission_request_received.done(): permission_request_received.set_result(request) raise RuntimeError("Handler error") session = await ctx.client.create_session(on_permission_request=on_permission_request) try: await session.send("Run 'echo test'. If you can't, say 'failed'.") permission_request = await asyncio.wait_for( permission_request_received, timeout=30, ) assert permission_request.kind == "shell" exchanges = await ctx.wait_for_exchanges(2, timeout=60) tool_messages = [ message for exchange in exchanges for message in exchange["request"]["messages"] if message.get("role") == "tool" and "Permission denied" in str(message.get("content", "")) ] assert tool_messages assert "could not request permission" in tool_messages[-1]["content"] finally: await session.disconnect() async def test_should_receive_toolcallid_in_permission_requests(self, ctx: E2ETestContext): """Test that toolCallId is included in permission requests""" received_tool_call_id = False def on_permission_request( request: PermissionRequest, invocation: dict ) -> PermissionRequestResult: nonlocal received_tool_call_id if request.tool_call_id: received_tool_call_id = True assert isinstance(request.tool_call_id, str) assert len(request.tool_call_id) > 0 return PermissionDecisionApproveOnce() session = await ctx.client.create_session(on_permission_request=on_permission_request) await session.send_and_wait("Run 'echo test'") assert received_tool_call_id await session.disconnect() async def test_should_wait_for_slow_permission_handler(self, ctx: E2ETestContext): """Slow permission handler blocks tool execution until released.""" handler_entered: asyncio.Future = asyncio.get_event_loop().create_future() release_handler: asyncio.Future = asyncio.get_event_loop().create_future() target_tool_call_id: asyncio.Future = asyncio.get_event_loop().create_future() lifecycle: list = [] def add_event(phase: str, tool_call_id: str | None) -> None: lifecycle.append((phase, tool_call_id)) async def slow_permission(request: PermissionRequest, invocation: dict): tool_call_id = request.tool_call_id add_event("permission-start", tool_call_id) if not target_tool_call_id.done(): target_tool_call_id.set_result(tool_call_id) if not handler_entered.done(): handler_entered.set_result(True) await asyncio.wait_for(release_handler, timeout=30.0) add_event("permission-complete", tool_call_id) return PermissionDecisionApproveOnce() session = await ctx.client.create_session(on_permission_request=slow_permission) def on_event(event): if event.type.value == "tool.execution_start": add_event("tool-start", event.data.tool_call_id) elif event.type.value == "tool.execution_complete": add_event("tool-complete", event.data.tool_call_id) unsubscribe = session.on(on_event) try: asyncio.ensure_future(session.send("Run 'echo slow_handler_test'")) await asyncio.wait_for(handler_entered, timeout=30.0) target_id = await asyncio.wait_for(target_tool_call_id, timeout=30.0) # Tool should not have completed yet while handler is blocking assert not any( phase == "tool-complete" and tid == target_id for phase, tid in lifecycle ), "Tool completed before permission handler returned" release_handler.set_result(True) from .testharness.helper import get_final_assistant_message message = await get_final_assistant_message(session, timeout=60.0) perm_start = next( ( i for i, (p, tid) in enumerate(lifecycle) if p == "permission-start" and tid == target_id ), -1, ) perm_complete = next( ( i for i, (p, tid) in enumerate(lifecycle) if p == "permission-complete" and tid == target_id ), -1, ) tool_start = next( ( i for i, (p, tid) in enumerate(lifecycle) if p == "tool-start" and tid == target_id ), -1, ) tool_complete = next( ( i for i, (p, tid) in enumerate(lifecycle) if p == "tool-complete" and tid == target_id ), -1, ) assert perm_start >= 0 assert perm_complete >= 0 assert tool_start >= 0 assert tool_complete >= 0 assert perm_complete < tool_complete, ( "Expected permission completion before target tool completion" ) assert tool_start < tool_complete, ( "Expected target tool start before target tool completion" ) assert message is not None assert "slow_handler_test" in (message.data.content or "") finally: if not release_handler.done(): release_handler.set_result(True) unsubscribe() await session.disconnect() async def test_should_deny_permission_with_noresult_kind(self, ctx: E2ETestContext): """NoResult permission kind leaves legacy permission requests unanswered.""" permission_called = asyncio.get_event_loop().create_future() def deny_noresult(request: PermissionRequest, invocation: dict) -> PermissionRequestResult: if not permission_called.done(): permission_called.set_result(True) return PermissionNoResult() session = await ctx.client.create_session(on_permission_request=deny_noresult) try: asyncio.ensure_future(session.send("Run 'node --version'")) await asyncio.wait_for(permission_called, timeout=30.0) await session.abort() finally: await session.disconnect() async def test_should_short_circuit_permission_handler_when_set_approve_all_enabled( self, ctx: E2ETestContext ): """When set_approve_all is true, the runtime short-circuits the handler.""" from copilot.rpc import PermissionsSetApproveAllRequest handler_call_count = 0 def counting_handler( request: PermissionRequest, invocation: dict ) -> PermissionRequestResult: nonlocal handler_call_count handler_call_count += 1 return PermissionDecisionApproveOnce() session = await ctx.client.create_session(on_permission_request=counting_handler) try: set_result = await session.rpc.permissions.set_approve_all( PermissionsSetApproveAllRequest(enabled=True) ) assert set_result.success tool_completed: asyncio.Future = asyncio.get_event_loop().create_future() def on_event(event): if ( event.type.value == "tool.execution_complete" and event.data.success and not tool_completed.done() ): tool_completed.set_result(event) unsubscribe = session.on(on_event) try: await session.send_and_wait( "Run 'echo test' and tell me what happens", timeout=60.0 ) await asyncio.wait_for(tool_completed, timeout=30.0) assert handler_call_count == 0, ( "Handler should not have been called when approve_all is enabled" ) finally: unsubscribe() finally: try: from copilot.rpc import PermissionsSetApproveAllRequest await session.rpc.permissions.set_approve_all( PermissionsSetApproveAllRequest(enabled=False) ) except Exception as exc: # Cleanup should not hide the primary test result, but should be visible in logs. print(f"Failed to disable approve_all during cleanup: {exc!r}") await session.disconnect() async def test_should_handle_concurrent_permission_requests_from_parallel_tools( self, ctx: E2ETestContext ): """Multiple simultaneous permission requests are all handled.""" from copilot.tools import Tool, ToolInvocation, ToolResult permission_request_count = 0 both_started: asyncio.Future = asyncio.get_event_loop().create_future() first_tool_called = False second_tool_called = False async def concurrent_permission(request: PermissionRequest, invocation: dict): nonlocal permission_request_count permission_request_count += 1 if permission_request_count >= 2 and not both_started.done(): both_started.set_result(True) await asyncio.wait_for(both_started, timeout=30.0) return PermissionDecisionApproveOnce() def first_tool_handler(invocation: ToolInvocation) -> ToolResult: nonlocal first_tool_called first_tool_called = True return ToolResult( text_result_for_llm="first_permission_tool completed after permission approval", result_type="rejected", ) def second_tool_handler(invocation: ToolInvocation) -> ToolResult: nonlocal second_tool_called second_tool_called = True return ToolResult( text_result_for_llm="second_permission_tool completed after permission approval", result_type="rejected", ) session = await ctx.client.create_session( on_permission_request=concurrent_permission, tools=[ Tool( name="first_permission_tool", description="First concurrent permission test tool", parameters={"type": "object", "properties": {}}, handler=first_tool_handler, ), Tool( name="second_permission_tool", description="Second concurrent permission test tool", parameters={"type": "object", "properties": {}}, handler=second_tool_handler, ), ], ) try: idle_future: asyncio.Future = asyncio.get_event_loop().create_future() tool_completes = [] def on_event(event): if event.type.value == "tool.execution_complete" and not event.data.success: tool_completes.append(event) elif event.type.value == "session.idle" and not idle_future.done(): idle_future.set_result(True) unsubscribe = session.on(on_event) try: await session.send( "Call both first_permission_tool and second_permission_tool in the same turn." " Do not call any other tools." ) await asyncio.wait_for(both_started, timeout=30.0) await asyncio.wait_for(idle_future, timeout=60.0) assert permission_request_count == 2, ( "Expected exactly 2 permission requests (one per tool)" ) assert first_tool_called, "first_permission_tool handler should have been called" assert second_tool_called, "second_permission_tool handler should have been called" assert len(tool_completes) >= 2, ( "Expected tool.execution_complete events for both tools" ) finally: unsubscribe() finally: await session.disconnect()