evaluation/eval/tools/code_verifier_utils.py

import ast
import json
import sys
import faulthandler
import platform
import re
# used for debugging to time steps
from datetime import datetime
import os
# to run the solution files we're using a timing based approach
import signal
import time
import numpy as np

from io import StringIO

# used for testing the code that reads from input
from unittest.mock import patch, mock_open

# from pyext import RuntimeModule
from types import ModuleType

from enum import Enum
from decimal import Decimal
import time
from typing import List

import_string = "import sys\nsys.setrecursionlimit(1 << 25)\nsys.set_int_max_str_digits(1 << 25)\nimport time\nimport re\nimport bisect\nimport itertools\nfrom itertools import accumulate, product, permutations, combinations\nimport collections\nfrom collections import Counter, OrderedDict, deque, defaultdict, ChainMap\nfrom functools import lru_cache\nimport math\nfrom math import sqrt, sin, cos, tan, ceil, fabs, floor, gcd, exp, log, log2\nimport fractions\nfrom typing import List, Tuple\nimport numpy as np\nimport random\nimport heapq\nfrom heapq import *\n"

def convert_python_literal_to_json(s):
    obj = ast.literal_eval(s)
    return json.dumps(obj)

def convert_python_literal(arg_str):
    s = arg_str.strip()
    # If the whole string starts with '[' and ends with ']', assume it's one literal
    # and force it to be treated as a single-element tuple by adding a trailing comma.
    if s.startswith('[') and s.endswith(']'):
        wrapped_str = f"({arg_str},)"
    else:
        wrapped_str = f"({arg_str})"

    try:
        parsed_tuple = ast.literal_eval(wrapped_str)
    except Exception as e:
        raise ValueError(f"Error parsing argument string: {arg_str}") from e
    return list(parsed_tuple)

def post_process_code(code):
    code = code.split("</code>")[0]
    code = code.replace("```python", "")
    code = code.split("```")[0]
    code = code.replace("<code>", "")
    return code

def extract_functions(code_str):
    tree = ast.parse(code_str)
    extracted = []

    # Iterate over top-level nodes in the module.
    for node in tree.body:
        # Check for import statements.
        if isinstance(node, (ast.Import, ast.ImportFrom, ast.Assign)):
            code_segment = ast.get_source_segment(code_str, node)
            extracted.append(code_segment)
        # Check for top-level function definitions.
        elif isinstance(node, ast.FunctionDef):
            code_segment = ast.get_source_segment(code_str, node)
            extracted.append(code_segment)
        # Check for class definitions.
        elif isinstance(node, ast.ClassDef):
            code_segment = ast.get_source_segment(code_str, node)
            extracted.append(code_segment)

    return extracted

PATTERN = re.compile(
    r"(?ms)^```python(?:\w+)?\n"          # opener at start of a line
    r"(?P<code>(?:(?!^```python).)*?)"    # content that never starts a new ###python
    r"^\s*```\s*$"                        # closer '###' on its own line
)

def has_code(response):
    matches = list(PATTERN.finditer(response))
    return matches[-1].group("code") if matches else None

