agent/test_execute_direct.py

#!/usr/bin/env python3
"""Direct test of the execute endpoint without full backend."""

import asyncio
import subprocess


async def execute_code_direct(command: str, timeout: float = 3600):
    """Execute a command directly using subprocess."""
    try:
        process = await asyncio.create_task(
            asyncio.to_thread(
                subprocess.run,
                command,
                shell=True,
                capture_output=True,
                text=True,
                timeout=timeout,
            )
        )
        
        output = process.stdout
        if process.stderr:
            output += f"\n[STDERR]\n{process.stderr}"
        
        success = process.returncode == 0
        
        return {"success": success, "output": output, "return_code": process.returncode}
        
    except subprocess.TimeoutExpired:
        return {"success": False, "error": f"Command exceeded timeout of {timeout} seconds", "return_code": -1}
    except Exception as e:
        return {"success": False, "error": str(e), "return_code": -1}


async def test_execute_code():
    """Test code execution."""
    print("=" * 60)
    print("Testing Direct Code Execution")
    print("=" * 60)
    
    # Test 1: Simple echo
    print("\nTest 1: Simple echo command")
    result = await execute_code_direct("echo 'Hello from execute_code!'")
    print(f"Success: {result['success']}")
    print(f"Output: {result.get('output', result.get('error'))}")
    
    # Test 2: Write a file
    print("\nTest 2: Write a Python file")
    result = await execute_code_direct("""cat > /tmp/test_script.py << 'EOF'
from sympy import symbols, diff, integrate

x = symbols('x')
f = x**3 + 2*x**2 + 3*x + 4

print("Function:", f)
print("Derivative:", diff(f, x))
print("Integral:", integrate(f, x))
EOF""")
    print(f"Success: {result['success']}")
    
    # Test 3: Install and run
    print("\nTest 3: Install sympy and run script")
    result = await execute_code_direct("pip install --quiet sympy && python3 /tmp/test_script.py", timeout=120)
    print(f"Success: {result['success']}")
    print(f"Output: {result.get('output', result.get('error'))}")
    
    # Test 4: List files
    print("\nTest 4: List files")
    result = await execute_code_direct("ls -la /tmp/*.py")
    print(f"Success: {result['success']}")
    print(f"Output: {result.get('output', result.get('error'))}")
    
    print("\n" + "=" * 60)
    print("Direct execution test complete!")
    print("=" * 60)


if __name__ == "__main__":
    asyncio.run(test_execute_code())