""" Tests for permission callback functionality """ import asyncio import pytest from copilot.session import PermissionHandler, PermissionRequest, PermissionRequestResult 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 PermissionRequestResult(kind="approved") 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.value == "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 PermissionRequestResult(kind="denied-interactively-by-user") session = await ctx.client.create_session(on_permission_request=on_permission_request) 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'.") # 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 PermissionRequestResult() session = await ctx.client.create_session(on_permission_request=deny_all) denied_events = [] done_event = asyncio.Event() def on_event(event): if event.type.value == "tool.execution_complete" and event.data.success is False: error = event.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) elif event.type.value == "session.idle": 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 PermissionRequestResult() session2 = await ctx.client.resume_session(session_id, on_permission_request=deny_all) denied_events = [] done_event = asyncio.Event() def on_event(event): if event.type.value == "tool.execution_complete" and event.data.success is False: error = event.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) elif event.type.value == "session.idle": 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) # Simulate async permission check (e.g., user prompt) await asyncio.sleep(0.01) return PermissionRequestResult(kind="approved") 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 PermissionRequestResult(kind="approved") 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""" def on_permission_request( request: PermissionRequest, invocation: dict ) -> PermissionRequestResult: raise RuntimeError("Handler error") session = await ctx.client.create_session(on_permission_request=on_permission_request) message = await session.send_and_wait("Run 'echo test'. If you can't, say 'failed'.") # Should handle the error and deny permission assert message is not None content_lower = message.data.content.lower() assert any(word in content_lower for word in ["fail", "cannot", "unable", "permission"]) 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 PermissionRequestResult(kind="approved") 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()