mirror of
https://github.com/RustPython/RustPython.git
synced 2026-10-28 18:59:04 +08:00
Validate TextIOWrapper chunk size (#8265)
* Validate TextIOWrapper chunk size Match CPython for non-positive, oversized, and index-compatible values. Assisted-by: Codex:GPT-5 * Use explicit AssertionError in io snippet Assisted-by: Codex:GPT-5 * Report original type in chunk size overflow Assisted-by: Codex:GPT-5 * Chain io snippet assertion error Assisted-by: Codex:GPT-5 * Match CPython chunk size setter behavior Assisted-by: Codex:GPT-5 * Format io snippet assertion Assisted-by: Codex:GPT-5.6 * Match CPython type name truncation Assisted-by: Codex:GPT-5 * Format non-ASCII chunk size test Assisted-by: Codex:GPT-5
This commit is contained in:
1 parent
abc35e5f83
commit
20cb866c67
2 files changed
+151
-12
No files matched your search
+38
-12
@@ -3206,19 +3206,45 @@ mod _io {
|
||||
}
|
||||
|
||||
#[pygetset(setter, name = "_CHUNK_SIZE")]
|
||||
fn set_chunksize(
|
||||
&self,
|
||||
chunk_size: PySetterValue<usize>,
|
||||
vm: &VirtualMachine,
|
||||
) -> PyResult<()> {
|
||||
let mut textio = self.lock(vm)?;
|
||||
match chunk_size {
|
||||
PySetterValue::Assign(chunk_size) => textio.chunk_size = chunk_size,
|
||||
PySetterValue::Delete => Err(vm.new_attribute_error("cannot delete attribute"))?,
|
||||
fn set_chunksize(&self, value: PySetterValue, vm: &VirtualMachine) -> PyResult<()> {
|
||||
{
|
||||
let textio = self.lock(vm)?;
|
||||
if vm.is_none(&textio.buffer) {
|
||||
return Err(vm.new_value_error("underlying buffer has been detached"));
|
||||
}
|
||||
}
|
||||
|
||||
let chunk_size: isize = match value {
|
||||
PySetterValue::Assign(object_value) => {
|
||||
let integer = object_value.try_index(vm)?;
|
||||
|
||||
integer.try_to_primitive::<isize>(vm).map_err(|_| {
|
||||
let class = object_value.class();
|
||||
let type_name = class.name();
|
||||
let mut end = type_name.len().min(200);
|
||||
while !type_name.is_char_boundary(end) {
|
||||
end -= 1;
|
||||
}
|
||||
vm.new_value_error(format!(
|
||||
"cannot fit '{}' into an index-sized integer",
|
||||
&type_name[..end]
|
||||
))
|
||||
})?
|
||||
}
|
||||
PySetterValue::Delete => {
|
||||
return Err(vm.new_attribute_error("cannot delete attribute"));
|
||||
}
|
||||
};
|
||||
// TODO: RUSTPYTHON
|
||||
// Change chunk_size type, validate it manually and throws ValueError if invalid.
|
||||
// https://github.com/python/cpython/blob/2e9da8e3522764d09f1d6054a2be567e91a30812/Modules/_io/textio.c#L3124-L3143
|
||||
|
||||
if chunk_size <= 0 {
|
||||
return Err(vm.new_value_error("a strictly positive integer is required"));
|
||||
}
|
||||
|
||||
let chunk_size = usize::try_from(chunk_size)
|
||||
.map_err(|_| vm.new_value_error("a strictly positive integer is required"))?;
|
||||
|
||||
let mut textio = self.lock(vm)?;
|
||||
textio.chunk_size = chunk_size;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
|
||||
@@ -84,3 +84,116 @@ textio = TextIOWrapper(raw, encoding="utf-8", write_through=True)
|
||||
raw.textio = textio
|
||||
with assert_raises(AttributeError):
|
||||
textio.writelines(["x"])
|
||||
|
||||
textio = TextIOWrapper(BytesIO())
|
||||
|
||||
for invalid_chunk_size in (0, -1, 2**100):
|
||||
try:
|
||||
textio._CHUNK_SIZE = invalid_chunk_size
|
||||
except ValueError:
|
||||
pass
|
||||
else:
|
||||
raise AssertionError(f"expected ValueError for {invalid_chunk_size!r}")
|
||||
|
||||
for invalid_chunk_size in (1.5, "4"):
|
||||
try:
|
||||
textio._CHUNK_SIZE = invalid_chunk_size
|
||||
except TypeError:
|
||||
pass
|
||||
else:
|
||||
raise AssertionError(f"expected TypeError for {invalid_chunk_size!r}")
|
||||
|
||||
|
||||
class ChunkSize:
|
||||
def __index__(self):
|
||||
return 16
|
||||
|
||||
|
||||
textio._CHUNK_SIZE = ChunkSize()
|
||||
assert textio._CHUNK_SIZE == 16
|
||||
|
||||
|
||||
class OversizedChunkSize:
|
||||
def __index__(self):
|
||||
return 2**100
|
||||
|
||||
|
||||
try:
|
||||
textio._CHUNK_SIZE = OversizedChunkSize()
|
||||
except ValueError as error:
|
||||
expected = "cannot fit 'OversizedChunkSize' into an index-sized integer"
|
||||
if str(error) != expected:
|
||||
raise AssertionError(f"unexpected error message: {error}") from error
|
||||
else:
|
||||
raise AssertionError("expected ValueError for oversized indexable object")
|
||||
|
||||
|
||||
def expect_value_error(expected, operation):
|
||||
try:
|
||||
operation()
|
||||
except ValueError as error:
|
||||
if str(error) != expected:
|
||||
raise AssertionError(f"unexpected error message: {error}") from error
|
||||
else:
|
||||
raise AssertionError(f"expected ValueError: {expected}")
|
||||
|
||||
|
||||
class UninitializedChunkSize:
|
||||
def __init__(self):
|
||||
self.called = False
|
||||
|
||||
def __index__(self):
|
||||
self.called = True
|
||||
return 16
|
||||
|
||||
|
||||
uninitialized_textio = TextIOWrapper.__new__(TextIOWrapper)
|
||||
uninitialized_chunk_size = UninitializedChunkSize()
|
||||
expect_value_error(
|
||||
"I/O operation on uninitialized object",
|
||||
lambda: setattr(uninitialized_textio, "_CHUNK_SIZE", uninitialized_chunk_size),
|
||||
)
|
||||
|
||||
if uninitialized_chunk_size.called:
|
||||
raise AssertionError(
|
||||
"__index__ should not be called for uninitialized TextIOWrapper"
|
||||
)
|
||||
|
||||
|
||||
detached_textio = TextIOWrapper(BytesIO())
|
||||
detached_textio.detach()
|
||||
expect_value_error(
|
||||
"underlying buffer has been detached",
|
||||
lambda: setattr(detached_textio, "_CHUNK_SIZE", 16),
|
||||
)
|
||||
expect_value_error(
|
||||
"underlying buffer has been detached",
|
||||
lambda: delattr(detached_textio, "_CHUNK_SIZE"),
|
||||
)
|
||||
|
||||
|
||||
long_type_name = "X" * 250
|
||||
LongNamedChunkSize = type(
|
||||
long_type_name,
|
||||
(),
|
||||
{"__index__": lambda self: 2**100},
|
||||
)
|
||||
expect_value_error(
|
||||
f"cannot fit '{long_type_name[:200]}' into an index-sized integer",
|
||||
lambda: setattr(textio, "_CHUNK_SIZE", LongNamedChunkSize()),
|
||||
)
|
||||
|
||||
|
||||
non_ascii_type_name = "é" * 250
|
||||
NonAsciiNamedChunkSize = type(
|
||||
non_ascii_type_name,
|
||||
(),
|
||||
{"__index__": lambda self: 2**100},
|
||||
)
|
||||
truncated_non_ascii_type_name = non_ascii_type_name.encode("utf-8")[:200].decode(
|
||||
"utf-8"
|
||||
)
|
||||
expect_value_error(
|
||||
f"cannot fit '{truncated_non_ascii_type_name}' into an index-sized integer",
|
||||
lambda: setattr(textio, "_CHUNK_SIZE", NonAsciiNamedChunkSize()),
|
||||
)
|
||||
Reference in new issue
Block a user