MCPcopy Index your code
hub / github.com/apache/fory / test_buffer_flush_stream

Function test_buffer_flush_stream

python/pyfory/tests/test_buffer.py:294–303  ·  view source on GitHub ↗
()

Source from the content-addressed store, hash-verified

292
293
294def test_buffer_flush_stream():
295 stream = PartialWriteStream()
296 buffer = Buffer.allocate(16)
297 output_stream = Buffer.wrap_output_stream(stream)
298 buffer.bind_output_stream(output_stream)
299 payload = b"stream-flush-buffer"
300 buffer.write_bytes(payload)
301 output_stream.force_flush()
302 assert stream.to_bytes() == payload
303 assert buffer.get_writer_index() == 0
304
305
306def test_wrap_output_stream_invalid_target_raises():

Callers

nothing calls this directly

Calls 6

to_bytesMethod · 0.95
PartialWriteStreamClass · 0.85
bind_output_streamMethod · 0.80
allocateMethod · 0.65
write_bytesMethod · 0.45
force_flushMethod · 0.45

Tested by

no test coverage detected