tests/test_executor.py

import pytest
from sandbox.executor import SandboxExecutor
from sandbox.models import ExecutionRequest, SandboxConfig, Language


# Skip all tests if Docker is not available
docker = pytest.importorskip("docker")


@pytest.fixture
def executor():
    """Sandbox executor fixture"""
    try:
        return SandboxExecutor(SandboxConfig())
    except RuntimeError:
        pytest.skip("Docker is not available")


def test_executor_initialization(executor):
    """Test executor can be initialized"""
    assert executor is not None
    assert executor.client is not None


def test_execute_python_simple(executor):
    """Test simple Python execution"""
    request = ExecutionRequest(
        code="print('Test')",
        language=Language.PYTHON
    )
    
    response = executor.execute(request)
    
    assert "Test" in response.stdout
    assert response.exit_code == 0
    assert response.error is None
    assert response.execution_time > 0


def test_execute_python_with_variables(executor):
    """Test Python with variable assignment"""
    request = ExecutionRequest(
        code="x = 10\ny = 20\nprint(x + y)",
        language=Language.PYTHON
    )
    
    response = executor.execute(request)
    
    assert "30" in response.stdout
    assert response.exit_code == 0


def test_execute_javascript_simple(executor):
    """Test simple JavaScript execution"""
    request = ExecutionRequest(
        code="console.log('JavaScript works!');",
        language=Language.JAVASCRIPT
    )
    
    response = executor.execute(request)
    
    assert "JavaScript works!" in response.stdout
    assert response.exit_code == 0


def test_execute_bash_simple(executor):
    """Test simple Bash execution"""
    request = ExecutionRequest(
        code="echo 'Bash test'",
        language=Language.BASH
    )
    
    response = executor.execute(request)
    
    assert "Bash test" in response.stdout
    assert response.exit_code == 0


def test_timeout_enforcement(executor):
    """Test that timeout is enforced"""
    request = ExecutionRequest(
        code="import time\nwhile True:\n    time.sleep(1)",
        language=Language.PYTHON,
        timeout=2
    )
    
    response = executor.execute(request)
    
    # Should timeout
    assert response.execution_time < 3
    assert response.exit_code == 124 or response.error is not None


def test_syntax_error_handling(executor):
    """Test handling of code with syntax errors"""
    request = ExecutionRequest(
        code="print('missing quote)",
        language=Language.PYTHON
    )
    
    response = executor.execute(request)
    
    assert response.exit_code != 0
    # Error should be captured in stderr or stdout
    assert len(response.stdout + response.stderr) > 0


def test_runtime_error_handling(executor):
    """Test handling of runtime errors"""
    request = ExecutionRequest(
        code="x = 1 / 0",  # Division by zero
        language=Language.PYTHON
    )
    
    response = executor.execute(request)
    
    assert response.exit_code != 0


def test_container_cleanup(executor):
    """Test that containers are cleaned up after execution"""
    import docker
    client = docker.from_env()
    
    # Get initial container count
    initial_containers = len(client.containers.list(all=True))
    
    # Execute code
    request = ExecutionRequest(
        code="print('cleanup test')",
        language=Language.PYTHON
    )
    executor.execute(request)
    
    # Check container count after execution
    final_containers = len(client.containers.list(all=True))
    
    # Should be the same (container was cleaned up)
    assert final_containers == initial_containers


def test_memory_limit_config(executor):
    """Test that memory limit is applied"""
    request = ExecutionRequest(
        code="print('memory test')",
        language=Language.PYTHON,
        memory_limit=128
    )
    
    response = executor.execute(request)
    
    # Should execute successfully with lower memory
    assert response.exit_code == 0


def test_output_truncation(executor):
    """Test that large output is truncated"""
    # Generate large output
    code = "for i in range(100000):\n    print('x' * 100)"
    
    request = ExecutionRequest(
        code=code,
        language=Language.PYTHON
    )
    
    response = executor.execute(request)
    
    # Output should be truncated
    assert "truncated" in response.stdout or len(response.stdout) <= executor.config.max_output_size + 100


def test_multiple_executions(executor):
    """Test multiple consecutive executions"""
    for i in range(5):
        request = ExecutionRequest(
            code=f"print('Execution {i}')",
            language=Language.PYTHON
        )
        
        response = executor.execute(request)
        
        assert f"Execution {i}" in response.stdout
        assert response.exit_code == 0