copilot-sdk/python/test_permission_decision_context.py at fix/rust-send-and-wait-root-agent-gate · github/copilot-sdk · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
from unittest.mock import AsyncMock, MagicMock

from copilot import PermissionResponseCapability
from copilot.rpc import (
    PermissionDecisionApproveOnce,
    PermissionDecisionContext,
    PermissionDecisionOutcome,
    PermissionDecisionSource,
    PermissionDecisionSurface,
)
from copilot.session import (
    AttributedPermissionResult,
    CopilotSession,
    PermissionNoResult,
    create_attributed_permission_result,
)
from copilot.session_events import PermissionRequestRead


def test_permission_response_capability_is_exported_from_package_root() -> None:
    assert PermissionResponseCapability.INTERACTIVE.value == "interactive"


def _context() -> PermissionDecisionContext:
    return PermissionDecisionContext(
        outcome=PermissionDecisionOutcome.AUTO_APPROVED,
        source=PermissionDecisionSource.HOST_POLICY,
        surface=PermissionDecisionSurface.SDK,
    )


def _session_with_captured_rpc() -> tuple[CopilotSession, AsyncMock]:
    session = CopilotSession("session-1", client=None)
    handle = AsyncMock()
    rpc = MagicMock()
    rpc.permissions.handle_pending_permission_request = handle
    session._rpc = rpc
    return session, handle


async def test_decision_context_serialized_as_sibling_of_result() -> None:
    session, handle = _session_with_captured_rpc()
    request = PermissionRequestRead(intention="Read", path="/workspace/file.txt")

    def handler(_request, _invocation):
        return create_attributed_permission_result(PermissionDecisionApproveOnce(), _context())

    await session._execute_permission_and_respond("permission-1", request, handler)

    handle.assert_awaited_once()
    sent = handle.await_args.args[0]
    params = sent.to_dict()

    assert params["decisionContext"] == {
        "outcome": "auto_approved",
        "source": "host_policy",
        "surface": "sdk",
    }
    assert "decisionContext" not in params["result"]
    assert params["result"]["kind"] == "approve-once"


async def test_no_context_omits_decision_context_key() -> None:
    session, handle = _session_with_captured_rpc()
    request = PermissionRequestRead(intention="Read", path="/workspace/file.txt")

    def handler(_request, _invocation):
        return PermissionDecisionApproveOnce()

    await session._execute_permission_and_respond("permission-1", request, handler)

    handle.assert_awaited_once()
    params = handle.await_args.args[0].to_dict()

    assert "decisionContext" not in params
    assert params["result"]["kind"] == "approve-once"


def test_attributed_result_replaces_rather_than_nests() -> None:
    first = PermissionDecisionContext(
        outcome=PermissionDecisionOutcome.PROMPTED_USER,
        source=PermissionDecisionSource.HUMAN_RESPONSE,
        surface=PermissionDecisionSurface.TUI,
    )
    second = _context()

    once_wrapped = create_attributed_permission_result(PermissionDecisionApproveOnce(), first)
    twice_wrapped = create_attributed_permission_result(once_wrapped, second)

    assert isinstance(twice_wrapped, AttributedPermissionResult)
    assert isinstance(twice_wrapped.result, PermissionDecisionApproveOnce)
    assert twice_wrapped.decision_context is second


async def test_no_result_with_context_still_suppresses_response() -> None:
    session, handle = _session_with_captured_rpc()
    request = PermissionRequestRead(intention="Read", path="/workspace/file.txt")

    def handler(_request, _invocation):
        return create_attributed_permission_result(PermissionNoResult(), _context())

    await session._execute_permission_and_respond("permission-1", request, handler)

    handle.assert_not_awaited()