"""Comprehensive tests for ClaudeRunner subprocess wrapper. Tests cover: - New session creation - Session resumption with context preservation - Timeout handling - JSON parsing (including edge cases) - Error handling (process failures, malformed output, permission denials) """ import asyncio import json import pytest from unittest.mock import AsyncMock, MagicMock, patch from claude_coordinator.claude_runner import ClaudeRunner, ClaudeResponse class TestClaudeRunner: """Test suite for ClaudeRunner class.""" @pytest.fixture def runner(self): """Create a ClaudeRunner instance for testing.""" return ClaudeRunner(default_timeout=30) @pytest.fixture def mock_process(self): """Create a mock subprocess for testing.""" process = AsyncMock() process.returncode = 0 return process def create_mock_response( self, result="Test response", session_id="test-session-123", is_error=False, cost=0.01 ): """Helper to create mock JSON response.""" return json.dumps({ "type": "result", "subtype": "success" if not is_error else "error", "is_error": is_error, "result": result, "session_id": session_id, "total_cost_usd": cost, "duration_ms": 2000, "permission_denials": [] }).encode('utf-8') @pytest.mark.asyncio async def test_new_session_creation(self, runner, mock_process): """Test creating a new Claude session without session_id. Verifies that: - Command is built correctly without --resume flag - JSON response is parsed successfully - Session ID is extracted from response """ stdout = self.create_mock_response( result="Hello! How can I help?", session_id="new-session-uuid-123" ) mock_process.communicate.return_value = (stdout, b"") with patch('asyncio.create_subprocess_exec', return_value=mock_process): response = await runner.run( message="Hello Claude", session_id=None ) assert response.success is True assert response.result == "Hello! How can I help?" assert response.session_id == "new-session-uuid-123" assert response.error is None assert response.cost == 0.01 @pytest.mark.asyncio async def test_session_resumption(self, runner, mock_process): """Test resuming an existing session with session_id. Verifies that: - --resume flag is included in command - Session ID is passed correctly - Response maintains session context """ existing_session_id = "existing-session-456" stdout = self.create_mock_response( result="You asked about Python before.", session_id=existing_session_id ) mock_process.communicate.return_value = (stdout, b"") with patch('asyncio.create_subprocess_exec', return_value=mock_process) as mock_exec: response = await runner.run( message="What was I asking about?", session_id=existing_session_id ) # Verify --resume flag was added call_args = mock_exec.call_args cmd = call_args[0] assert "--resume" in cmd assert existing_session_id in cmd assert response.success is True assert response.session_id == existing_session_id @pytest.mark.asyncio async def test_timeout_handling(self, runner, mock_process): """Test subprocess timeout with asyncio.TimeoutError. Verifies that: - Process is killed on timeout - Error response is returned - Timeout duration is respected """ async def slow_communicate(): """Simulate a slow response that times out.""" await asyncio.sleep(100) # Longer than timeout return (b"", b"") mock_process.communicate = slow_communicate with patch('asyncio.create_subprocess_exec', return_value=mock_process): response = await runner.run( message="Slow command", timeout=1 # 1 second timeout ) assert response.success is False assert "timed out" in response.error.lower() assert mock_process.kill.called @pytest.mark.asyncio async def test_malformed_json_handling(self, runner, mock_process): """Test handling of malformed JSON output. Verifies that: - JSON parse errors are caught - Error response is returned with details - Raw output is logged for debugging """ stdout = b"This is not valid JSON {{{" mock_process.communicate.return_value = (stdout, b"") with patch('asyncio.create_subprocess_exec', return_value=mock_process): response = await runner.run(message="Test") assert response.success is False assert "Malformed JSON" in response.error @pytest.mark.asyncio async def test_process_error_handling(self, runner, mock_process): """Test handling of non-zero exit codes. Verifies that: - Non-zero return codes are detected - stderr is captured and returned - Error response is generated """ mock_process.returncode = 1 stderr = b"Claude CLI error: Invalid token" mock_process.communicate.return_value = (b"", stderr) with patch('asyncio.create_subprocess_exec', return_value=mock_process): response = await runner.run(message="Test") assert response.success is False assert "exited with code 1" in response.error assert "Invalid token" in response.error @pytest.mark.asyncio async def test_claude_error_response(self, runner, mock_process): """Test handling of Claude API errors in JSON response. Verifies that: - is_error flag is detected - Error message is extracted from result field - Error response is returned """ stdout = self.create_mock_response( result="API rate limit exceeded", is_error=True ) mock_process.communicate.return_value = (stdout, b"") with patch('asyncio.create_subprocess_exec', return_value=mock_process): response = await runner.run(message="Test") assert response.success is False assert "API rate limit exceeded" in response.error @pytest.mark.asyncio async def test_permission_denial_handling(self, runner, mock_process): """Test detection of permission denials. Verifies that: - permission_denials array is checked - Error response is returned if permissions denied - Denial details are included in error """ response_data = json.dumps({ "type": "result", "is_error": False, "result": "Cannot execute", "session_id": "test-123", "permission_denials": [{"tool": "Write", "reason": "Not allowed"}] }).encode('utf-8') mock_process.communicate.return_value = (response_data, b"") with patch('asyncio.create_subprocess_exec', return_value=mock_process): response = await runner.run(message="Test") assert response.success is False assert "Permission denied" in response.error assert response.permission_denials is not None @pytest.mark.asyncio async def test_command_building_with_all_options(self, runner, mock_process): """Test command building with all optional parameters. Verifies that: - All flags are included in command - Values are passed correctly - Command structure is valid """ stdout = self.create_mock_response() mock_process.communicate.return_value = (stdout, b"") with patch('asyncio.create_subprocess_exec', return_value=mock_process) as mock_exec: await runner.run( message="Test message", session_id="test-session", allowed_tools=["Bash", "Read", "Write"], system_prompt="Custom system prompt", model="sonnet" ) # Extract command from call cmd = mock_exec.call_args[0] # Verify all flags present assert "claude" in cmd assert "-p" in cmd assert "Test message" in cmd assert "--output-format" in cmd assert "json" in cmd assert "--permission-mode" in cmd assert "bypassPermissions" in cmd assert "--resume" in cmd assert "test-session" in cmd assert "--allowed-tools" in cmd assert "Bash,Read,Write" in cmd assert "--system-prompt" in cmd assert "Custom system prompt" in cmd assert "--model" in cmd assert "sonnet" in cmd @pytest.mark.asyncio async def test_environment_preparation(self, runner, mock_process): """Test environment variable preparation. Verifies that: - CLAUDECODE is unset to allow nested sessions - CLAUDE_CODE_OAUTH_TOKEN is set if provided - Environment is passed to subprocess """ stdout = self.create_mock_response() mock_process.communicate.return_value = (stdout, b"") runner_with_token = ClaudeRunner(oauth_token="test-oauth-token") with patch('asyncio.create_subprocess_exec', return_value=mock_process) as mock_exec: with patch.dict('os.environ', {'CLAUDECODE': 'some-value'}): await runner_with_token.run(message="Test") # Check environment passed to subprocess env = mock_exec.call_args[1]['env'] # CLAUDECODE should be removed assert 'CLAUDECODE' not in env # OAuth token should be set assert env.get('CLAUDE_CODE_OAUTH_TOKEN') == 'test-oauth-token' @pytest.mark.asyncio async def test_cwd_parameter(self, runner, mock_process): """Test working directory parameter. Verifies that: - cwd is passed to subprocess - Claude runs in the correct directory """ stdout = self.create_mock_response() mock_process.communicate.return_value = (stdout, b"") test_cwd = "/opt/projects/test-project" with patch('asyncio.create_subprocess_exec', return_value=mock_process) as mock_exec: await runner.run( message="Test", cwd=test_cwd ) # Verify cwd was passed assert mock_exec.call_args[1]['cwd'] == test_cwd def test_parse_response_edge_cases(self, runner): """Test JSON parsing with various edge cases. Verifies handling of: - Empty result field - Missing optional fields - Unusual but valid JSON structures """ # Test with minimal valid response minimal_json = json.dumps({ "is_error": False, "result": "" }) response = runner._parse_response(minimal_json) assert response.success is True assert response.result == "" assert response.session_id is None # Test with all optional fields present complete_json = json.dumps({ "type": "result", "subtype": "success", "is_error": False, "result": "Complete response", "session_id": "uuid-123", "total_cost_usd": 0.05, "duration_ms": 5000, "permission_denials": [] }) response = runner._parse_response(complete_json) assert response.success is True assert response.result == "Complete response" assert response.session_id == "uuid-123" assert response.cost == 0.05 assert response.duration_ms == 5000 # Integration test (requires actual Claude CLI installed) @pytest.mark.integration @pytest.mark.asyncio async def test_real_claude_session(): """Integration test with real Claude CLI (requires authentication). This test is marked as integration and will be skipped unless explicitly run with: pytest -m integration Verifies: - Real session creation works - Session resumption preserves context - JSON parsing works with real output """ runner = ClaudeRunner(default_timeout=60) # Create new session response1 = await runner.run( message="Please respond with exactly: 'Integration test response'" ) assert response1.success is True assert "Integration test response" in response1.result assert response1.session_id is not None # Resume session session_id = response1.session_id response2 = await runner.run( message="What did I just ask you to say?", session_id=session_id ) assert response2.success is True assert response2.session_id == session_id assert "Integration test response" in response2.result.lower() if __name__ == "__main__": # Run tests with: python -m pytest tests/test_claude_runner.py -v pytest.main(["-v", __file__])