Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
34 changes: 33 additions & 1 deletion ais_bench/benchmark/datasets/livecodebench/testing_util.py
Original file line number Diff line number Diff line change
Expand Up @@ -63,11 +63,43 @@ def timeout_handler(signum, frame):
# used to capture stdout as a list
# from https://stackoverflow.com/a/16571630/6416660
# alternative use redirect_stdout() from contextlib
class _StdoutWithBuffer(StringIO):
"""StringIO that also exposes a binary ``.buffer`` view, like real stdout.

Generated solutions often use ``sys.stdout.buffer.write(...)`` for fast
output. A plain StringIO has no ``buffer`` attribute, so such solutions
used to fail with an AttributeError and were counted as runtime errors.
The bytes written through ``.buffer`` are decoded and forwarded to this
same StringIO, so ``getvalue()`` still returns the complete output.
"""

class _Buffer:

def __init__(self, owner):
self._owner = owner

def write(self, data):
if isinstance(data, (bytes, bytearray)):
data = data.decode()

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[review] data.decode() 使用默认 UTF-8 且无 errors 参数:真实 stdout.buffer 可写入任意字节,而这里遇到非 UTF-8 字节(如 latin-1、部分二进制串)会抛 UnicodeDecodeError;另外若一次多字节字符被拆成两次 write(分批 flush 的写法),第二次解码也会失败。这些异常仍会被 run_test 捕获成 Runtime Error,属于本次要修的问题的残留。建议改为 data.decode('utf-8', errors='replace')(或 'surrogateescape'),或按字节缓冲、只在 getvalue/比较阶段统一解码。

return self._owner.write(data)

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[review] 返回值语义与真实 sys.stdout.buffer 不一致:真实 file/buffer 对象的 write() 返回「写入的字节数」,这里返回的是 StringIO.write 的「字符数」。对多字节 UTF-8 输出两者不等,若生成代码用 n = sys.stdout.buffer.write(data) 或断言写入长度(这类写法在竞赛代码里存在),仍会被判错,偏离「不再误判」的目标。建议在解码前记录字节长度并返回它,例如:raw = data if isinstance(data, (bytes, bytearray)) else data.encode(); self._owner.write(raw.decode()); return len(raw)。


def writelines(self, lines):
for line in lines:
self.write(line)

def flush(self):

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[review] _Buffer 只实现了 write/writelines/flush,缺少 close/isatty/fileno 等常见接口,且没有 getattr 兜底;生成代码里出现 sys.stdout.buffer.close()、sys.stdout.buffer.isatty()、fileno() 时依旧 AttributeError → 仍被记成 Runtime Error。建议补上 close()(no-op)、isatty() -> False,或加 __getattr__ 委托给 owner,使伪 buffer 更接近真实对象。顺带一提:同一 hunk 里的 self._stringio.close = lambda x: 1 需要一个位置参数,调用 sys.stdout.close()(真实 stdout 无参)会抛 TypeError,属既有代码,但既然在补 stdin/stdout 伪装,建议一并改成 lambda *a, **kw: 1。

pass

def __init__(self):
super().__init__()
self.buffer = _StdoutWithBuffer._Buffer(self)


class Capturing(list):

def __enter__(self):
self._stdout = sys.stdout
sys.stdout = self._stringio = StringIO()
sys.stdout = self._stringio = _StdoutWithBuffer()
# Make closing the StringIO a no-op
self._stringio.close = lambda x: 1
return self
Expand Down
56 changes: 56 additions & 0 deletions tests/UT/datasets/livecodebench/test_livecodebench.py
Original file line number Diff line number Diff line change
Expand Up @@ -427,6 +427,62 @@ def test_load_with_cot(self):
self.assertIn('train', result)


class TestStdIOBuffer(unittest.TestCase):
"""测试标准输入/输出的二进制 .buffer 视图

竞赛类题解常用 sys.stdin.buffer.read() / sys.stdout.buffer.write() 做快速
读写,评测时伪造的 stdin/stdout 必须提供 .buffer,否则这类题解会抛
AttributeError 被误判为 Runtime Error。
"""

@classmethod
def setUpClass(cls):
try:
from ais_bench.benchmark.datasets.livecodebench import testing_util
cls.testing_util = testing_util
except ImportError:
raise unittest.SkipTest('testing_util not available')

def _run(self, source, inputs):
namespace = {}
exec(compile(source, '<generated>', 'exec'), namespace)
with self.testing_util.Capturing() as output:
self.testing_util.call_method(namespace['main'], inputs)
return ''.join(output).strip()

def test_stdin_buffer_read(self):
"""sys.stdin.buffer.read() 应读到测试输入"""
source = ('import sys\n'
'def main():\n'
' data = sys.stdin.buffer.read().split()\n'
' print(sum(map(int, data)))\n')
self.assertEqual(self._run(source, '1 2\n3\n'), '6')

def test_stdin_buffer_readline(self):

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[review] 新增的 stdin 用例与既有覆盖重复且没有覆盖上游已知缺陷:call_method 在本 PR 前就已支持 sys.stdin.buffer(见同目录 test_testing_util.py 中未改动的 test_call_method_supports_stdin_buffer / test_call_method_shares_text_and_binary_stream_position),本次改动只作用于 stdout,所以 test_stdin_buffer_read / test_stdin_buffer_readline 属于既有行为的重复验证。更值得补的是上游 issue #160 指出的那个坑:官方 MockBuffer.readline() 永远返回第一行,连续两次 sys.stdin.buffer.readline() 会拿到同一行导致 Wrong Answer(如 abc391_f)。建议把这两个用例合并/精简,并新增一条连续两次 readline 应返回两行、以及 text(input()) 与 binary 读取共享位置的多行用例,才能真正防住这类误判。

"""sys.stdin.buffer.readline() 应按行读到测试输入"""
source = ('import sys\n'
'def main():\n'
' n = int(sys.stdin.buffer.readline())\n'
' print(n * 2)\n')
self.assertEqual(self._run(source, '21\n'), '42')

def test_stdout_buffer_write(self):
"""sys.stdout.buffer.write() 的输出应被 Capturing 收集到"""
source = ('import sys\n'
'def main():\n'
' sys.stdout.buffer.write(b"42\\n")\n')
self.assertEqual(self._run(source, ''), '42')

def test_mixed_text_and_binary_output(self):
"""混用 print 和 sys.stdout.buffer.write 时输出顺序应保持一致"""
source = ('import sys\n'
'def main():\n'
' print("a")\n'
' sys.stdout.buffer.write(b"b\\n")\n'
' print("c")\n')
self.assertEqual(self._run(source, ''), 'a\nb\nc')


if __name__ == '__main__':
unittest.main()

Loading