mirror of
https://github.com/anthropics/claude-code-sdk-python.git
synced 2025-12-23 09:19:52 +00:00
9dd6fe5f91
## Problem ClaudeSDKClient initialization would hang indefinitely on Windows, timing out after 60 seconds. The SDK successfully spawned the Claude CLI subprocess but control requests sent via stdin never reached the subprocess due to Windows subprocess stdin buffering behavior with Python's asyncio. ## Root Cause On Windows, when using asyncio subprocess streams, data written to stdin can remain buffered and not immediately sent to the child process. The CLI subprocess waits for the initialization request that's stuck in Python's buffer, causing the 60-second timeout. ## Solution 1. Added `flush_stdin()` method to Transport base class (non-abstract, default no-op) 2. Implemented Windows-specific flush in SubprocessCLITransport that calls `drain()` on the asyncio StreamWriter when available 3. Call `flush_stdin()` after all control protocol writes in Query class: - After sending control requests (_send_control_request) - After responding to incoming requests (_handle_control_request) ## Tests Added - test_flush_stdin_on_windows: Verifies drain() called on Windows - test_flush_stdin_on_non_windows: Verifies no-op on other platforms - test_flush_stdin_without_process: Tests graceful handling of missing process - test_flush_stdin_fallback_to_inner_stream: Tests wrapped stream fallback - test_flush_stdin_called_after_control_requests: Integration test for outgoing requests - test_flush_stdin_called_after_control_responses: Integration test for incoming requests All tests pass on macOS, and the fix is platform-specific to Windows only. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <noreply@anthropic.com>
956 lines
37 KiB
Python
956 lines
37 KiB
Python
"""Tests for ClaudeSDKClient streaming functionality and query() with async iterables."""
|
|
|
|
import asyncio
|
|
import json
|
|
import sys
|
|
import tempfile
|
|
from pathlib import Path
|
|
from unittest.mock import AsyncMock, Mock, patch
|
|
|
|
import anyio
|
|
import pytest
|
|
|
|
from claude_agent_sdk import (
|
|
AssistantMessage,
|
|
ClaudeAgentOptions,
|
|
ClaudeSDKClient,
|
|
CLIConnectionError,
|
|
PermissionResultAllow,
|
|
ResultMessage,
|
|
TextBlock,
|
|
UserMessage,
|
|
query,
|
|
)
|
|
from claude_agent_sdk._internal.query import Query
|
|
from claude_agent_sdk._internal.transport import Transport
|
|
from claude_agent_sdk._internal.transport.subprocess_cli import SubprocessCLITransport
|
|
|
|
|
|
def create_mock_transport(with_init_response=True):
|
|
"""Create a properly configured mock transport.
|
|
|
|
Args:
|
|
with_init_response: If True, automatically respond to initialization request
|
|
"""
|
|
mock_transport = AsyncMock()
|
|
mock_transport.connect = AsyncMock()
|
|
mock_transport.close = AsyncMock()
|
|
mock_transport.end_input = AsyncMock()
|
|
mock_transport.write = AsyncMock()
|
|
mock_transport.flush_stdin = AsyncMock()
|
|
mock_transport.is_ready = Mock(return_value=True)
|
|
|
|
# Track written messages to simulate control protocol responses
|
|
written_messages = []
|
|
|
|
async def mock_write(data):
|
|
written_messages.append(data)
|
|
|
|
mock_transport.write.side_effect = mock_write
|
|
|
|
# Default read_messages to handle control protocol
|
|
async def control_protocol_generator():
|
|
# Wait for initialization request if needed
|
|
if with_init_response:
|
|
# Wait a bit for the write to happen
|
|
await asyncio.sleep(0.01)
|
|
|
|
# Check if initialization was requested
|
|
for msg_str in written_messages:
|
|
try:
|
|
msg = json.loads(msg_str.strip())
|
|
if (
|
|
msg.get("type") == "control_request"
|
|
and msg.get("request", {}).get("subtype") == "initialize"
|
|
):
|
|
# Send initialization response
|
|
yield {
|
|
"type": "control_response",
|
|
"response": {
|
|
"request_id": msg.get("request_id"),
|
|
"subtype": "success",
|
|
"commands": [],
|
|
"output_style": "default",
|
|
},
|
|
}
|
|
break
|
|
except (json.JSONDecodeError, KeyError, AttributeError):
|
|
pass
|
|
|
|
# Keep checking for other control requests (like interrupt)
|
|
last_check = len(written_messages)
|
|
timeout_counter = 0
|
|
while timeout_counter < 100: # Avoid infinite loop
|
|
await asyncio.sleep(0.01)
|
|
timeout_counter += 1
|
|
|
|
# Check for new messages
|
|
for msg_str in written_messages[last_check:]:
|
|
try:
|
|
msg = json.loads(msg_str.strip())
|
|
if msg.get("type") == "control_request":
|
|
subtype = msg.get("request", {}).get("subtype")
|
|
if subtype == "interrupt":
|
|
# Send interrupt response
|
|
yield {
|
|
"type": "control_response",
|
|
"response": {
|
|
"request_id": msg.get("request_id"),
|
|
"subtype": "success",
|
|
},
|
|
}
|
|
return # End after interrupt
|
|
except (json.JSONDecodeError, KeyError, AttributeError):
|
|
pass
|
|
last_check = len(written_messages)
|
|
|
|
# Then end the stream
|
|
return
|
|
|
|
mock_transport.read_messages = control_protocol_generator
|
|
return mock_transport
|
|
|
|
|
|
class TestClaudeSDKClientStreaming:
|
|
"""Test ClaudeSDKClient streaming functionality."""
|
|
|
|
def test_auto_connect_with_context_manager(self):
|
|
"""Test automatic connection when using context manager."""
|
|
|
|
async def _test():
|
|
with patch(
|
|
"claude_agent_sdk._internal.transport.subprocess_cli.SubprocessCLITransport"
|
|
) as mock_transport_class:
|
|
mock_transport = create_mock_transport()
|
|
mock_transport_class.return_value = mock_transport
|
|
|
|
async with ClaudeSDKClient() as client:
|
|
# Verify connect was called
|
|
mock_transport.connect.assert_called_once()
|
|
assert client._transport is mock_transport
|
|
|
|
# Verify disconnect was called on exit
|
|
mock_transport.close.assert_called_once()
|
|
|
|
anyio.run(_test)
|
|
|
|
def test_manual_connect_disconnect(self):
|
|
"""Test manual connect and disconnect."""
|
|
|
|
async def _test():
|
|
with patch(
|
|
"claude_agent_sdk._internal.transport.subprocess_cli.SubprocessCLITransport"
|
|
) as mock_transport_class:
|
|
mock_transport = create_mock_transport()
|
|
mock_transport_class.return_value = mock_transport
|
|
|
|
client = ClaudeSDKClient()
|
|
await client.connect()
|
|
|
|
# Verify connect was called
|
|
mock_transport.connect.assert_called_once()
|
|
assert client._transport is mock_transport
|
|
|
|
await client.disconnect()
|
|
# Verify disconnect was called
|
|
mock_transport.close.assert_called_once()
|
|
assert client._transport is None
|
|
|
|
anyio.run(_test)
|
|
|
|
def test_connect_with_string_prompt(self):
|
|
"""Test connecting with a string prompt."""
|
|
|
|
async def _test():
|
|
with patch(
|
|
"claude_agent_sdk._internal.transport.subprocess_cli.SubprocessCLITransport"
|
|
) as mock_transport_class:
|
|
mock_transport = create_mock_transport()
|
|
mock_transport_class.return_value = mock_transport
|
|
|
|
client = ClaudeSDKClient()
|
|
await client.connect("Hello Claude")
|
|
|
|
# Verify transport was created with string prompt
|
|
call_kwargs = mock_transport_class.call_args.kwargs
|
|
assert call_kwargs["prompt"] == "Hello Claude"
|
|
|
|
anyio.run(_test)
|
|
|
|
def test_connect_with_async_iterable(self):
|
|
"""Test connecting with an async iterable."""
|
|
|
|
async def _test():
|
|
with patch(
|
|
"claude_agent_sdk._internal.transport.subprocess_cli.SubprocessCLITransport"
|
|
) as mock_transport_class:
|
|
mock_transport = create_mock_transport()
|
|
mock_transport_class.return_value = mock_transport
|
|
|
|
async def message_stream():
|
|
yield {"type": "user", "message": {"role": "user", "content": "Hi"}}
|
|
yield {
|
|
"type": "user",
|
|
"message": {"role": "user", "content": "Bye"},
|
|
}
|
|
|
|
client = ClaudeSDKClient()
|
|
stream = message_stream()
|
|
await client.connect(stream)
|
|
|
|
# Verify transport was created with async iterable
|
|
call_kwargs = mock_transport_class.call_args.kwargs
|
|
# Should be the same async iterator
|
|
assert call_kwargs["prompt"] is stream
|
|
|
|
anyio.run(_test)
|
|
|
|
def test_query(self):
|
|
"""Test sending a query."""
|
|
|
|
async def _test():
|
|
with patch(
|
|
"claude_agent_sdk._internal.transport.subprocess_cli.SubprocessCLITransport"
|
|
) as mock_transport_class:
|
|
mock_transport = create_mock_transport()
|
|
mock_transport_class.return_value = mock_transport
|
|
|
|
async with ClaudeSDKClient() as client:
|
|
await client.query("Test message")
|
|
|
|
# Verify write was called with correct format
|
|
# Should have at least 2 writes: init request and user message
|
|
assert mock_transport.write.call_count >= 2
|
|
|
|
# Find the user message in the write calls
|
|
user_msg_found = False
|
|
for call in mock_transport.write.call_args_list:
|
|
data = call[0][0]
|
|
try:
|
|
msg = json.loads(data.strip())
|
|
if msg.get("type") == "user":
|
|
assert msg["message"]["content"] == "Test message"
|
|
assert msg["session_id"] == "default"
|
|
user_msg_found = True
|
|
break
|
|
except (json.JSONDecodeError, KeyError, AttributeError):
|
|
pass
|
|
assert user_msg_found, "User message not found in write calls"
|
|
|
|
anyio.run(_test)
|
|
|
|
def test_send_message_with_session_id(self):
|
|
"""Test sending a message with custom session ID."""
|
|
|
|
async def _test():
|
|
with patch(
|
|
"claude_agent_sdk._internal.transport.subprocess_cli.SubprocessCLITransport"
|
|
) as mock_transport_class:
|
|
mock_transport = create_mock_transport()
|
|
mock_transport_class.return_value = mock_transport
|
|
|
|
async with ClaudeSDKClient() as client:
|
|
await client.query("Test", session_id="custom-session")
|
|
|
|
# Find the user message with custom session ID
|
|
session_found = False
|
|
for call in mock_transport.write.call_args_list:
|
|
data = call[0][0]
|
|
try:
|
|
msg = json.loads(data.strip())
|
|
if msg.get("type") == "user":
|
|
assert msg["session_id"] == "custom-session"
|
|
session_found = True
|
|
break
|
|
except (json.JSONDecodeError, KeyError, AttributeError):
|
|
pass
|
|
assert session_found, "User message with custom session not found"
|
|
|
|
anyio.run(_test)
|
|
|
|
def test_send_message_not_connected(self):
|
|
"""Test sending message when not connected raises error."""
|
|
|
|
async def _test():
|
|
client = ClaudeSDKClient()
|
|
with pytest.raises(CLIConnectionError, match="Not connected"):
|
|
await client.query("Test")
|
|
|
|
anyio.run(_test)
|
|
|
|
def test_receive_messages(self):
|
|
"""Test receiving messages."""
|
|
|
|
async def _test():
|
|
with patch(
|
|
"claude_agent_sdk._internal.transport.subprocess_cli.SubprocessCLITransport"
|
|
) as mock_transport_class:
|
|
mock_transport = create_mock_transport()
|
|
mock_transport_class.return_value = mock_transport
|
|
|
|
# Mock the message stream with control protocol support
|
|
async def mock_receive():
|
|
# First handle initialization
|
|
await asyncio.sleep(0.01)
|
|
written = mock_transport.write.call_args_list
|
|
for call in written:
|
|
data = call[0][0]
|
|
try:
|
|
msg = json.loads(data.strip())
|
|
if (
|
|
msg.get("type") == "control_request"
|
|
and msg.get("request", {}).get("subtype")
|
|
== "initialize"
|
|
):
|
|
yield {
|
|
"type": "control_response",
|
|
"response": {
|
|
"request_id": msg.get("request_id"),
|
|
"subtype": "success",
|
|
"commands": [],
|
|
"output_style": "default",
|
|
},
|
|
}
|
|
break
|
|
except (json.JSONDecodeError, KeyError, AttributeError):
|
|
pass
|
|
|
|
# Then yield the actual messages
|
|
yield {
|
|
"type": "assistant",
|
|
"message": {
|
|
"role": "assistant",
|
|
"content": [{"type": "text", "text": "Hello!"}],
|
|
"model": "claude-opus-4-1-20250805",
|
|
},
|
|
}
|
|
yield {
|
|
"type": "user",
|
|
"message": {"role": "user", "content": "Hi there"},
|
|
}
|
|
|
|
mock_transport.read_messages = mock_receive
|
|
|
|
async with ClaudeSDKClient() as client:
|
|
messages = []
|
|
async for msg in client.receive_messages():
|
|
messages.append(msg)
|
|
if len(messages) == 2:
|
|
break
|
|
|
|
assert len(messages) == 2
|
|
assert isinstance(messages[0], AssistantMessage)
|
|
assert isinstance(messages[0].content[0], TextBlock)
|
|
assert messages[0].content[0].text == "Hello!"
|
|
assert isinstance(messages[1], UserMessage)
|
|
assert messages[1].content == "Hi there"
|
|
|
|
anyio.run(_test)
|
|
|
|
def test_receive_response(self):
|
|
"""Test receive_response stops at ResultMessage."""
|
|
|
|
async def _test():
|
|
with patch(
|
|
"claude_agent_sdk._internal.transport.subprocess_cli.SubprocessCLITransport"
|
|
) as mock_transport_class:
|
|
mock_transport = create_mock_transport()
|
|
mock_transport_class.return_value = mock_transport
|
|
|
|
# Mock the message stream with control protocol support
|
|
async def mock_receive():
|
|
# First handle initialization
|
|
await asyncio.sleep(0.01)
|
|
written = mock_transport.write.call_args_list
|
|
for call in written:
|
|
data = call[0][0]
|
|
try:
|
|
msg = json.loads(data.strip())
|
|
if (
|
|
msg.get("type") == "control_request"
|
|
and msg.get("request", {}).get("subtype")
|
|
== "initialize"
|
|
):
|
|
yield {
|
|
"type": "control_response",
|
|
"response": {
|
|
"request_id": msg.get("request_id"),
|
|
"subtype": "success",
|
|
"commands": [],
|
|
"output_style": "default",
|
|
},
|
|
}
|
|
break
|
|
except (json.JSONDecodeError, KeyError, AttributeError):
|
|
pass
|
|
|
|
# Then yield the actual messages
|
|
yield {
|
|
"type": "assistant",
|
|
"message": {
|
|
"role": "assistant",
|
|
"content": [{"type": "text", "text": "Answer"}],
|
|
"model": "claude-opus-4-1-20250805",
|
|
},
|
|
}
|
|
yield {
|
|
"type": "result",
|
|
"subtype": "success",
|
|
"duration_ms": 1000,
|
|
"duration_api_ms": 800,
|
|
"is_error": False,
|
|
"num_turns": 1,
|
|
"session_id": "test",
|
|
"total_cost_usd": 0.001,
|
|
}
|
|
# This should not be yielded
|
|
yield {
|
|
"type": "assistant",
|
|
"message": {
|
|
"role": "assistant",
|
|
"content": [
|
|
{"type": "text", "text": "Should not see this"}
|
|
],
|
|
},
|
|
"model": "claude-opus-4-1-20250805",
|
|
}
|
|
|
|
mock_transport.read_messages = mock_receive
|
|
|
|
async with ClaudeSDKClient() as client:
|
|
messages = []
|
|
async for msg in client.receive_response():
|
|
messages.append(msg)
|
|
|
|
# Should only get 2 messages (assistant + result)
|
|
assert len(messages) == 2
|
|
assert isinstance(messages[0], AssistantMessage)
|
|
assert isinstance(messages[1], ResultMessage)
|
|
|
|
anyio.run(_test)
|
|
|
|
def test_interrupt(self):
|
|
"""Test interrupt functionality."""
|
|
|
|
async def _test():
|
|
with patch(
|
|
"claude_agent_sdk._internal.transport.subprocess_cli.SubprocessCLITransport"
|
|
) as mock_transport_class:
|
|
mock_transport = create_mock_transport()
|
|
mock_transport_class.return_value = mock_transport
|
|
|
|
async with ClaudeSDKClient() as client:
|
|
# Interrupt is now handled via control protocol
|
|
await client.interrupt()
|
|
# Check that a control request was sent via write
|
|
write_calls = mock_transport.write.call_args_list
|
|
interrupt_found = False
|
|
for call in write_calls:
|
|
data = call[0][0]
|
|
try:
|
|
msg = json.loads(data.strip())
|
|
if (
|
|
msg.get("type") == "control_request"
|
|
and msg.get("request", {}).get("subtype") == "interrupt"
|
|
):
|
|
interrupt_found = True
|
|
break
|
|
except (json.JSONDecodeError, KeyError, AttributeError):
|
|
pass
|
|
assert interrupt_found, "Interrupt control request not found"
|
|
|
|
anyio.run(_test)
|
|
|
|
def test_interrupt_not_connected(self):
|
|
"""Test interrupt when not connected raises error."""
|
|
|
|
async def _test():
|
|
client = ClaudeSDKClient()
|
|
with pytest.raises(CLIConnectionError, match="Not connected"):
|
|
await client.interrupt()
|
|
|
|
anyio.run(_test)
|
|
|
|
def test_client_with_options(self):
|
|
"""Test client initialization with options."""
|
|
|
|
async def _test():
|
|
options = ClaudeAgentOptions(
|
|
cwd="/custom/path",
|
|
allowed_tools=["Read", "Write"],
|
|
system_prompt="Be helpful",
|
|
)
|
|
|
|
with patch(
|
|
"claude_agent_sdk._internal.transport.subprocess_cli.SubprocessCLITransport"
|
|
) as mock_transport_class:
|
|
mock_transport = create_mock_transport()
|
|
mock_transport_class.return_value = mock_transport
|
|
|
|
client = ClaudeSDKClient(options=options)
|
|
await client.connect()
|
|
|
|
# Verify options were passed to transport
|
|
call_kwargs = mock_transport_class.call_args.kwargs
|
|
assert call_kwargs["options"] is options
|
|
|
|
anyio.run(_test)
|
|
|
|
def test_concurrent_send_receive(self):
|
|
"""Test concurrent sending and receiving messages."""
|
|
|
|
async def _test():
|
|
with patch(
|
|
"claude_agent_sdk._internal.transport.subprocess_cli.SubprocessCLITransport"
|
|
) as mock_transport_class:
|
|
mock_transport = create_mock_transport()
|
|
mock_transport_class.return_value = mock_transport
|
|
|
|
# Mock receive to wait then yield messages with control protocol support
|
|
async def mock_receive():
|
|
# First handle initialization
|
|
await asyncio.sleep(0.01)
|
|
written = mock_transport.write.call_args_list
|
|
for call in written:
|
|
if call:
|
|
data = call[0][0]
|
|
try:
|
|
msg = json.loads(data.strip())
|
|
if (
|
|
msg.get("type") == "control_request"
|
|
and msg.get("request", {}).get("subtype")
|
|
== "initialize"
|
|
):
|
|
yield {
|
|
"type": "control_response",
|
|
"response": {
|
|
"request_id": msg.get("request_id"),
|
|
"subtype": "success",
|
|
"commands": [],
|
|
"output_style": "default",
|
|
},
|
|
}
|
|
break
|
|
except (json.JSONDecodeError, KeyError, AttributeError):
|
|
pass
|
|
|
|
# Then yield the actual messages
|
|
await asyncio.sleep(0.1)
|
|
yield {
|
|
"type": "assistant",
|
|
"message": {
|
|
"role": "assistant",
|
|
"content": [{"type": "text", "text": "Response 1"}],
|
|
"model": "claude-opus-4-1-20250805",
|
|
},
|
|
}
|
|
await asyncio.sleep(0.1)
|
|
yield {
|
|
"type": "result",
|
|
"subtype": "success",
|
|
"duration_ms": 1000,
|
|
"duration_api_ms": 800,
|
|
"is_error": False,
|
|
"num_turns": 1,
|
|
"session_id": "test",
|
|
"total_cost_usd": 0.001,
|
|
}
|
|
|
|
mock_transport.read_messages = mock_receive
|
|
|
|
async with ClaudeSDKClient() as client:
|
|
# Helper to get next message
|
|
async def get_next_message():
|
|
return await client.receive_response().__anext__()
|
|
|
|
# Start receiving in background
|
|
receive_task = asyncio.create_task(get_next_message())
|
|
|
|
# Send message while receiving
|
|
await client.query("Question 1")
|
|
|
|
# Wait for first message
|
|
first_msg = await receive_task
|
|
assert isinstance(first_msg, AssistantMessage)
|
|
|
|
anyio.run(_test)
|
|
|
|
def test_flush_stdin_called_after_control_responses(self):
|
|
"""Test that flush_stdin is called after responding to control requests (issue #208)."""
|
|
|
|
async def _test():
|
|
# Create a mock transport
|
|
mock_transport = AsyncMock(spec=Transport)
|
|
mock_transport.is_ready = Mock(return_value=True)
|
|
|
|
# Track write and flush calls
|
|
write_calls = []
|
|
flush_calls = []
|
|
|
|
async def mock_write(data):
|
|
write_calls.append(data)
|
|
|
|
async def mock_flush():
|
|
flush_calls.append(True)
|
|
|
|
mock_transport.write = AsyncMock(side_effect=mock_write)
|
|
mock_transport.flush_stdin = AsyncMock(side_effect=mock_flush)
|
|
|
|
# Create mock read_messages that doesn't yield anything
|
|
async def mock_read_messages():
|
|
# Just wait forever (test will complete before this matters)
|
|
await asyncio.sleep(1000)
|
|
yield {}
|
|
|
|
mock_transport.read_messages = mock_read_messages
|
|
|
|
# Create Query with streaming mode
|
|
query = Query(transport=mock_transport, is_streaming_mode=True)
|
|
await query.start()
|
|
|
|
# Simulate an incoming tool permission request
|
|
permission_request = {
|
|
"type": "control_request",
|
|
"request_id": "test_req_123",
|
|
"request": {
|
|
"subtype": "can_use_tool",
|
|
"tool_name": "Read",
|
|
"input": {"file_path": "/test.txt"},
|
|
"permission_suggestions": [],
|
|
},
|
|
}
|
|
|
|
# Set up a permission callback that allows the tool
|
|
async def mock_can_use_tool(tool_name, input_data, context):
|
|
return PermissionResultAllow()
|
|
|
|
query.can_use_tool = mock_can_use_tool
|
|
|
|
# Clear previous calls
|
|
write_calls.clear()
|
|
flush_calls.clear()
|
|
|
|
# Handle the control request
|
|
await query._handle_control_request(permission_request)
|
|
|
|
# Give it a moment to complete
|
|
await asyncio.sleep(0.01)
|
|
|
|
# Verify that flush_stdin was called after writing the response
|
|
assert len(write_calls) == 1, "Should have written one control response"
|
|
assert len(flush_calls) == 1, (
|
|
"flush_stdin should be called after writing response"
|
|
)
|
|
|
|
# Verify the response was a success
|
|
response_data = json.loads(write_calls[0])
|
|
assert response_data["type"] == "control_response"
|
|
assert response_data["response"]["subtype"] == "success"
|
|
|
|
await query.close()
|
|
|
|
anyio.run(_test)
|
|
|
|
|
|
class TestQueryWithAsyncIterable:
|
|
"""Test query() function with async iterable inputs."""
|
|
|
|
def test_query_with_async_iterable(self):
|
|
"""Test query with async iterable of messages."""
|
|
|
|
async def _test():
|
|
async def message_stream():
|
|
yield {"type": "user", "message": {"role": "user", "content": "First"}}
|
|
yield {"type": "user", "message": {"role": "user", "content": "Second"}}
|
|
|
|
# Create a simple test script that validates stdin and outputs a result
|
|
with tempfile.NamedTemporaryFile(mode="w", suffix=".py", delete=False) as f:
|
|
test_script = f.name
|
|
f.write("""#!/usr/bin/env python3
|
|
import sys
|
|
import json
|
|
|
|
# Read stdin messages
|
|
stdin_messages = []
|
|
while True:
|
|
line = sys.stdin.readline()
|
|
if not line:
|
|
break
|
|
|
|
try:
|
|
msg = json.loads(line.strip())
|
|
# Handle control requests
|
|
if msg.get("type") == "control_request":
|
|
request_id = msg.get("request_id")
|
|
request = msg.get("request", {})
|
|
|
|
# Send control response for initialize
|
|
if request.get("subtype") == "initialize":
|
|
response = {
|
|
"type": "control_response",
|
|
"response": {
|
|
"subtype": "success",
|
|
"request_id": request_id,
|
|
"response": {
|
|
"commands": [],
|
|
"output_style": "default"
|
|
}
|
|
}
|
|
}
|
|
print(json.dumps(response))
|
|
sys.stdout.flush()
|
|
else:
|
|
stdin_messages.append(line.strip())
|
|
except:
|
|
stdin_messages.append(line.strip())
|
|
|
|
# Verify we got 2 user messages
|
|
assert len(stdin_messages) == 2
|
|
assert '"First"' in stdin_messages[0]
|
|
assert '"Second"' in stdin_messages[1]
|
|
|
|
# Output a valid result
|
|
print('{"type": "result", "subtype": "success", "duration_ms": 100, "duration_api_ms": 50, "is_error": false, "num_turns": 1, "session_id": "test", "total_cost_usd": 0.001}')
|
|
""")
|
|
|
|
# Make script executable (Unix-style systems)
|
|
if sys.platform != "win32":
|
|
Path(test_script).chmod(0o755)
|
|
|
|
try:
|
|
# Mock _find_cli to return the test script path directly
|
|
with patch.object(
|
|
SubprocessCLITransport, "_find_cli", return_value=test_script
|
|
):
|
|
# Mock _build_command to properly execute Python script
|
|
original_build_command = SubprocessCLITransport._build_command
|
|
|
|
def mock_build_command(self):
|
|
# Get original command
|
|
cmd = original_build_command(self)
|
|
# On Windows, we need to use python interpreter to run the script
|
|
if sys.platform == "win32":
|
|
# Replace first element with python interpreter and script
|
|
cmd[0:1] = [sys.executable, test_script]
|
|
else:
|
|
# On Unix, just use the script directly
|
|
cmd[0] = test_script
|
|
return cmd
|
|
|
|
with patch.object(
|
|
SubprocessCLITransport, "_build_command", mock_build_command
|
|
):
|
|
# Run query with async iterable
|
|
messages = []
|
|
async for msg in query(prompt=message_stream()):
|
|
messages.append(msg)
|
|
|
|
# Should get the result message
|
|
assert len(messages) == 1
|
|
assert isinstance(messages[0], ResultMessage)
|
|
assert messages[0].subtype == "success"
|
|
finally:
|
|
# Clean up
|
|
Path(test_script).unlink()
|
|
|
|
anyio.run(_test)
|
|
|
|
|
|
class TestClaudeSDKClientEdgeCases:
|
|
"""Test edge cases and error scenarios."""
|
|
|
|
def test_receive_messages_not_connected(self):
|
|
"""Test receiving messages when not connected."""
|
|
|
|
async def _test():
|
|
client = ClaudeSDKClient()
|
|
with pytest.raises(CLIConnectionError, match="Not connected"):
|
|
async for _ in client.receive_messages():
|
|
pass
|
|
|
|
anyio.run(_test)
|
|
|
|
def test_receive_response_not_connected(self):
|
|
"""Test receive_response when not connected."""
|
|
|
|
async def _test():
|
|
client = ClaudeSDKClient()
|
|
with pytest.raises(CLIConnectionError, match="Not connected"):
|
|
async for _ in client.receive_response():
|
|
pass
|
|
|
|
anyio.run(_test)
|
|
|
|
def test_double_connect(self):
|
|
"""Test connecting twice."""
|
|
|
|
async def _test():
|
|
with patch(
|
|
"claude_agent_sdk._internal.transport.subprocess_cli.SubprocessCLITransport"
|
|
) as mock_transport_class:
|
|
# Create a new mock transport for each call
|
|
mock_transport_class.side_effect = [
|
|
create_mock_transport(),
|
|
create_mock_transport(),
|
|
]
|
|
|
|
client = ClaudeSDKClient()
|
|
await client.connect()
|
|
# Second connect should create new transport
|
|
await client.connect()
|
|
|
|
# Should have been called twice
|
|
assert mock_transport_class.call_count == 2
|
|
|
|
anyio.run(_test)
|
|
|
|
def test_disconnect_without_connect(self):
|
|
"""Test disconnecting without connecting first."""
|
|
|
|
async def _test():
|
|
client = ClaudeSDKClient()
|
|
# Should not raise error
|
|
await client.disconnect()
|
|
|
|
anyio.run(_test)
|
|
|
|
def test_context_manager_with_exception(self):
|
|
"""Test context manager cleans up on exception."""
|
|
|
|
async def _test():
|
|
with patch(
|
|
"claude_agent_sdk._internal.transport.subprocess_cli.SubprocessCLITransport"
|
|
) as mock_transport_class:
|
|
mock_transport = create_mock_transport()
|
|
mock_transport_class.return_value = mock_transport
|
|
|
|
with pytest.raises(ValueError):
|
|
async with ClaudeSDKClient():
|
|
raise ValueError("Test error")
|
|
|
|
# Disconnect should still be called
|
|
mock_transport.close.assert_called_once()
|
|
|
|
anyio.run(_test)
|
|
|
|
def test_receive_response_list_comprehension(self):
|
|
"""Test collecting messages with list comprehension as shown in examples."""
|
|
|
|
async def _test():
|
|
with patch(
|
|
"claude_agent_sdk._internal.transport.subprocess_cli.SubprocessCLITransport"
|
|
) as mock_transport_class:
|
|
mock_transport = create_mock_transport()
|
|
mock_transport_class.return_value = mock_transport
|
|
|
|
# Mock the message stream with control protocol support
|
|
async def mock_receive():
|
|
# First handle initialization
|
|
await asyncio.sleep(0.01)
|
|
written = mock_transport.write.call_args_list
|
|
for call in written:
|
|
if call:
|
|
data = call[0][0]
|
|
try:
|
|
msg = json.loads(data.strip())
|
|
if (
|
|
msg.get("type") == "control_request"
|
|
and msg.get("request", {}).get("subtype")
|
|
== "initialize"
|
|
):
|
|
yield {
|
|
"type": "control_response",
|
|
"response": {
|
|
"request_id": msg.get("request_id"),
|
|
"subtype": "success",
|
|
"commands": [],
|
|
"output_style": "default",
|
|
},
|
|
}
|
|
break
|
|
except (json.JSONDecodeError, KeyError, AttributeError):
|
|
pass
|
|
|
|
# Then yield the actual messages
|
|
yield {
|
|
"type": "assistant",
|
|
"message": {
|
|
"role": "assistant",
|
|
"content": [{"type": "text", "text": "Hello"}],
|
|
"model": "claude-opus-4-1-20250805",
|
|
},
|
|
}
|
|
yield {
|
|
"type": "assistant",
|
|
"message": {
|
|
"role": "assistant",
|
|
"content": [{"type": "text", "text": "World"}],
|
|
"model": "claude-opus-4-1-20250805",
|
|
},
|
|
}
|
|
yield {
|
|
"type": "result",
|
|
"subtype": "success",
|
|
"duration_ms": 1000,
|
|
"duration_api_ms": 800,
|
|
"is_error": False,
|
|
"num_turns": 1,
|
|
"session_id": "test",
|
|
"total_cost_usd": 0.001,
|
|
}
|
|
|
|
mock_transport.read_messages = mock_receive
|
|
|
|
async with ClaudeSDKClient() as client:
|
|
# Test list comprehension pattern from docstring
|
|
messages = [msg async for msg in client.receive_response()]
|
|
|
|
assert len(messages) == 3
|
|
assert all(
|
|
isinstance(msg, AssistantMessage | ResultMessage)
|
|
for msg in messages
|
|
)
|
|
assert isinstance(messages[-1], ResultMessage)
|
|
|
|
anyio.run(_test)
|
|
|
|
def test_flush_stdin_called_after_control_requests(self):
|
|
"""Test that flush_stdin is called after sending control requests (issue #208)."""
|
|
|
|
async def _test():
|
|
with patch(
|
|
"claude_agent_sdk._internal.transport.subprocess_cli.SubprocessCLITransport"
|
|
) as mock_transport_class:
|
|
mock_transport = create_mock_transport()
|
|
|
|
# Add flush_stdin mock and tracking
|
|
flush_calls = []
|
|
|
|
async def mock_flush():
|
|
flush_calls.append(True)
|
|
|
|
mock_transport.flush_stdin = AsyncMock(side_effect=mock_flush)
|
|
|
|
mock_transport_class.return_value = mock_transport
|
|
|
|
async with ClaudeSDKClient() as client:
|
|
# Initialization should call flush_stdin
|
|
# Wait a bit for initialization to complete
|
|
await asyncio.sleep(0.05)
|
|
|
|
# Verify flush_stdin was called at least once (for initialization)
|
|
assert len(flush_calls) >= 1, (
|
|
"flush_stdin should be called during initialization"
|
|
)
|
|
initial_flush_count = len(flush_calls)
|
|
|
|
# Send interrupt control request
|
|
await client.interrupt()
|
|
await asyncio.sleep(0.05)
|
|
|
|
# Verify flush_stdin was called again (for interrupt request)
|
|
assert len(flush_calls) > initial_flush_count, (
|
|
"flush_stdin should be called after interrupt"
|
|
)
|
|
|
|
anyio.run(_test)
|