FilesExpand file tree

main

/

test_commands.py

Copy path

Blame

More file actions

Blame

More file actions

Latest commit

 

History

History

History

212 lines (165 loc) · 6.92 KB

main

/

test_commands.py

Copy path

Top

File metadata and controls

• Code
• Blame

212 lines (165 loc) · 6.92 KB

Raw

Copy raw file

Download raw file

Open symbols panel

Edit and raw actions

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

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

145

146

147

148

149

150

151

152

153

154

155

156

157

158

159

160

161

162

163

164

165

166

167

168

169

170

171

172

173

174

175

176

177

178

179

180

181

182

183

184

185

186

187

188

189

190

191

192

193

194

195

196

197

198

199

200

201

202

203

204

205

206

207

208

209

210

211

212

"""E2E Commands Tests

Mirrors nodejs/test/e2e/commands.test.ts

Multi-client test: a second client joining a session with commands should

trigger a ``commands.changed`` broadcast event visible to the first client.

"""

import asyncio

import os

import shutil

import tempfile

import pytest

import pytest_asyncio

from copilot import CopilotClient

from copilot.client import ExternalServerConfig, SubprocessConfig

from copilot.session import CommandDefinition, PermissionHandler

from .testharness.context import SNAPSHOTS_DIR, get_cli_path_for_tests

from .testharness.proxy import CapiProxy

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

# ---------------------------------------------------------------------------

# Multi-client context (TCP mode) — same pattern as test_multi_client.py

# ---------------------------------------------------------------------------

class CommandsMultiClientContext:

"""Test context that manages two clients connected to the same CLI server."""

def __init__(self):

self.cli_path: str = ""

self.home_dir: str = ""

self.work_dir: str = ""

self.proxy_url: str = ""

self._proxy: CapiProxy | None = None

self._client1: CopilotClient | None = None

self._client2: CopilotClient | None = None

async def setup(self):

self.cli_path = get_cli_path_for_tests()

self.home_dir = tempfile.mkdtemp(prefix="copilot-cmd-config-")

self.work_dir = tempfile.mkdtemp(prefix="copilot-cmd-work-")

self._proxy = CapiProxy()

self.proxy_url = await self._proxy.start()

github_token = (

"fake-token-for-e2e-tests" if os.environ.get("GITHUB_ACTIONS") == "true" else None

)

# Client 1 uses TCP mode so a second client can connect

self._client1 = CopilotClient(

SubprocessConfig(

cli_path=self.cli_path,

cwd=self.work_dir,

env=self._get_env(),

use_stdio=False,

github_token=github_token,

)

)

# Trigger connection to get the port

init_session = await self._client1.create_session(

on_permission_request=PermissionHandler.approve_all,

)

await init_session.disconnect()

actual_port = self._client1.actual_port

assert actual_port is not None

self._client2 = CopilotClient(ExternalServerConfig(url=f"localhost:{actual_port}"))

async def teardown(self, test_failed: bool = False):

for c in (self._client2, self._client1):

if c:

try:

await c.stop()

except Exception:

pass # Best-effort cleanup during teardown

self._client1 = self._client2 = None

if self._proxy:

await self._proxy.stop(skip_writing_cache=test_failed)

self._proxy = None

for d in (self.home_dir, self.work_dir):

if d and os.path.exists(d):

shutil.rmtree(d, ignore_errors=True)

async def configure_for_test(self, test_file: str, test_name: str):

import re

sanitized_name = re.sub(r"[^a-zA-Z0-9]", "_", test_name).lower()

snapshot_path = SNAPSHOTS_DIR / test_file / f"{sanitized_name}.yaml"

if self._proxy:

await self._proxy.configure(str(snapshot_path.resolve()), self.work_dir)

from pathlib import Path

for d in (self.home_dir, self.work_dir):

for item in Path(d).iterdir():

if item.is_dir():

shutil.rmtree(item, ignore_errors=True)

else:

item.unlink(missing_ok=True)

def _get_env(self) -> dict:

env = os.environ.copy()

env.update(

{

"COPILOT_API_URL": self.proxy_url,

"XDG_CONFIG_HOME": self.home_dir,

"XDG_STATE_HOME": self.home_dir,

}

)

return env

@property

def client1(self) -> CopilotClient:

assert self._client1 is not None

return self._client1

@property

def client2(self) -> CopilotClient:

assert self._client2 is not None

return self._client2

# ---------------------------------------------------------------------------

# Fixtures

# ---------------------------------------------------------------------------

@pytest.hookimpl(tryfirst=True, hookwrapper=True)

def pytest_runtest_makereport(item, call):

outcome = yield

rep = outcome.get_result()

if rep.when == "call" and rep.failed:

item.session.stash.setdefault("any_test_failed", False)

item.session.stash["any_test_failed"] = True

@pytest_asyncio.fixture(scope="module", loop_scope="module")

async def mctx(request):

context = CommandsMultiClientContext()

await context.setup()

yield context

any_failed = request.session.stash.get("any_test_failed", False)

await context.teardown(test_failed=any_failed)

@pytest_asyncio.fixture(autouse=True, loop_scope="module")

async def configure_cmd_test(request, mctx):

test_name = request.node.name

if test_name.startswith("test_"):

test_name = test_name[5:]

await mctx.configure_for_test("multi_client", test_name)

yield

# ---------------------------------------------------------------------------

# Tests

# ---------------------------------------------------------------------------

class TestCommands:

async def test_client_receives_commands_changed_when_another_client_joins(

self, mctx: CommandsMultiClientContext

):

"""Client receives commands.changed when another client joins with commands."""

# Client 1 creates a session without commands

session1 = await mctx.client1.create_session(

on_permission_request=PermissionHandler.approve_all,

)

# Listen for the commands.changed event

commands_changed = asyncio.Event()

commands_data: dict = {}

def on_event(event):

if event.type.value == "commands.changed":

commands_data["commands"] = getattr(event.data, "commands", None)

commands_changed.set()

session1.on(on_event)

# Client 2 joins the same session with commands

session2 = await mctx.client2.resume_session(

session1.session_id,

on_permission_request=PermissionHandler.approve_all,

commands=[

CommandDefinition(

name="deploy",

description="Deploy the app",

handler=lambda ctx: None,

),

],

)

# Wait for the commands.changed event (with timeout)

await asyncio.wait_for(commands_changed.wait(), timeout=15.0)

# Verify the event contains the deploy command

assert commands_data.get("commands") is not None

cmd_names = [c.name for c in commands_data["commands"]]

assert "deploy" in cmd_names

await session2.disconnect()