`testing_util.py` — 这是代码评测的底层执行引擎¶

模块路径: verl.utils.reward_score.prime_code.testing_util

文件概述¶

这是代码评测的底层执行引擎，实现了在受限环境中运行生成代码并比较输出的逻辑。它支持两种代码类型：函数调用型（call-based）和标准输入型（standard_input），并包含一套安全机制（reliability_guard）防止恶意代码。

这是整个 prime_code 子包中最长、最复杂的文件。

关键代码讲解¶

1. 代码类型枚举¶

class CODE_TYPE(Enum):
    call_based = 0       # 通过函数调用测试（如 LeetCode 风格）
    standard_input = 1   # 通过标准输入/输出测试（如竞赛编程风格）

2. 标准输出捕获器¶

class Capturing(list):
    """捕获 sys.stdout 的输出"""
    def __enter__(self):
        self._stdout = sys.stdout
        sys.stdout = self._stringio = StringIO()
        self._stringio.close = lambda x: 1  # 防止代码关闭 stdout
        return self
    def __exit__(self, *args):
        self.append(self._stringio.getvalue())
        sys.stdout = self._stdout

3. 核心测试函数 `run_test`¶

函数调用型的测试流程：

if which_type == CODE_TYPE.call_based:
    sol += test  # 拼接用户代码
    tmp_sol = RuntimeModule.from_string("tmp_sol", "", sol)
    method = getattr(tmp_sol, method_name)
    # 对每个测试用例
    for index, inputs in enumerate(in_outs["inputs"]):
        inputs = [json.loads(line) for line in inputs.split("\n")]
        output = method(*inputs)          # 调用函数
        tmp_result = output == in_outs["outputs"][index]  # 比较输出

标准输入型的测试流程更复杂，需要将代码包装成函数并 mock 标准输入：

elif which_type == CODE_TYPE.standard_input:
    # 将代码包装成 code() 函数
    new_test = "def code():\n" + indented_code
    # Mock 标准输入
    with Capturing() as output:
        call_method(method, inputs)
    # 比较输出（多种比较策略）

4. 输出比较策略¶

代码对输出的比较非常宽容，有多层回退策略： 1. 直接字符串比较 2. 去掉首尾空格后比较 3. 按行分割后比较 4. 转为浮点数用 np.allclose 比较 5. 按单词分割后用集合比较

5. 安全保护 `reliability_guard`¶

def reliability_guard(maximum_memory_bytes=None):
    """禁用危险的系统函数"""
    faulthandler.disable()
    builtins.exit = None
    builtins.quit = None
    os.kill = None
    os.system = None
    os.remove = None
    os.fork = None
    shutil.rmtree = None
    subprocess.Popen = None
    # ... 禁用大量危险函数

这个函数禁用了大量危险的系统操作（如删除文件、创建进程、修改权限等），防止生成的代码对系统造成破坏。注意这不是安全沙箱，只是减少风险。

6. 输入 Mock¶

def call_method(method, inputs):
    @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.read", lambda *args: inputs)
    def _inner_call_method(_method):
        return _method()
    return _inner_call_method(method)

使用 unittest.mock.patch 模拟标准输入和文件读取，使代码能从"虚拟"输入中读取数据。

核心类/函数列表¶

类/函数名	作用
`run_test`	核心测试执行函数
`CODE_TYPE`	代码类型枚举
`Capturing`	stdout 捕获器
`call_method`	带输入 mock 的函数调用
`reliability_guard`	安全保护（禁用危险操作）
`custom_compare_`	宽容的输出比较
`truncatefn`	截断长字符串用于调试

与其他模块的关系¶

被 prime_code/utils.py 的 check_correctness 在子进程中调用
使用 pyext.RuntimeModule 动态编译执行代码
使用 numpy.allclose 做浮点数近似比较

小结¶

这是一个功能完整但代码量较大的代码执行引擎。它的设计面临两个核心挑战：(1) 如何安全地执行不可信的代码 -- 通过进程隔离 + reliability_guard；(2) 如何宽容地比较输出 -- 通过多层回退的比较策略。标准输入型代码的处理尤其复杂，需要动态改造代码结构并 mock 输入源。

testing_util.py — 这是代码评测的底层执行引擎¶