FilesExpand file tree

main

/

test_scenario_cloud_e2e.py

Copy path

Blame

More file actions

Blame

More file actions

Latest commit

 

History

History

History

159 lines (141 loc) · 6.23 KB

main

/

test_scenario_cloud_e2e.py

Copy path

Top

File metadata and controls

• Code
• Blame

159 lines (141 loc) · 6.23 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

"""Scenario-parity E2Es for cloud connection and remote-steering workflows."""

from __future__ import annotations

import asyncio

import pytest

from copilot import CloudSessionOptions, CloudSessionRepository

from copilot.rpc import ConnectRemoteSessionParams, RemoteNotifySteerableChangedRequest

from copilot.session import PermissionHandler

from copilot.session_events import (

AssistantMessageData,

SessionRemoteSteerableChangedData,

SessionStartData,

)

from ._scenario_fake_cli import create_scenario_client, read_scenario_capture

from .testharness import E2ETestContext

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

class TestScenarioCloud:

async def test_should_notify_steerability_before_first_send_without_remote_enable(

self,

ctx: E2ETestContext,

):

client, capture_path = create_scenario_client(ctx, "send")

events = []

steerability_received = asyncio.Event()

def on_event(event) -> None:

events.append(event)

if isinstance(event.data, SessionRemoteSteerableChangedData):

steerability_received.set()

try:

session = await client.create_session(

session_id="scenario-session",

on_event=on_event,

on_permission_request=PermissionHandler.approve_all,

)

try:

await session.rpc.remote.notify_steerable_changed(

RemoteNotifySteerableChangedRequest(remote_steerable=True)

)

response = await session.send_and_wait("Send the first cloud message.")

await asyncio.wait_for(steerability_received.wait(), timeout=5)

assert response is not None

assert isinstance(response.data, AssistantMessageData)

assert response.data.content == "scenario response"

remote_event = next(

event

for event in events

if isinstance(event.data, SessionRemoteSteerableChangedData)

)

assert remote_event.data.remote_steerable is True

methods = [

request["method"] for request in read_scenario_capture(capture_path)["requests"]

]

assert methods.index("session.remote.notifySteerableChanged") < methods.index(

"session.send"

)

assert "session.remote.enable" not in methods

finally:

await session.disconnect()

finally:

await client.stop()

async def test_should_route_cloud_session_start_for_server_assigned_session_id(

self,

ctx: E2ETestContext,

):

client, _capture_path = create_scenario_client(ctx, "cloud")

start_events = []

session_start_received = asyncio.Event()

def on_event(event) -> None:

if isinstance(event.data, SessionStartData):

start_events.append(event)

session_start_received.set()

try:

session = await client.create_session(

cloud=CloudSessionOptions(

repository=CloudSessionRepository(

owner="github",

name="copilot-sdk",

branch="scenario-branch",

)

),

on_event=on_event,

on_permission_request=PermissionHandler.approve_all,

)

try:

await asyncio.wait_for(session_start_received.wait(), timeout=5)

assert session.session_id == "cloud-runtime-session"

assert len(start_events) == 1

assert start_events[0].data.session_id == session.session_id

assert start_events[0].data.producer == "scenario-fake-cli"

assert start_events[0].data.remote_steerable is False

finally:

await session.disconnect()

finally:

await client.stop()

async def test_should_resume_using_runtime_id_returned_by_cloud_connect(

self,

ctx: E2ETestContext,

):

client, capture_path = create_scenario_client(ctx, "cloud-connect")

try:

await client.start()

connection = await client.rpc.sessions.connect(

ConnectRemoteSessionParams(session_id="remote-resource-id")

)

session = await client.resume_session(

connection.session_id,

on_permission_request=PermissionHandler.approve_all,

)

try:

assert connection.session_id == "runtime-session-id"

assert session.session_id == "runtime-session-id"

requests = read_scenario_capture(capture_path)["requests"]

resume = next(

request for request in requests if request["method"] == "session.resume"

)

assert resume["params"]["sessionId"] == "runtime-session-id"

finally:

await session.disconnect()

finally:

await client.stop()

async def test_should_expose_cloud_resource_mismatch_before_resume(

self,

ctx: E2ETestContext,

):

client, _capture_path = create_scenario_client(ctx, "cloud-connect")

try:

await client.start()

connection = await client.rpc.sessions.connect(

ConnectRemoteSessionParams(session_id="remote-resource-id")

)

assert connection.session_id == "runtime-session-id"

assert connection.metadata.session_id == "remote-resource-id"

assert connection.metadata.resource_id == "remote-resource-id"

assert connection.metadata.session_id != connection.session_id

assert connection.metadata.repository.owner == "github"

assert connection.metadata.repository.name == "copilot-sdk"

assert connection.metadata.repository.branch == "scenario-branch"

assert connection.metadata.pull_request_number == 42

assert connection.metadata.state == "running"

assert connection.metadata.summary == "Remote task summary"

finally:

await client.stop()