Repository navigation
fix(livecodebench): expose .buffer on mocked stdin/stdout #518
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
base: master
Are you sure you want to change the base?
Changes from all commits
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -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() | ||
| return self._owner.write(data) | ||
|
Collaborator
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. [review] 返回值语义与真实 sys.stdout.buffer 不一致:真实 file/buffer 对象的 write() 返回「写入的字节数」,这里返回的是 StringIO.write 的「字符数」。对多字节 UTF-8 输出两者不等,若生成代码用 |
||
|
|
||
| def writelines(self, lines): | ||
| for line in lines: | ||
| self.write(line) | ||
|
|
||
| def flush(self): | ||
|
Collaborator
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. [review] _Buffer 只实现了 write/writelines/flush,缺少 close/isatty/fileno 等常见接口,且没有 getattr 兜底;生成代码里出现 |
||
| 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 | ||
|
|
||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -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): | ||
|
Collaborator
There was a problem hiding this comment. Choose a reason for hiding this commentThe 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() 应按行读到测试输入""" | ||
| 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() | ||
|
|
||
There was a problem hiding this comment.
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/比较阶段统一解码。