ai-assistant-discord-bot/tests/test_claude_runner.py

"""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__])