Spaces:
Running
Running
File size: 7,029 Bytes
3ae68d6 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 | import os
import re
import sys
import uuid
import stat
import shutil
import subprocess
from typing import Tuple
def remove_readonly(func, path, excinfo):
"""
On Windows, files can sometimes be marked read-only or locked,
preventing shutil.rmtree from working. This helper removes
the read-only attribute and retries the removal.
"""
try:
os.chmod(path, stat.S_IWRITE)
func(path)
except Exception:
pass
def run_tests(language: str, code: str, tests: str) -> Tuple[bool, str]:
"""
Writes the wrapper code and unit test code to an isolated temporary sandbox directory,
runs the language-specific test suite via a subprocess, and returns a tuple
of (test_passed, console_logs).
"""
# Create a unique sandbox directory under a root 'temp' folder
root_temp_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..", "temp"))
os.makedirs(root_temp_dir, exist_ok=True)
run_id = str(uuid.uuid4())
sandbox_dir = os.path.join(root_temp_dir, f"run_{run_id}")
os.makedirs(sandbox_dir, exist_ok=True)
language = language.lower().strip()
timeout_val = 30 if language in ["go", "golang", "java"] else 15
code_filename = ""
test_filename = ""
run_args = []
env = os.environ.copy()
# Enable Java assertions by default
if "JAVA_TOOL_OPTIONS" not in env:
env["JAVA_TOOL_OPTIONS"] = "-ea"
else:
env["JAVA_TOOL_OPTIONS"] += " -ea"
try:
if language in ["python", "py"]:
code_filename = "client.py"
test_filename = "test_client.py"
# Add sandbox directory to PYTHONPATH so pytest can import the client module
env["PYTHONPATH"] = f"{sandbox_dir}{os.pathsep}{env.get('PYTHONPATH', '')}"
# Use sys.executable to ensure we use the virtual environment's interpreter
run_args = [sys.executable, "-m", "pytest", "--tb=short", test_filename]
elif language in ["javascript", "js"]:
code_filename = "client.js"
test_filename = "test_client.test.js"
run_args = ["node", test_filename]
elif language in ["typescript", "ts"]:
code_filename = "client.ts"
test_filename = "test_client.test.ts"
# Write a basic tsconfig.json so ts-node works consistently
tsconfig_path = os.path.join(sandbox_dir, "tsconfig.json")
tsconfig_content = """{
"compilerOptions": {
"target": "es2020",
"module": "commonjs",
"esModuleInterop": true,
"strict": false,
"skipLibCheck": true
}
}"""
with open(tsconfig_path, "w", encoding="utf-8") as f:
f.write(tsconfig_content)
run_args = ["npx", "ts-node", "--transpile-only", test_filename]
elif language in ["go", "golang"]:
code_filename = "client.go"
test_filename = "client_test.go"
# Initialize a temporary go module to prevent module loading errors
subprocess.run(
["go", "mod", "init", "sandbox"],
cwd=sandbox_dir,
capture_output=True,
timeout=5
)
run_args = ["go", "test", "-v", code_filename, test_filename]
elif language in ["java"]:
# Strip package declarations to avoid compile/runtime classpath resolution issues
code = re.sub(r"^\s*package\s+[\w\.]+;\s*", "", code, flags=re.MULTILINE)
tests = re.sub(r"^\s*package\s+[\w\.]+;\s*", "", tests, flags=re.MULTILINE)
def find_class_name(source: str, default: str) -> str:
# Matches public/non-public class names safely
match = re.search(r"(?:public\s+)?class\s+(\w+)", source)
return match.group(1) if match else default
code_classname = find_class_name(code, "MyAPIClient")
test_classname = find_class_name(tests, "TestClient")
code_filename = f"{code_classname}.java"
test_filename = f"{test_classname}.java"
else:
raise ValueError(f"Unsupported language for self-healing execution: {language}")
# Write files to sandbox
code_path = os.path.join(sandbox_dir, code_filename)
test_path = os.path.join(sandbox_dir, test_filename)
with open(code_path, "w", encoding="utf-8") as f:
f.write(code)
with open(test_path, "w", encoding="utf-8") as f:
f.write(tests)
# Execute the subprocess
if language == "java":
# 1. Compile Java files
compile_process = subprocess.run(
["javac", code_filename, test_filename],
cwd=sandbox_dir,
capture_output=True,
text=True,
timeout=timeout_val
)
if compile_process.returncode != 0:
return False, f"Compilation Error:\n{compile_process.stderr}\nStdout:\n{compile_process.stdout}"
# 2. Execute Java test entrypoint
test_class = test_filename[:-5]
run_process = subprocess.run(
["java", "-ea", test_class],
cwd=sandbox_dir,
capture_output=True,
text=True,
timeout=timeout_val,
env=env
)
success = (run_process.returncode == 0)
logs = f"Stdout:\n{run_process.stdout}\nStderr:\n{run_process.stderr}"
return success, logs
else:
# On Windows, prepend cmd.exe /c for CMD-based tools like npx
actual_args = run_args
if os.name == "nt" and run_args and run_args[0] == "npx":
actual_args = ["cmd.exe", "/c"] + run_args
# Run the command with strict timeout to prevent CPU hung states
process = subprocess.run(
actual_args,
cwd=sandbox_dir,
capture_output=True,
text=True,
timeout=timeout_val,
env=env
)
success = (process.returncode == 0)
logs = f"Stdout:\n{process.stdout}\nStderr:\n{process.stderr}"
return success, logs
except subprocess.TimeoutExpired as e:
return False, f"Execution timed out after 15 seconds.\nStdout:\n{e.stdout}\nStderr:\n{e.stderr}"
except Exception as e:
return False, f"Executor encountered an internal exception: {str(e)}"
finally:
# Clean up sandbox directory using the Windows-resilient onerror handler
try:
shutil.rmtree(sandbox_dir, onerror=remove_readonly)
except Exception:
pass
|