FilesExpand file tree

main

/

test_client_start.py

Copy path

Blame

More file actions

Blame

More file actions

Latest commit

 

History

History

History

95 lines (72 loc) · 2.89 KB

main

/

test_client_start.py

Copy path

Top

File metadata and controls

• Code
• Blame

95 lines (72 loc) · 2.89 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

"""Startup concurrency regressions without a live CLI runtime."""

import asyncio

from unittest.mock import AsyncMock, Mock

import pytest

from copilot import CopilotClient, RuntimeConnection

@pytest.fixture

def client():

client = CopilotClient(connection=RuntimeConnection.for_stdio(path="unused-cli"))

client._start_cli_server = AsyncMock()

client._connect_to_server = AsyncMock()

client._verify_protocol_version = AsyncMock()

return client

@pytest.mark.parametrize(

"phase", ["_start_cli_server", "_connect_to_server", "_verify_protocol_version"]

)

async def test_concurrent_start_initializes_transport_once(client, phase):

entered = asyncio.Event()

release = asyncio.Event()

async def block():

entered.set()

await release.wait()

getattr(client, phase).side_effect = block

first = asyncio.create_task(client.start())

await asyncio.wait_for(entered.wait(), timeout=1)

second = asyncio.create_task(client.start())

try:

await asyncio.sleep(0)

assert not second.done()

finally:

release.set()

await asyncio.wait_for(asyncio.gather(first, second), timeout=1)

client._start_cli_server.assert_awaited_once()

client._connect_to_server.assert_awaited_once()

client._verify_protocol_version.assert_awaited_once()

await client.start()

client._start_cli_server.assert_awaited_once()

async def test_start_can_retry_after_failure(client):

client._start_cli_server.side_effect = [RuntimeError("startup failed"), None]

with pytest.raises(RuntimeError, match="startup failed"):

await client.start()

await asyncio.wait_for(client.start(), timeout=1)

assert client._start_cli_server.await_count == 2

client._connect_to_server.assert_awaited_once()

client._verify_protocol_version.assert_awaited_once()

async def test_cancelling_waiting_start_does_not_cancel_active_start(client):

entered = asyncio.Event()

release = asyncio.Event()

async def block():

entered.set()

await release.wait()

client._start_cli_server.side_effect = block

first = asyncio.create_task(client.start())

await asyncio.wait_for(entered.wait(), timeout=1)

second = asyncio.create_task(client.start())

try:

await asyncio.sleep(0)

second.cancel()

with pytest.raises(asyncio.CancelledError):

_ = await second

assert not first.done()

finally:

release.set()

await asyncio.wait_for(first, timeout=1)

await client.start()

client._start_cli_server.assert_awaited_once()

client._verify_protocol_version.assert_awaited_once()

@pytest.mark.parametrize("method", ["stop", "force_stop"])

async def test_shutdown_cancels_pending_inference_requests(client, method):

adapter = Mock()

client._llm_inference_adapter = adapter

await getattr(client, method)()

adapter.cancel_pending.assert_called_once_with()