water3 / test_execute_direct.py
onewayto's picture
Upload 187 files
070daf8 verified
#!/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())