def truncatefn(s, length=300):
    if isinstance(s, str):
        pass
    else:
        s = str(s)
    if len(s) <= length:
        return s

    return s[: length // 2] + "...(truncated) ..." + s[-length // 2 :]


class CODE_TYPE(Enum):
    call_based = 0
    standard_input = 1


# stuff for setting up signal timer
class TimeoutException(Exception):
    pass


def timeout_handler(signum, frame):
    print("timeout occured: alarm went off")
    raise TimeoutException


# used to capture stdout as a list
# from https://stackoverflow.com/a/16571630/6416660
# alternative use redirect_stdout() from contextlib
class Capturing(list):
    def __enter__(self):
        self._stdout = sys.stdout
        sys.stdout = self._stringio = StringIO()
        # Make closing the StringIO a no-op
        self._stringio.close = lambda x: 1
        return self

    def __exit__(self, *args):
        self.append(self._stringio.getvalue())
        del self._stringio  # free up some memory
        sys.stdout = self._stdout


def clean_if_name(code: str) -> str:
    try:
        astree = ast.parse(code)
        last_block = astree.body[-1]
        if isinstance(last_block, ast.If):
            condition = last_block.test
            if ast.unparse(condition).strip() == "__name__ == '__main__'":
                code = (
                    ast.unparse(astree.body[:-1]) + "\n" + ast.unparse(last_block.body)  # type: ignore
                )
    except:
        pass

    return code


def make_function(code: str) -> str:
    try:
        import_stmts = []
        all_other_stmts = []
        astree = ast.parse(code)
        for stmt in astree.body:
            if isinstance(stmt, (ast.Import, ast.ImportFrom)):
                import_stmts.append(stmt)
            else:
                all_other_stmts.append(stmt)

        function_ast = ast.FunctionDef(
            name="wrapped_function",
            args=ast.arguments(
                posonlyargs=[], args=[], kwonlyargs=[], kw_defaults=[], defaults=[]
            ),
            body=all_other_stmts,
            decorator_list=[],
            lineno=-1,
        )
        main_code = (
            import_string
            + "\n"
            + ast.unparse(import_stmts)  # type: ignore
            + "\n"
            + ast.unparse(function_ast)  # type: ignore
        )
        return main_code
    except Exception as e:
        return code


def call_method(method, inputs):

    if isinstance(inputs, list):
        inputs = "\n".join(inputs)

    inputs_line_iterator = iter(inputs.split("\n"))

    @patch("builtins.open", mock_open(read_data=inputs))
    @patch("sys.stdin", StringIO(inputs))
    @patch("sys.stdin.readline", lambda *args: next(inputs_line_iterator))
    @patch("sys.stdin.readlines", lambda *args: inputs.split("\n"))
    @patch("sys.stdin.read", lambda *args: inputs)
    def _inner_call_method(_method):
        try:
            return _method()
        except SystemExit as e:
            pass
        finally:
            pass

    return _inner_call_method(method)


def get_function(compiled_sol, fn_name: str):  # type: ignore
    try:
        assert hasattr(compiled_sol, fn_name)
        return getattr(compiled_sol, fn_name)
    except Exception as e:
        return


def compile_code(code: str, timeout: int):
    signal.alarm(timeout)
    try:
        tmp_sol = ModuleType("tmp_sol", "")
        exec(code, tmp_sol.__dict__)
        if "class Solution" in code:
            # leetcode wraps solutions in `Solution`
            # this is a hack to check if it is leetcode solution or not
            # currently livecodebench only supports LeetCode but
            # else condition allows future extensibility to other platforms
            compiled_sol = tmp_sol.Solution()
        else:
            # do nothing in the other case since function is accesible
            compiled_sol = tmp_sol

        assert compiled_sol is not None
    finally:
        signal.alarm(0)

    return compiled_sol


# def convert_line_to_decimals(line: str) -> tuple[bool, List[Float]]:
def convert_line_to_decimals(line: str):
    try:
        decimal_line = [float(elem) for elem in line.split()]
    except:
        return False, []
    return True, decimal_line


def get_stripped_lines(val: str):
    ## you don't want empty lines to add empty list after splitlines!
    val = val.strip()

    return [val_line.strip() for val_line in val.split("\n")]


def grade_call_based(
    code: str, all_inputs: list, all_outputs: list, fn_name: str, timeout: int
):

    # call-based clean up logic
    # need to wrap in try-catch logic after to catch the correct errors, but for now this is fine.
    base_code = import_string + "\n\n" + code + '\n\n'

    if "class Solution" in code:
        base_code += 'def run_main_code(self):\n'
    else:
        base_code += 'def run_main_code():\n'

    all_inputs = [
            inputs.replace("\n", ",") for inputs in all_inputs
    ]

    triggered = False

    try:
        all_outputs = [json.loads(output) for output in all_outputs]
        triggered = True
    except Exception:
        pass

    if not triggered:
        try:
            all_outputs = [
                json.loads(convert_python_literal_to_json(outputs)) for outputs in all_outputs
            ]
            triggered = True
        except Exception:
            pass

    if not triggered:
        all_outputs = [
            convert_python_literal(outputs) for outputs in all_outputs
        ]

    total_execution = 0
    all_results = []
    for idx, (gt_inp, gt_out) in enumerate(zip(all_inputs, all_outputs)):
        code = base_code
        if 'class Solution' in code:
            code += f"    return self.{fn_name}({gt_inp})\nSolution.run_main_code = run_main_code\n"
        else:
            code += f"    return {fn_name}({gt_inp})\n"

        compiled_sol = compile_code(code, timeout)
        if compiled_sol is None:
            return

        method = get_function(compiled_sol, "run_main_code")

        if method is None:
            return

        signal.alarm(timeout)
        faulthandler.enable()
        try:
            # can lock here so time is useful
            start = time.time()
            prediction = method()
            total_execution += time.time() - start
            signal.alarm(0)

            # don't penalize model if it produces tuples instead of lists
            # ground truth sequences are not tuples
            if isinstance(prediction, tuple):
                prediction = list(prediction)

            tmp_result = prediction == gt_out

            try:
                tmp_result = tmp_result or (json.dumps(prediction) == json.dumps(gt_out))
            except Exception:
                pass

            try:
                tmp_result = tmp_result or (np.allclose(float(prediction), float(gt_out)))
            except Exception:
                pass

            if isinstance(prediction, list):
                try:
                    if len(prediction) == len(gt_out):
                        all_matched = True
                        for i in range(len(prediction)):
                            if np.allclose(float(prediction[i]), float(gt_out[i])) == False:
                                all_matched = False
                                break
                        tmp_result = tmp.result or all_matched
                except Exception:
                    pass


            # handle floating point comparisons

            all_results.append(tmp_result)

            if not tmp_result:
                return all_results, {
                    "inputs": truncatefn(gt_inp),
                    "output": truncatefn(prediction),
                    "expected": truncatefn(gt_out),
                    "error_code": -2,
                    "error_message": "Wrong Answer",
                }
        except Exception as e:
            signal.alarm(0)
            if "timeoutexception" in repr(e).lower():
                all_results.append(-3)
                return all_results, {
                    "error": repr(e),
                    "error_code": -3,
                    "error_message": "Time Limit Exceeded",
                    "inputs": truncatefn(gt_inp),
                    "expected": truncatefn(gt_out),
                }
            else:
                all_results.append(-4)
                return all_results, {
                    "error": repr(e),
                    "error_code": -4,
                    "error_message": "Runtime Error",
                    "inputs": truncatefn(gt_inp),
                    "expected": truncatefn(gt_out),
                }

        finally:
            signal.alarm(0)
            faulthandler.disable()

    return all_results, {"execution time": total_execution}


def grade_stdio(
    code: str,
    all_inputs: list,
    all_outputs: list,
    timeout: int,
):
    ## runtime doesn't interact well with __name__ == '__main__'
    code = clean_if_name(code)

    ## we wrap the given code inside another function
    code = make_function(code)

    compiled_sol = compile_code(code, timeout)

    if compiled_sol is None:
        return

    method = get_function(compiled_sol, "wrapped_function")

    if method is None:
        return

    all_results = []
    total_execution_time = 0
    for idx, (gt_inp, gt_out) in enumerate(zip(all_inputs, all_outputs)):
        signal.alarm(timeout)
        faulthandler.enable()

        signal.alarm(timeout)
        with Capturing() as captured_output:
            try:
                start = time.time()
                call_method(method, gt_inp)
                total_execution_time += time.time() - start
                # reset the alarm
                signal.alarm(0)
            except Exception as e:
                signal.alarm(0)
                if "timeoutexception" in repr(e).lower():
                    all_results.append(-3)
                    return all_results, {
                        "error": repr(e),
                        "error_code": -3,
                        "error_message": "Time Limit Exceeded",
                        "inputs": truncatefn(gt_inp),
                        "expected": truncatefn(gt_out),
                    }
                else:
                    all_results.append(-4)
                    return all_results, {
                        "error": repr(e),
                        "error_code": -4,
                        "error_message": "Runtime Error",
                        "inputs": truncatefn(gt_inp),
                        "expected": truncatefn(gt_out),
                    }

            finally:
                signal.alarm(0)
                faulthandler.disable()

        prediction = captured_output[0]

        stripped_prediction_lines = get_stripped_lines(prediction)
        stripped_gt_out_lines = get_stripped_lines(gt_out)

        ## WA happens in multiple circumstances
        ## so cache the return to make it clean!
        WA_send_args = {
            "inputs": truncatefn(gt_inp),
            "output": truncatefn(prediction),
            "expected": truncatefn(gt_out),
            "error_code": -2,
        }

        if len(stripped_prediction_lines) != len(stripped_gt_out_lines):
            all_results.append(-2)
            WA_send_args["error_message"] = "Wrong answer: mismatched output length"
            return all_results, WA_send_args

        for output_line_idx, (
            stripped_prediction_line,
            stripped_gt_out_line,
        ) in enumerate(zip(stripped_prediction_lines, stripped_gt_out_lines)):
            WA_send_args["error_message"] = (
                f"Wrong answer at {output_line_idx=}: {truncatefn(stripped_prediction_line)} != {truncatefn(stripped_gt_out_line)}"
            )
            ## CASE 1: exact match
            if stripped_prediction_line == stripped_gt_out_line:
                continue

            ## CASE 2: element-wise comparision
            ## if there are floating elements
            ## use `decimal` library for good floating point comparision
            ## otherwise gotcha: np.isclose(50000000000000000, 50000000000000001) = True
            ## note that we should always be able to convert to decimals

            success, decimal_prediction_line = convert_line_to_decimals(
                stripped_prediction_line
            )

            if not success:
                all_results.append(-2)
                return all_results, WA_send_args

            success, decimal_gtout_line = convert_line_to_decimals(stripped_gt_out_line)

            if not success:
                all_results.append(-2)
                return all_results, WA_send_args

            if decimal_prediction_line == decimal_gtout_line:
                continue

            ok = False
            try:
                if len(decimal_prediction_line) == len(decimal_gtout_line) and np.allclose(decimal_prediction_line, decimal_gtout_line):
                    ok = True
            except Exception:
                pass

            if ok:
                continue
            all_results.append(-2)
            return all_results, WA_send_args
        all_results.append(True)

    return all_results, {"execution time": total_execution_time}


def run_test(problem_to_check, debug=False, timeout=6):
    """
    if test(generated_code) is not None it'll try to run the code.
    otherwise it'll just return an input and output pair.
    """
    signal.signal(signal.SIGALRM, timeout_handler)
    # Disable functionalities that can make destructive changes to the test.
    # max memory is set to 4GB
    reliability_guard()

    generation = problem_to_check['generation']

    generation = has_code(generation)

    if generation is None:
        return [-1], "No code found."

    if debug:
        print(f"start = {datetime.now().time()}")

    if problem_to_check['starter_code'] == '':
        which_type = CODE_TYPE.standard_input  # Standard input
        method_name = None
    else:
        starter = problem_to_check['starter_code']
        which_type = CODE_TYPE.call_based  # Call-based
        method_name = problem_to_check['starter_code']

    generation = post_process_code(generation).replace("\t", "    ").replace('if __name__ == "__main__":', 'if True:').replace("if __name__ == '__main__':", 'if True:')

    inputs, outputs = [], []
    for in_out in problem_to_check['input_output']:
        if os.path.exists(in_out['input']): # File based
            with open(in_out['input'], 'r', encoding='utf-8') as f:
                inputs.append(f.read().strip())
            with open(in_out['output'], 'r', encoding='utf-8') as f:
                outputs.append(f.read().strip())
        else: # plain-text based
            inputs.append(in_out['input'])
            outputs.append(in_out['output'])

    assert(len(inputs) == len(outputs))

    if debug:
        print(f"loaded {len(inputs)} Input-Output pairs = {datetime.now().time()}")

    results = []
    if debug:
        print(f"loading test code = {datetime.now().time()}")

    if which_type == CODE_TYPE.call_based:
        signal.alarm(timeout)
        try:
            results, metadata = grade_call_based(
                code=generation,
                all_inputs=inputs,
                all_outputs=outputs,
                fn_name=method_name,
                timeout=timeout,
            )
            return results, metadata
        except Exception as e:
            return [-4], f"Call-based Error as {e}"
        finally:
            signal.alarm(0)

    elif which_type == CODE_TYPE.standard_input:
        signal.alarm(timeout)
        try:
            results, metadata = grade_stdio(
                code=generation,
                all_inputs=inputs,
                all_outputs=outputs,
                timeout=timeout,
            )
            return results, metadata
        except Exception as e:
            return [-4], f"Stdin-based Error as {e}"
        finally:
            signal.alarm(0)


def reliability_guard(maximum_memory_bytes=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