mirror of
https://github.com/anthropics/claude-agent-sdk-python.git
synced 2025-10-06 01:00:03 +03:00
d86c47f2d6
Simplify CLI detection by removing redundant node installation check before throwing CLINotFoundError. 🤖 Generated with [Claude Code](https://claude.ai/code) --------- Co-authored-by: Claude <noreply@anthropic.com>
439 lines
15 KiB
Python
439 lines
15 KiB
Python
"""Tests for Claude SDK transport layer."""
|
|
|
|
import os
|
|
import uuid
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import anyio
|
|
import pytest
|
|
|
|
from claude_agent_sdk._internal.transport.subprocess_cli import SubprocessCLITransport
|
|
from claude_agent_sdk.types import ClaudeAgentOptions
|
|
|
|
|
|
class TestSubprocessCLITransport:
|
|
"""Test subprocess transport implementation."""
|
|
|
|
def test_find_cli_not_found(self):
|
|
"""Test CLI not found error."""
|
|
from claude_agent_sdk._errors import CLINotFoundError
|
|
|
|
with (
|
|
patch("shutil.which", return_value=None),
|
|
patch("pathlib.Path.exists", return_value=False),
|
|
pytest.raises(CLINotFoundError) as exc_info,
|
|
):
|
|
SubprocessCLITransport(prompt="test", options=ClaudeAgentOptions())
|
|
|
|
assert "Claude Code not found" in str(exc_info.value)
|
|
|
|
def test_build_command_basic(self):
|
|
"""Test building basic CLI command."""
|
|
transport = SubprocessCLITransport(
|
|
prompt="Hello", options=ClaudeAgentOptions(), cli_path="/usr/bin/claude"
|
|
)
|
|
|
|
cmd = transport._build_command()
|
|
assert cmd[0] == "/usr/bin/claude"
|
|
assert "--output-format" in cmd
|
|
assert "stream-json" in cmd
|
|
assert "--print" in cmd
|
|
assert "Hello" in cmd
|
|
|
|
def test_cli_path_accepts_pathlib_path(self):
|
|
"""Test that cli_path accepts pathlib.Path objects."""
|
|
from pathlib import Path
|
|
|
|
transport = SubprocessCLITransport(
|
|
prompt="Hello",
|
|
options=ClaudeAgentOptions(),
|
|
cli_path=Path("/usr/bin/claude"),
|
|
)
|
|
|
|
assert transport._cli_path == "/usr/bin/claude"
|
|
|
|
def test_build_command_with_system_prompt_string(self):
|
|
"""Test building CLI command with system prompt as string."""
|
|
transport = SubprocessCLITransport(
|
|
prompt="test",
|
|
options=ClaudeAgentOptions(
|
|
system_prompt="Be helpful",
|
|
),
|
|
cli_path="/usr/bin/claude",
|
|
)
|
|
|
|
cmd = transport._build_command()
|
|
assert "--system-prompt" in cmd
|
|
assert "Be helpful" in cmd
|
|
|
|
def test_build_command_with_system_prompt_preset(self):
|
|
"""Test building CLI command with system prompt preset."""
|
|
transport = SubprocessCLITransport(
|
|
prompt="test",
|
|
options=ClaudeAgentOptions(
|
|
system_prompt={"type": "preset", "preset": "claude_code"},
|
|
),
|
|
cli_path="/usr/bin/claude",
|
|
)
|
|
|
|
cmd = transport._build_command()
|
|
assert "--system-prompt" not in cmd
|
|
assert "--append-system-prompt" not in cmd
|
|
|
|
def test_build_command_with_system_prompt_preset_and_append(self):
|
|
"""Test building CLI command with system prompt preset and append."""
|
|
transport = SubprocessCLITransport(
|
|
prompt="test",
|
|
options=ClaudeAgentOptions(
|
|
system_prompt={
|
|
"type": "preset",
|
|
"preset": "claude_code",
|
|
"append": "Be concise.",
|
|
},
|
|
),
|
|
cli_path="/usr/bin/claude",
|
|
)
|
|
|
|
cmd = transport._build_command()
|
|
assert "--system-prompt" not in cmd
|
|
assert "--append-system-prompt" in cmd
|
|
assert "Be concise." in cmd
|
|
|
|
def test_build_command_with_options(self):
|
|
"""Test building CLI command with options."""
|
|
transport = SubprocessCLITransport(
|
|
prompt="test",
|
|
options=ClaudeAgentOptions(
|
|
allowed_tools=["Read", "Write"],
|
|
disallowed_tools=["Bash"],
|
|
model="claude-3-5-sonnet",
|
|
permission_mode="acceptEdits",
|
|
max_turns=5,
|
|
),
|
|
cli_path="/usr/bin/claude",
|
|
)
|
|
|
|
cmd = transport._build_command()
|
|
assert "--allowedTools" in cmd
|
|
assert "Read,Write" in cmd
|
|
assert "--disallowedTools" in cmd
|
|
assert "Bash" in cmd
|
|
assert "--model" in cmd
|
|
assert "claude-3-5-sonnet" in cmd
|
|
assert "--permission-mode" in cmd
|
|
assert "acceptEdits" in cmd
|
|
assert "--max-turns" in cmd
|
|
assert "5" in cmd
|
|
|
|
def test_build_command_with_add_dirs(self):
|
|
"""Test building CLI command with add_dirs option."""
|
|
from pathlib import Path
|
|
|
|
transport = SubprocessCLITransport(
|
|
prompt="test",
|
|
options=ClaudeAgentOptions(
|
|
add_dirs=["/path/to/dir1", Path("/path/to/dir2")]
|
|
),
|
|
cli_path="/usr/bin/claude",
|
|
)
|
|
|
|
cmd = transport._build_command()
|
|
cmd_str = " ".join(cmd)
|
|
|
|
# Check that the command string contains the expected --add-dir flags
|
|
assert "--add-dir /path/to/dir1 --add-dir /path/to/dir2" in cmd_str
|
|
|
|
def test_session_continuation(self):
|
|
"""Test session continuation options."""
|
|
transport = SubprocessCLITransport(
|
|
prompt="Continue from before",
|
|
options=ClaudeAgentOptions(
|
|
continue_conversation=True, resume="session-123"
|
|
),
|
|
cli_path="/usr/bin/claude",
|
|
)
|
|
|
|
cmd = transport._build_command()
|
|
assert "--continue" in cmd
|
|
assert "--resume" in cmd
|
|
assert "session-123" in cmd
|
|
|
|
def test_connect_close(self):
|
|
"""Test connect and close lifecycle."""
|
|
|
|
async def _test():
|
|
with patch("anyio.open_process") as mock_exec:
|
|
mock_process = MagicMock()
|
|
mock_process.returncode = None
|
|
mock_process.terminate = MagicMock()
|
|
mock_process.wait = AsyncMock()
|
|
mock_process.stdout = MagicMock()
|
|
mock_process.stderr = MagicMock()
|
|
|
|
# Mock stdin with aclose method
|
|
mock_stdin = MagicMock()
|
|
mock_stdin.aclose = AsyncMock()
|
|
mock_process.stdin = mock_stdin
|
|
|
|
mock_exec.return_value = mock_process
|
|
|
|
transport = SubprocessCLITransport(
|
|
prompt="test",
|
|
options=ClaudeAgentOptions(),
|
|
cli_path="/usr/bin/claude",
|
|
)
|
|
|
|
await transport.connect()
|
|
assert transport._process is not None
|
|
assert transport.is_ready()
|
|
|
|
await transport.close()
|
|
mock_process.terminate.assert_called_once()
|
|
|
|
anyio.run(_test)
|
|
|
|
def test_read_messages(self):
|
|
"""Test reading messages from CLI output."""
|
|
# This test is simplified to just test the transport creation
|
|
# The full async stream handling is tested in integration tests
|
|
transport = SubprocessCLITransport(
|
|
prompt="test", options=ClaudeAgentOptions(), cli_path="/usr/bin/claude"
|
|
)
|
|
|
|
# The transport now just provides raw message reading via read_messages()
|
|
# So we just verify the transport can be created and basic structure is correct
|
|
assert transport._prompt == "test"
|
|
assert transport._cli_path == "/usr/bin/claude"
|
|
|
|
def test_connect_with_nonexistent_cwd(self):
|
|
"""Test that connect raises CLIConnectionError when cwd doesn't exist."""
|
|
from claude_agent_sdk._errors import CLIConnectionError
|
|
|
|
async def _test():
|
|
transport = SubprocessCLITransport(
|
|
prompt="test",
|
|
options=ClaudeAgentOptions(cwd="/this/directory/does/not/exist"),
|
|
cli_path="/usr/bin/claude",
|
|
)
|
|
|
|
with pytest.raises(CLIConnectionError) as exc_info:
|
|
await transport.connect()
|
|
|
|
assert "/this/directory/does/not/exist" in str(exc_info.value)
|
|
|
|
anyio.run(_test)
|
|
|
|
def test_build_command_with_settings_file(self):
|
|
"""Test building CLI command with settings as file path."""
|
|
transport = SubprocessCLITransport(
|
|
prompt="test",
|
|
options=ClaudeAgentOptions(settings="/path/to/settings.json"),
|
|
cli_path="/usr/bin/claude",
|
|
)
|
|
|
|
cmd = transport._build_command()
|
|
assert "--settings" in cmd
|
|
assert "/path/to/settings.json" in cmd
|
|
|
|
def test_build_command_with_settings_json(self):
|
|
"""Test building CLI command with settings as JSON object."""
|
|
settings_json = '{"permissions": {"allow": ["Bash(ls:*)"]}}'
|
|
transport = SubprocessCLITransport(
|
|
prompt="test",
|
|
options=ClaudeAgentOptions(settings=settings_json),
|
|
cli_path="/usr/bin/claude",
|
|
)
|
|
|
|
cmd = transport._build_command()
|
|
assert "--settings" in cmd
|
|
assert settings_json in cmd
|
|
|
|
def test_build_command_with_extra_args(self):
|
|
"""Test building CLI command with extra_args for future flags."""
|
|
transport = SubprocessCLITransport(
|
|
prompt="test",
|
|
options=ClaudeAgentOptions(
|
|
extra_args={
|
|
"new-flag": "value",
|
|
"boolean-flag": None,
|
|
"another-option": "test-value",
|
|
}
|
|
),
|
|
cli_path="/usr/bin/claude",
|
|
)
|
|
|
|
cmd = transport._build_command()
|
|
cmd_str = " ".join(cmd)
|
|
|
|
# Check flags with values
|
|
assert "--new-flag value" in cmd_str
|
|
assert "--another-option test-value" in cmd_str
|
|
|
|
# Check boolean flag (no value)
|
|
assert "--boolean-flag" in cmd
|
|
# Make sure boolean flag doesn't have a value after it
|
|
boolean_idx = cmd.index("--boolean-flag")
|
|
# Either it's the last element or the next element is another flag
|
|
assert boolean_idx == len(cmd) - 1 or cmd[boolean_idx + 1].startswith("--")
|
|
|
|
def test_build_command_with_mcp_servers(self):
|
|
"""Test building CLI command with mcp_servers option."""
|
|
import json
|
|
|
|
mcp_servers = {
|
|
"test-server": {
|
|
"type": "stdio",
|
|
"command": "/path/to/server",
|
|
"args": ["--option", "value"],
|
|
}
|
|
}
|
|
|
|
transport = SubprocessCLITransport(
|
|
prompt="test",
|
|
options=ClaudeAgentOptions(mcp_servers=mcp_servers),
|
|
cli_path="/usr/bin/claude",
|
|
)
|
|
|
|
cmd = transport._build_command()
|
|
|
|
# Find the --mcp-config flag and its value
|
|
assert "--mcp-config" in cmd
|
|
mcp_idx = cmd.index("--mcp-config")
|
|
mcp_config_value = cmd[mcp_idx + 1]
|
|
|
|
# Parse the JSON and verify structure
|
|
config = json.loads(mcp_config_value)
|
|
assert "mcpServers" in config
|
|
assert config["mcpServers"] == mcp_servers
|
|
|
|
def test_build_command_with_mcp_servers_as_file_path(self):
|
|
"""Test building CLI command with mcp_servers as file path."""
|
|
from pathlib import Path
|
|
|
|
# Test with string path
|
|
transport = SubprocessCLITransport(
|
|
prompt="test",
|
|
options=ClaudeAgentOptions(mcp_servers="/path/to/mcp-config.json"),
|
|
cli_path="/usr/bin/claude",
|
|
)
|
|
|
|
cmd = transport._build_command()
|
|
assert "--mcp-config" in cmd
|
|
mcp_idx = cmd.index("--mcp-config")
|
|
assert cmd[mcp_idx + 1] == "/path/to/mcp-config.json"
|
|
|
|
# Test with Path object
|
|
transport = SubprocessCLITransport(
|
|
prompt="test",
|
|
options=ClaudeAgentOptions(mcp_servers=Path("/path/to/mcp-config.json")),
|
|
cli_path="/usr/bin/claude",
|
|
)
|
|
|
|
cmd = transport._build_command()
|
|
assert "--mcp-config" in cmd
|
|
mcp_idx = cmd.index("--mcp-config")
|
|
assert cmd[mcp_idx + 1] == "/path/to/mcp-config.json"
|
|
|
|
def test_build_command_with_mcp_servers_as_json_string(self):
|
|
"""Test building CLI command with mcp_servers as JSON string."""
|
|
json_config = '{"mcpServers": {"server": {"type": "stdio", "command": "test"}}}'
|
|
transport = SubprocessCLITransport(
|
|
prompt="test",
|
|
options=ClaudeAgentOptions(mcp_servers=json_config),
|
|
cli_path="/usr/bin/claude",
|
|
)
|
|
|
|
cmd = transport._build_command()
|
|
assert "--mcp-config" in cmd
|
|
mcp_idx = cmd.index("--mcp-config")
|
|
assert cmd[mcp_idx + 1] == json_config
|
|
|
|
def test_env_vars_passed_to_subprocess(self):
|
|
"""Test that custom environment variables are passed to the subprocess."""
|
|
|
|
async def _test():
|
|
test_value = f"test-{uuid.uuid4().hex[:8]}"
|
|
custom_env = {
|
|
"MY_TEST_VAR": test_value,
|
|
}
|
|
|
|
options = ClaudeAgentOptions(env=custom_env)
|
|
|
|
# Mock the subprocess to capture the env argument
|
|
with patch(
|
|
"anyio.open_process", new_callable=AsyncMock
|
|
) as mock_open_process:
|
|
mock_process = MagicMock()
|
|
mock_process.stdout = MagicMock()
|
|
mock_stdin = MagicMock()
|
|
mock_stdin.aclose = AsyncMock() # Add async aclose method
|
|
mock_process.stdin = mock_stdin
|
|
mock_process.returncode = None
|
|
mock_open_process.return_value = mock_process
|
|
|
|
transport = SubprocessCLITransport(
|
|
prompt="test",
|
|
options=options,
|
|
cli_path="/usr/bin/claude",
|
|
)
|
|
|
|
await transport.connect()
|
|
|
|
# Verify open_process was called with correct env vars
|
|
mock_open_process.assert_called_once()
|
|
call_kwargs = mock_open_process.call_args.kwargs
|
|
assert "env" in call_kwargs
|
|
env_passed = call_kwargs["env"]
|
|
|
|
# Check that custom env var was passed
|
|
assert env_passed["MY_TEST_VAR"] == test_value
|
|
|
|
# Verify SDK identifier is present
|
|
assert "CLAUDE_CODE_ENTRYPOINT" in env_passed
|
|
assert env_passed["CLAUDE_CODE_ENTRYPOINT"] == "sdk-py"
|
|
|
|
# Verify system env vars are also included with correct values
|
|
if "PATH" in os.environ:
|
|
assert "PATH" in env_passed
|
|
assert env_passed["PATH"] == os.environ["PATH"]
|
|
|
|
anyio.run(_test)
|
|
|
|
def test_connect_as_different_user(self):
|
|
"""Test connect as different user."""
|
|
|
|
async def _test():
|
|
custom_user = "claude"
|
|
options = ClaudeAgentOptions(user=custom_user)
|
|
|
|
# Mock the subprocess to capture the env argument
|
|
with patch(
|
|
"anyio.open_process", new_callable=AsyncMock
|
|
) as mock_open_process:
|
|
mock_process = MagicMock()
|
|
mock_process.stdout = MagicMock()
|
|
mock_stdin = MagicMock()
|
|
mock_stdin.aclose = AsyncMock() # Add async aclose method
|
|
mock_process.stdin = mock_stdin
|
|
mock_process.returncode = None
|
|
mock_open_process.return_value = mock_process
|
|
|
|
transport = SubprocessCLITransport(
|
|
prompt="test",
|
|
options=options,
|
|
cli_path="/usr/bin/claude",
|
|
)
|
|
|
|
await transport.connect()
|
|
|
|
# Verify open_process was called with correct user
|
|
mock_open_process.assert_called_once()
|
|
call_kwargs = mock_open_process.call_args.kwargs
|
|
assert "user" in call_kwargs
|
|
user_passed = call_kwargs["user"]
|
|
|
|
# Check that user was passed
|
|
assert user_passed == "claude"
|
|
|
|
anyio.run(_test)
|