"""Tests for Claude SDK transport layer.""" import os import uuid from unittest.mock import AsyncMock, MagicMock, patch import anyio import pytest from claude_code_sdk._internal.transport.subprocess_cli import SubprocessCLITransport from claude_code_sdk.types import ClaudeCodeOptions class TestSubprocessCLITransport: """Test subprocess transport implementation.""" def test_find_cli_not_found(self): """Test CLI not found error.""" from claude_code_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=ClaudeCodeOptions()) assert "Claude Code requires Node.js" in str(exc_info.value) def test_build_command_basic(self): """Test building basic CLI command.""" transport = SubprocessCLITransport( prompt="Hello", options=ClaudeCodeOptions(), 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=ClaudeCodeOptions(), cli_path=Path("/usr/bin/claude"), ) assert transport._cli_path == "/usr/bin/claude" def test_build_command_with_options(self): """Test building CLI command with options.""" transport = SubprocessCLITransport( prompt="test", options=ClaudeCodeOptions( system_prompt="Be helpful", 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 "--system-prompt" in cmd assert "Be helpful" in cmd 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=ClaudeCodeOptions( 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=ClaudeCodeOptions(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=ClaudeCodeOptions(), 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=ClaudeCodeOptions(), 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_code_sdk._errors import CLIConnectionError async def _test(): transport = SubprocessCLITransport( prompt="test", options=ClaudeCodeOptions(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=ClaudeCodeOptions(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=ClaudeCodeOptions(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=ClaudeCodeOptions( 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=ClaudeCodeOptions(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=ClaudeCodeOptions(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=ClaudeCodeOptions(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=ClaudeCodeOptions(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 = ClaudeCodeOptions(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)