aeg.raf: StreamStorage adapter; accept io.BytesIO directly

StreamStorage wraps any seek-based file-like object using the same
lock + seek/read/write pattern as FileStorage's fallback for platforms
without pread/pwrite. create()/open() auto-wrap io.BytesIO, so in-memory
use no longer requires the explicit BytesIOStorage wrapper. The caller
keeps ownership of the stream (Raf does not close it).
This commit is contained in:
LeoVasanko committed 2026-10-04 14:30:59 +00:00
1 parent f2a21856b3
commit 32954c4eba
3 files changed
+86 -3

No files matched your search

+20
View File
@@ -25,6 +25,26 @@ def test_roundtrip_bytesio(c):
assert f.read() == data
@pytest.mark.parametrize("c", CIPHERS, ids=cipher_id)
def test_roundtrip_io_bytesio(c):
"""A bare io.BytesIO is auto-wrapped in StreamStorage."""
import io
key = c.random_key()
data = bytes(range(256)) * 20
bio = io.BytesIO()
with raf.create(bio, key, c, chunk_size=CHUNK) as f:
f.write(data)
f.pwrite(b"XY", 1000) # unaligned write across a chunk boundary
data = data[:1000] + b"XY" + data[1002:]
f.truncate(2000)
with raf.open(bio, key, c) as f:
assert f.size == 2000
assert f.pread(6, 999) == data[999:1005]
assert f.read() == data[:2000]
bio.close()
@pytest.mark.parametrize("c", CIPHERS, ids=cipher_id)
def test_roundtrip_filestorage(c, tmp_path):
key = c.random_key()