copilot-sdk/python/e2e/test_rpc_commands_e2e.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
105
106
107
108
109
110
111
112
113
114
115
116
117
"""E2E coverage for session.commands RPC methods."""

from __future__ import annotations

import pytest

from copilot.rpc import (
    CommandsInvokeRequest,
    CommandsRespondToQueuedCommandRequest,
    ExecuteCommandParams,
    QueuedCommandHandled,
    SessionCommandsListRequest,
    SlashCommandKind,
    SlashCommandTextResult,
)
from copilot.session import CommandContext, CommandDefinition, PermissionHandler

from .testharness import E2ETestContext

pytestmark = pytest.mark.asyncio(loop_scope="module")


class TestRpcCommands:
    async def test_should_list_builtin_and_client_commands(self, ctx: E2ETestContext):
        session = await ctx.client.create_session(
            on_permission_request=PermissionHandler.approve_all,
            commands=[
                CommandDefinition(
                    name="deploy",
                    description="Deploy the app",
                    handler=lambda _: None,
                )
            ],
        )
        try:
            commands = await session.rpc.commands.list(SessionCommandsListRequest())
            by_name = {command.name: command for command in commands.commands}

            builtins = [
                command for command in commands.commands if command.kind == SlashCommandKind.BUILTIN
            ]
            assert builtins
            if "model" in by_name:
                assert by_name["model"].kind == SlashCommandKind.BUILTIN
            if "compact" in by_name:
                assert by_name["compact"].kind == SlashCommandKind.BUILTIN

            assert "deploy" in by_name
            assert by_name["deploy"].kind == SlashCommandKind.CLIENT
            assert by_name["deploy"].description == "Deploy the app"
        finally:
            await session.disconnect()

    async def test_should_invoke_builtin_model_command(self, ctx: E2ETestContext):
        session = await ctx.client.create_session(
            on_permission_request=PermissionHandler.approve_all,
        )
        try:
            result = await session.rpc.commands.invoke(CommandsInvokeRequest(name="model"))
            assert result is not None
            if isinstance(result, SlashCommandTextResult):
                assert result.text.strip()
            else:
                assert getattr(result, "kind", None) in {
                    "agent-prompt",
                    "completed",
                    "select-subcommand",
                    "text",
                }
        finally:
            await session.disconnect()

    async def test_should_execute_registered_command_with_arguments(self, ctx: E2ETestContext):
        calls: list[CommandContext] = []

        def deploy(context: CommandContext) -> None:
            calls.append(context)

        session = await ctx.client.create_session(
            on_permission_request=PermissionHandler.approve_all,
            commands=[
                CommandDefinition(
                    name="deploy",
                    description="Deploy the app",
                    handler=deploy,
                )
            ],
        )
        try:
            result = await session.rpc.commands.execute(
                ExecuteCommandParams(command_name="deploy", args="production")
            )
            assert result.error is None
            assert len(calls) == 1
            assert calls[0].session_id == session.session_id
            assert calls[0].command_name == "deploy"
            assert calls[0].args == "production"
            assert calls[0].command == "/deploy production"
        finally:
            await session.disconnect()

    async def test_should_return_false_for_unknown_queued_command_response(
        self, ctx: E2ETestContext
    ):
        session = await ctx.client.create_session(
            on_permission_request=PermissionHandler.approve_all,
        )
        try:
            result = await session.rpc.commands.respond_to_queued_command(
                CommandsRespondToQueuedCommandRequest(
                    request_id="missing-queued-command",
                    result=QueuedCommandHandled(stop_processing_queue=True),
                )
            )
            assert result.success is False
        finally:
            await session.disconnect()