test_permissions.py

"""
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()