diff options
Diffstat (limited to 'code_eval/OpenCodeEval/eval')
| -rw-r--r-- | code_eval/OpenCodeEval/eval/__init__.py | 0 | ||||
| -rw-r--r-- | code_eval/OpenCodeEval/eval/func_eval.py | 232 | ||||
| -rw-r--r-- | code_eval/OpenCodeEval/eval/sanitize.py | 196 | ||||
| -rw-r--r-- | code_eval/OpenCodeEval/eval/sql_test.py | 187 | ||||
| -rw-r--r-- | code_eval/OpenCodeEval/eval/unit_test.py | 472 |
5 files changed, 1087 insertions, 0 deletions
diff --git a/code_eval/OpenCodeEval/eval/__init__.py b/code_eval/OpenCodeEval/eval/__init__.py new file mode 100644 index 0000000..e69de29 --- /dev/null +++ b/code_eval/OpenCodeEval/eval/__init__.py diff --git a/code_eval/OpenCodeEval/eval/func_eval.py b/code_eval/OpenCodeEval/eval/func_eval.py new file mode 100644 index 0000000..493c0c6 --- /dev/null +++ b/code_eval/OpenCodeEval/eval/func_eval.py @@ -0,0 +1,232 @@ +import io +import os + +import signal +import tempfile +import platform +import contextlib +import faulthandler +import multiprocessing + +from typing import Optional, Callable, Dict + + +# WARNING +# This program exists to execute untrusted model-generated code. Although +# it is highly unlikely that model-generated code will do something overtly +# malicious in response to this test suite, model-generated code may act +# destructively due to a lack of model capability or alignment. +# Users are strongly encouraged to sandbox this evaluation suite so that it +# does not perform destructive actions on their host or network. For more +# information on how OpenAI sandboxes its code, see the accompanying paper. +# Once you have read this disclaimer and taken appropriate precautions, +# uncomment the 58 line and proceed at your own risk + +def check_correctness(task_id: int, + completion_id: int, + solution: str, + time_out: float, + ) -> Dict: + """ + Evaluates the functional correctness of a completion by running the test + suite provided in the problem. + + :param completion_id: an optional completion ID so we can match + the results later even if execution finishes asynchronously. + """ + + def unsafe_execute(): + + with create_tempdir(): + + # These system calls are needed when cleaning up tempdir. + import os + import shutil + rmtree = shutil.rmtree + rmdir = os.rmdir + chdir = os.chdir + + # Disable functionalities that can make destructive changes to the test. + reliability_guard() + + # Construct the check program and run it. + check_program = solution + + try: + exec_globals = {} + with swallow_io(): + with time_limit(time_out): + exec(check_program, exec_globals) + result.append("passed") + except TimeoutException: + result.append("timed out") + except BaseException as e: + result.append(f"failed: {e}") + + # Needed for cleaning up. + shutil.rmtree = rmtree + os.rmdir = rmdir + os.chdir = chdir + + manager = multiprocessing.Manager() + result = manager.list() + + p = multiprocessing.Process(target=unsafe_execute) + p.start() + p.join(timeout=time_out + 1) + if p.is_alive(): + p.kill() + + if not result: + result.append("timed out") + + return dict( + task_id = task_id, + completion_id = completion_id, + passed = result[0] == "passed", + result = result[0], + solution = solution + ) + + +@contextlib.contextmanager +def time_limit(seconds: float): + def signal_handler(signum, frame): + raise TimeoutException("Timed out!") + signal.setitimer(signal.ITIMER_REAL, seconds) + signal.signal(signal.SIGALRM, signal_handler) + try: + yield + finally: + signal.setitimer(signal.ITIMER_REAL, 0) + + +@contextlib.contextmanager +def swallow_io(): + stream = WriteOnlyStringIO() + with contextlib.redirect_stdout(stream): + with contextlib.redirect_stderr(stream): + with redirect_stdin(stream): + yield + + +@contextlib.contextmanager +def create_tempdir(): + with tempfile.TemporaryDirectory() as dirname: + with chdir(dirname): + yield dirname + + +class TimeoutException(Exception): + pass + + +class WriteOnlyStringIO(io.StringIO): + """ StringIO that throws an exception when it's read from """ + + def read(self, *args, **kwargs): + raise IOError + + def readline(self, *args, **kwargs): + raise IOError + + def readlines(self, *args, **kwargs): + raise IOError + + def readable(self, *args, **kwargs): + """ Returns True if the IO object can be read. """ + return False + + +class redirect_stdin(contextlib._RedirectStream): # type: ignore + _stream = 'stdin' + + +@contextlib.contextmanager +def chdir(root): + if root == ".": + yield + return + cwd = os.getcwd() + os.chdir(root) + try: + yield + except BaseException as exc: + raise exc + finally: + os.chdir(cwd) + + +def reliability_guard(maximum_memory_bytes: Optional[int] = None): + """ + This disables various destructive functions and prevents the generated code + from interfering with the test (e.g. fork bomb, killing other processes, + removing filesystem files, etc.) + + WARNING + This function is NOT a security sandbox. Untrusted code, including, model- + generated code, should not be blindly executed outside of one. See the + Codex paper for more information about OpenAI's code sandbox, and proceed + with caution. + """ + + if maximum_memory_bytes is not None: + import resource + resource.setrlimit(resource.RLIMIT_AS, (maximum_memory_bytes, maximum_memory_bytes)) + resource.setrlimit(resource.RLIMIT_DATA, (maximum_memory_bytes, maximum_memory_bytes)) + if not platform.uname().system == 'Darwin': + resource.setrlimit(resource.RLIMIT_STACK, (maximum_memory_bytes, maximum_memory_bytes)) + + faulthandler.disable() + + import builtins + builtins.exit = None + builtins.quit = None + + import os + os.environ['OMP_NUM_THREADS'] = '1' + + os.kill = None + os.system = None + os.putenv = None + os.remove = None + os.removedirs = None + os.rmdir = None + os.fchdir = None + os.setuid = None + os.fork = None + os.forkpty = None + os.killpg = None + os.rename = None + os.renames = None + os.truncate = None + os.replace = None + os.unlink = None + os.fchmod = None + os.fchown = None + os.chmod = None + os.chown = None + os.chroot = None + os.fchdir = None + os.lchflags = None + os.lchmod = None + os.lchown = None + os.getcwd = None + os.chdir = None + + import shutil + shutil.rmtree = None + shutil.move = None + shutil.chown = None + + import subprocess + subprocess.Popen = None # type: ignore + + __builtins__['help'] = None + + import sys + sys.modules['ipdb'] = None + sys.modules['joblib'] = None + sys.modules['resource'] = None + sys.modules['psutil'] = None + sys.modules['tkinter'] = None
\ No newline at end of file diff --git a/code_eval/OpenCodeEval/eval/sanitize.py b/code_eval/OpenCodeEval/eval/sanitize.py new file mode 100644 index 0000000..a82ea65 --- /dev/null +++ b/code_eval/OpenCodeEval/eval/sanitize.py @@ -0,0 +1,196 @@ +"""Post-processing LLM-generated Python code implemented using AST.""" +import ast +import traceback +import signal + +from typing import Dict, List, Optional, Set, Tuple +from OpenCodeEval.utils import refine_text + +def syntax_check(code, verbose = False): + try: + ast.parse(code) + return True + except (SyntaxError, MemoryError): + if verbose: + traceback.print_exc() + return False + +def extract_longest_valid_code(text: str) -> str: + lines = text.splitlines() + + max_valid_lines = 0 + max_valid_snippet = "" + + for i in range(len(lines)): + for j in range(i, len(lines)): + current_snippet = "\n".join(lines[i:j+1]) + if syntax_check(current_snippet): + valid_line_count = sum(1 for line in lines[i:j+1] if line.strip()) + if valid_line_count > max_valid_lines: + max_valid_lines = valid_line_count + max_valid_snippet = current_snippet + + return max_valid_snippet + +def get_deps(nodes: List[Tuple[str, ast.AST]]) -> Dict[str, Set[str]]: + name2deps = {} + for name, node in nodes: + deps = set() + stack = [node] + while stack: + current = stack.pop() + for child in ast.iter_child_nodes(current): + if isinstance(child, ast.Name): + deps.add(child.id) + elif isinstance(child, ast.Attribute): + deps.add(child.attr) + else: + stack.append(child) + name2deps[name] = deps + return name2deps + +def get_function_dependency(entrypoint: str, call_graph: Dict[str, Set[str]]) -> Set[str]: + visited = set() + to_visit = [entrypoint] + + while to_visit: + current = to_visit.pop(0) + if current not in visited: + visited.add(current) + to_visit.extend(call_graph.get(current, set()) - visited) + + return visited + +def get_definition_name(node: ast.AST) -> Optional[str]: + if isinstance(node, (ast.FunctionDef, ast.ClassDef)): + return node.name + elif isinstance(node, ast.Assign): + targets = node.targets + if targets and isinstance(targets[0], ast.Name): + return targets[0].id + return None + +def has_return_statement(node: ast.AST) -> bool: + return any(isinstance(n, ast.Return) for n in ast.walk(node)) + +def has_yield_statement(node: ast.AST) -> bool: + return any(isinstance(n, ast.Yield) for n in ast.walk(node)) + +class TimeoutException(Exception): pass + +def _timeout_handler(signum, frame): + raise TimeoutException() +""" +def sanitize(text: str, entrypoint: Optional[str] = None) -> str: + # 设置信号处理器 + signal.signal(signal.SIGALRM, _timeout_handler) + signal.alarm(30) # 30秒后发送SIGALRM + + text = refine_text(text) + + try: + code = extract_longest_valid_code(text) + + tree = ast.parse(code) + + definitions = {} + + imports = [] + + for node in tree.body: + if isinstance(node, (ast.Import, ast.ImportFrom)): + imports.append(node) + elif isinstance(node, ast.ClassDef): + name = node.name + definitions[name] = ('class', node) + elif isinstance(node, ast.FunctionDef): + name = node.name + if has_return_statement(node) or has_yield_statement(node): + definitions[name] = ('function', node) + elif isinstance(node, ast.Assign): + name = get_definition_name(node) + if name: + definitions[name] = ('variable', node) + + if entrypoint: + name2deps = get_deps([(name, node) for name, (_, node) in definitions.items()]) + reachable = get_function_dependency(entrypoint, name2deps) + + sanitized_output = [] + + for node in imports: + sanitized_output.append(ast.unparse(node)) + + for name, (_, node) in definitions.items(): + if not entrypoint or name in reachable: + sanitized_output.append(ast.unparse(node)) + + return "\n".join(sanitized_output) + + except Exception as e: + print(f"Error extracting longest valid code: {e}") + return "" + finally: + signal.alarm(0) # 取消定时器 +""" +from multiprocessing import Process, Queue, TimeoutError + +def sanitize(text: str, entrypoint: Optional[str] = None) -> str: + def _sanitize_worker(q, text, entrypoint): + try: + # 原有处理逻辑保持不变 + text = refine_text(text) + code = extract_longest_valid_code(text) + tree = ast.parse(code) + definitions = {} + imports = [] + for node in tree.body: + if isinstance(node, (ast.Import, ast.ImportFrom)): + imports.append(node) + elif isinstance(node, ast.ClassDef): + name = node.name + definitions[name] = ('class', node) + elif isinstance(node, ast.FunctionDef): + name = node.name + if has_return_statement(node) or has_yield_statement(node): + definitions[name] = ('function', node) + elif isinstance(node, ast.Assign): + name = get_definition_name(node) + if name: + definitions[name] = ('variable', node) + + if entrypoint: + name2deps = get_deps([(name, node) for name, (_, node) in definitions.items()]) + reachable = get_function_dependency(entrypoint, name2deps) + + sanitized_output = [] + for node in imports: + sanitized_output.append(ast.unparse(node)) + for name, (_, node) in definitions.items(): + if not entrypoint or name in reachable: + sanitized_output.append(ast.unparse(node)) + + q.put("\n".join(sanitized_output)) + + except Exception as e: + print(f"Error extracting longest valid code: {e}") + q.put("") + + # 使用多进程实现超时控制 + q = Queue() + p = Process(target=_sanitize_worker, args=(q, text, entrypoint)) + p.start() + + try: + # 等待3秒获取结果 + result = q.get(timeout=3) + + p.terminate() + p.join() # 确保进程退出 + return result + except TimeoutError: + print("Function timed out after 3 seconds") + p.terminate() # 强制终止进程 + p.join() + return "" +
\ No newline at end of file diff --git a/code_eval/OpenCodeEval/eval/sql_test.py b/code_eval/OpenCodeEval/eval/sql_test.py new file mode 100644 index 0000000..9b0724e --- /dev/null +++ b/code_eval/OpenCodeEval/eval/sql_test.py @@ -0,0 +1,187 @@ +import re +import random +import sqlite3 + +from loguru import logger +from itertools import product +from collections import defaultdict +from typing import Tuple, List, Set +from func_timeout import func_timeout, FunctionTimedOut + +def permute_tuple(element: Tuple, perm: Tuple) -> Tuple: + assert len(element) == len(perm) + return tuple([element[i] for i in perm]) + + +def unorder_row(row: Tuple) -> Tuple: + return tuple(sorted(row, key=lambda x: str(x) + str(type(x)))) + + +# unorder each row in the table +# [result_1 and result_2 has the same bag of unordered row] +# is a necessary condition of +# [result_1 and result_2 are equivalent in denotation] +def quick_rej(result1: List[Tuple], result2: List[Tuple], order_matters: bool) -> bool: + s1 = [unorder_row(row) for row in result1] + s2 = [unorder_row(row) for row in result2] + if order_matters: + return s1 == s2 + else: + return set(s1) == set(s2) + +# return whether two bag of relations are equivalent +def multiset_eq(l1: List, l2: List) -> bool: + if len(l1) != len(l2): + return False + d = defaultdict(int) + for e in l1: + d[e] = d[e] + 1 + for e in l2: + d[e] = d[e] - 1 + if d[e] < 0: + return False + return True + +def get_constraint_permutation(tab1_sets_by_columns: List[Set], result2: List[Tuple]): + num_cols = len(result2[0]) + perm_constraints = [{i for i in range(num_cols)} for _ in range(num_cols)] + if num_cols <= 3: + return product(*perm_constraints) + + # we sample 20 rows and constrain the space of permutations + for _ in range(20): + random_tab2_row = random.choice(result2) + + for tab1_col in range(num_cols): + for tab2_col in set(perm_constraints[tab1_col]): + if random_tab2_row[tab2_col] not in tab1_sets_by_columns[tab1_col]: + perm_constraints[tab1_col].remove(tab2_col) + return product(*perm_constraints) + +# check whether two denotations are correct +def result_eq(result1: List[Tuple], result2: List[Tuple], order_matters: bool) -> bool: + if len(result1) == 0 and len(result2) == 0: + return True + + # if length is not the same, then they are definitely different bag of rows + if len(result1) != len(result2): + return False + + num_cols = len(result1[0]) + + # if the results do not have the same number of columns, they are different + if len(result2[0]) != num_cols: + return False + + # unorder each row and compare whether the denotation is the same + # this can already find most pair of denotations that are different + if not quick_rej(result1, result2, order_matters): + return False + + # the rest of the problem is in fact more complicated than one might think + # we want to find a permutation of column order and a permutation of row order, + # s.t. result_1 is the same as result_2 + # we return true if we can find such column & row permutations + # and false if we cannot + tab1_sets_by_columns = [{row[i] for row in result1} for i in range(num_cols)] + + # on a high level, we enumerate all possible column permutations that might make result_1 == result_2 + # we decrease the size of the column permutation space by the function get_constraint_permutation + # if one of the permutation make result_1, result_2 equivalent, then they are equivalent + for perm in get_constraint_permutation(tab1_sets_by_columns, result2): + if len(perm) != len(set(perm)): + continue + if num_cols == 1: + result2_perm = result2 + else: + result2_perm = [permute_tuple(element, perm) for element in result2] + if order_matters: + if result1 == result2_perm: + return True + else: + # in fact the first condition must hold if the second condition holds + # but the first is way more efficient implementation-wise + # and we use it to quickly reject impossible candidates + if set(result1) == set(result2_perm) and multiset_eq(result1, result2_perm): + return True + return False + +# postprocess the model predictions to avoid execution errors +# e.g. removing spaces between ">" and "=" +def postprocess(query: str) -> str: + query = query.replace("> =", ">=").replace("< =", "<=").replace("! =", "!=") + return query + +def find_detail(reason): + patterns = [ + "ambiguous column name", + "no such column", + "no such table", + "no such function", + "syntax error", + ] + + for p in patterns: + if p in reason: + return p + return "others" + +def execute_sql(predicted_sql,ground_truth, db_path, method): + conn = sqlite3.connect(db_path) + # Connect to the database + cursor = conn.cursor() + cursor.execute(predicted_sql) + predicted_res = cursor.fetchall() + cursor.execute(ground_truth) + ground_truth_res = cursor.fetchall() + result = "failed" + passed = False + + if predicted_res is None: + sql_return = [[]] + else: + sql_return = predicted_res + + if method == "set_match": + + if set(predicted_res) == set(ground_truth_res): + result = "passed" + passed = True + return result, passed, sql_return + + elif method == "exact_match": + + order_matters = "orderby" in re.sub(r"\s+", ground_truth.lower(), "") + if result_eq(predicted_res, ground_truth_res, order_matters = order_matters): + result = "passed" + passed = True + else: + passed = False + + return result, passed, sql_return + + else: + logger.error(f"Unknown evaluation method: {method}") + +def check_correctness(predicted_sql,ground_truth, db_path, meta_time_out, method): + try: + result, passed, sql_return = func_timeout( + meta_time_out, + execute_sql, + args = ( + predicted_sql, + ground_truth, + db_path, + method + ) + ) + except FunctionTimedOut: + result = "timeout" + passed = False + sql_return = [[]] + except Exception as e: + result = f"error:{e}" + passed = False + sql_return = [[]] + + return result, passed, sql_return
\ No newline at end of file diff --git a/code_eval/OpenCodeEval/eval/unit_test.py b/code_eval/OpenCodeEval/eval/unit_test.py new file mode 100644 index 0000000..5dd1f49 --- /dev/null +++ b/code_eval/OpenCodeEval/eval/unit_test.py @@ -0,0 +1,472 @@ +# The MIT License +# +# Copyright (c) OpenAI (https://openai.com) +# +# Permission is hereby granted, free of charge, to any person obtaining a copy +# of this software and associated documentation files (the "Software"), to deal +# in the Software without restriction, including without limitation the rights +# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +# copies of the Software, and to permit persons to whom the Software is +# furnished to do so, subject to the following conditions: +# +# The above copyright notice and this permission notice shall be included in +# all copies or substantial portions of the Software. +# +# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN +# THE SOFTWARE. + +import io +import os +import sys +import time +import types +import unittest +import contextlib +import faulthandler +import platform +import signal +import tempfile +import subprocess +import multiprocessing +import numpy as np + +from multiprocessing import Array, Value, Manager +from typing import Any, Dict, List, Tuple, Union + +@contextlib.contextmanager +def swallow_subprocess_output(): + """Context manager to swallow stdout and stderr for subprocesses.""" + original_popen = subprocess.Popen + original_run = subprocess.run + + def _popen_patch(*args, **kwargs): + if 'capture_output' in kwargs and kwargs['capture_output']: + # Avoid setting stdout or stderr if capture_output is True + kwargs.pop('stdout', None) + kwargs.pop('stderr', None) + else: + kwargs.setdefault('stdout', subprocess.PIPE) + kwargs.setdefault('stderr', subprocess.PIPE) + return original_popen(*args, **kwargs) + + def _run_patch(*args, **kwargs): + if 'capture_output' in kwargs and kwargs['capture_output']: + # Avoid setting stdout or stderr if capture_output is True + kwargs.pop('stdout', None) + kwargs.pop('stderr', None) + else: + kwargs.setdefault('stdout', subprocess.PIPE) + kwargs.setdefault('stderr', subprocess.PIPE) + return original_run(*args, **kwargs) + + subprocess.Popen = _popen_patch + subprocess.run = _run_patch + try: + yield + finally: + subprocess.Popen = original_popen + subprocess.run = original_run + +@contextlib.contextmanager +def swallow_io(): + stream = WriteOnlyStringIO() + with contextlib.redirect_stdout(stream): + with contextlib.redirect_stderr(stream): + with redirect_stdin(stream): + with swallow_subprocess_output(): + yield + + +@contextlib.contextmanager +def time_limit(seconds: float): + def signal_handler(signum, frame): + raise TimeoutException("Timed out!") + + signal.setitimer(signal.ITIMER_REAL, seconds) + signal.signal(signal.SIGALRM, signal_handler) + try: + yield + finally: + signal.setitimer(signal.ITIMER_REAL, 0) + + +@contextlib.contextmanager +def create_tempdir(): + with tempfile.TemporaryDirectory() as dirname: + with chdir(dirname): + yield dirname + + +@contextlib.contextmanager +def chdir(root): + if root == ".": + yield + return + cwd = os.getcwd() + os.chdir(root) + try: + yield + except BaseException as exc: + raise exc + finally: + os.chdir(cwd) + + +@contextlib.contextmanager +def safe_environment(): + # Save original functions + original_kill = os.kill + original_killpg = os.killpg + original_system = os.system + original_subprocess_call = subprocess.call + original_subprocess_check_output = subprocess.check_output + original_subprocess_run = subprocess.run + original_subprocess_popen = subprocess.Popen + original_os_popen = os.popen + original_os_execv = os.execv + original_os_execvp = os.execvp + original_os_execvpe = os.execvpe + + current_pid = os.getpid() + current_pgid = os.getpgid(current_pid) + manager = multiprocessing.Manager() + child_pids = manager.list() + + def safe_kill(pid, sig): + try: + pgid = os.getpgid(pid) + if pid == current_pid or pid in child_pids: + original_kill(pid, sig) + else: + print(f"Prevented attempt to kill PID {pid} with signal {sig}") + except ProcessLookupError: + pass + + def safe_killpg(pgid, sig): + if pgid == current_pgid or pgid in {os.getpgid(pid) for pid in child_pids}: + original_killpg(pgid, sig) + else: + print(f"Prevented attempt to kill PGID {pgid} with signal {sig}") + + def safe_system(command): + print(f"Intercepted system command: {command}") + if 'kill' in command or 'killall' in command: + return 0 # Simulate successful execution without doing anything + return original_system(command) + + def safe_subprocess_call(command, *args, **kwargs): + print(f"Intercepted subprocess call: {command}") + if 'kill' in command or 'killall' in command: + return 0 # Simulate successful execution without doing anything + return original_subprocess_call(command, *args, **kwargs) + + def safe_subprocess_check_output(command, *args, **kwargs): + print(f"Intercepted command: {command}") + if 'ps' in command: + return b"" # Simulate no processes found + return original_subprocess_check_output(command, *args, **kwargs) + + def safe_subprocess_run(*args, **kwargs): + print(f"Intercepted subprocess run command: {args}") + if 'kill' in args[0] or 'killall' in args[0]: + return subprocess.CompletedProcess(args, 0, b'', b'') # Simulate successful execution + return original_subprocess_run(*args, **kwargs) + + class SafePopen(subprocess.Popen): + def __init__(self, *args, **kwargs): + print(f"Intercepted Popen command: {args}") + kwargs['preexec_fn'] = os.setsid # Start the process in a new session + super().__init__(*args, **kwargs) + child_pids.append(self.pid) + + def communicate(self, *args, **kwargs): + try: + return super().communicate(*args, **kwargs) + except subprocess.TimeoutExpired: + print("Timeout expired, intercepted and returning None") + return None, None + + def kill(self): + print(f"Intercepted kill call for PID {self.pid}") + safe_kill(self.pid, signal.SIGTERM) + + def terminate(self): + print(f"Intercepted terminate call for PID {self.pid}") + safe_kill(self.pid, signal.SIGTERM) + + def safe_os_popen(command): + print(f"Intercepted os.popen command: {command}") + if 'kill' in command or 'killall' in command: + return os.popen('echo Intercepted') + return original_os_popen(command) + + def safe_exec(*args, **kwargs): + print(f"Intercepted exec command: {args}") + + # Override the risky functions with the safe versions + os.kill = safe_kill + os.killpg = safe_killpg + os.system = safe_system + subprocess.call = safe_subprocess_call + subprocess.check_output = safe_subprocess_check_output + subprocess.run = safe_subprocess_run + subprocess.Popen = SafePopen + os.popen = safe_os_popen + os.execv = safe_exec + os.execvp = safe_exec + os.execvpe = safe_exec + + try: + yield + finally: + for pid in child_pids: + try: + os.kill(pid, signal.SIGTERM) + for _ in range(10): + time.sleep(0.1) + try: + os.kill(pid, 0) + except ProcessLookupError: + break + else: + os.kill(pid, signal.SIGKILL) + except ProcessLookupError: + pass + except Exception as e: + print(f"Error handling process {pid}: {e}") + + os.kill = original_kill + os.killpg = original_killpg + os.system = original_system + subprocess.call = original_subprocess_call + subprocess.check_output = original_subprocess_check_output + subprocess.run = original_subprocess_run + subprocess.Popen = original_subprocess_popen + os.popen = original_os_popen + os.execv = original_os_execv + os.execvp = original_os_execvp + os.execvpe = original_os_execvpe + + +class TimeoutException(Exception): + pass + + +class WriteOnlyStringIO(io.StringIO): + """StringIO that throws an exception when it's read from""" + + def read(self, *args, **kwargs): + raise IOError + + def readline(self, *args, **kwargs): + raise IOError + + def readlines(self, *args, **kwargs): + raise IOError + + def readable(self, *args, **kwargs): + """Returns True if the IO object can be read.""" + return False + + +class redirect_stdin(contextlib._RedirectStream): # type: ignore + _stream = "stdin" + + +def reliability_guard(max_as_limit, max_data_limit, max_stack_limit): + """ + This disables various destructive functions and prevents the generated code + from interfering with the test (e.g. fork bomb, killing other processes, + removing filesystem files, etc.) + + WARNING + This function is NOT a security sandbox. Untrusted code, including, model- + generated code, should not be blindly executed outside of one. See the + Codex paper for more information about OpenAI's code sandbox, and proceed + with caution. + """ + + import os + import time + from datetime import datetime + + os.environ['TZ'] = 'UTC' + time.tzset() + + os.environ["OMP_NUM_THREADS"] = "1" + os.environ['TF_CPP_MIN_LOG_LEVEL'] = "3" + os.environ['TF_ENABLE_ONEDNN_OPTS'] = "0" + + if max_as_limit and max_data_limit and max_stack_limit: + import resource + + max_as_limit = max_as_limit * 1024 * 1024 + max_data_limit = max_data_limit * 1024 * 1024 + max_stack_limit = max_stack_limit * 1024 * 1024 + + resource.setrlimit( + resource.RLIMIT_AS, (max_as_limit, max_as_limit) + ) + resource.setrlimit( + resource.RLIMIT_DATA, (max_data_limit, max_data_limit) + ) + if not platform.uname().system == "Darwin": + resource.setrlimit( + resource.RLIMIT_STACK, (max_stack_limit, max_stack_limit) + ) + + faulthandler.disable() + + import builtins + + builtins.exit = None + builtins.quit = None + + import matplotlib.pyplot as plt + plt.close('all') + + +PASS = "pass" +FAIL = "fail" +TIMEOUT = "timeout" + +_SUCCESS = 0 +_FAILED = 1 +_TIMEOUT = 2 +_UNKNOWN = 3 + +_mapping = {_SUCCESS: PASS, _FAILED: FAIL, _TIMEOUT: TIMEOUT, _UNKNOWN: None} + + +def unsafe_execute( + code: str, + test_code: str, + timeout: float, + stat, # Value + details, # Array +): + with safe_environment(), create_tempdir(): + # These system calls are needed when cleaning up tempdir. + import os + import shutil + import builtins + + rmtree = shutil.rmtree + rmdir = os.rmdir + chdir = os.chdir + # Disable functionalities that can make destructive changes to the test. + reliability_guard(max_as_limit = 30720, max_data_limit = 30720, max_stack_limit = 10) + module_name = "__test__" + new_module = types.ModuleType(module_name) + # Set necessary attributes for the module + new_module.__dict__.update({ + '__builtins__': builtins, + '__file__': f"{module_name}.py", + '__package__': None, + '__doc__': None, + 'sys': sys, + 'os': os, + 'environ': os.environ, + }) + + try: + full_code = code + "\n" + test_code + + with swallow_io(): + exec(compile(full_code, f"{module_name}.py", 'exec'), new_module.__dict__) + sys.modules[module_name] = new_module + TestCases = getattr(new_module, 'TestCases') + loader = unittest.TestLoader() + suite = loader.loadTestsFromTestCase(TestCases) + test_result = unittest.TestResult() + + with time_limit(timeout): + suite.run(test_result) + + issues = test_result.failures + test_result.errors + for test, trace in issues: + details[test.id().split(".")[-1]] = trace + stat.value = _SUCCESS + except BaseException as e: + details["ALL"] = str(e) + stat.value = _FAILED + # Needed for cleaning up. + shutil.rmtree = rmtree + os.rmdir = rmdir + os.chdir = chdir + + +import psutil + +def terminate_process_tree(pid): + try: + parent = psutil.Process(pid) + children = parent.children(recursive=True) + for child in children: + try: + if child.is_running(): + os.kill(child.pid, signal.SIGKILL) + except psutil.NoSuchProcess: + continue + if parent.is_running(): + os.kill(parent.pid, signal.SIGKILL) + except psutil.NoSuchProcess: + pass + +def check_correctness( + task_id: int, + solution_id: int, + solution: str, + test: str, + timeout: float, +) -> Tuple[str, np.ndarray]: + + result = { + "task_id": task_id, + "solution_id": solution_id + } + + # shared memory objects + stat = Value("i", _UNKNOWN) + manager = Manager() + details = manager.dict() + + p = multiprocessing.Process( + target=unsafe_execute, + args=( + solution, + test, + timeout, + stat, + details, + ), + ) + + p.start() + p.join(timeout=timeout+1) + + if p.is_alive(): + terminate_process_tree(p.pid) + stat.value = _TIMEOUT + + stat = _mapping[stat.value] + details = dict(details) + + if not stat: + stat = TIMEOUT + if stat == PASS: + if details: + stat = FAIL + + result["passed"] = stat == PASS + result["result"] = details + result["solution"] = solution + + manager.shutdown() + + return result
\ No newline at end of file |