()
| 375 | |
| 376 | |
| 377 | def test_stream_buffer_read(): |
| 378 | writer = Buffer.allocate(32) |
| 379 | writer.write_uint32(0x01020304) |
| 380 | writer.write_int64(-1234567890) |
| 381 | writer.write_var_uint32(300) |
| 382 | writer.write_varint64(-4567890123) |
| 383 | writer.write_tagged_uint64(0x123456789) |
| 384 | writer.write_var_uint64(0x1FFFF) |
| 385 | writer.write_bytes_and_size(b"stream-data") |
| 386 | writer.write_string("hello-stream") |
| 387 | |
| 388 | data = writer.get_bytes(0, writer.get_writer_index()) |
| 389 | stream = OneByteStream(data) |
| 390 | reader = Buffer.from_stream(stream) |
| 391 | |
| 392 | assert reader.read_uint32() == 0x01020304 |
| 393 | assert reader.read_int64() == -1234567890 |
| 394 | assert reader.read_var_uint32() == 300 |
| 395 | assert reader.read_varint64() == -4567890123 |
| 396 | assert reader.read_tagged_uint64() == 0x123456789 |
| 397 | assert reader.read_var_uint64() == 0x1FFFF |
| 398 | assert reader.read_bytes_and_size() == b"stream-data" |
| 399 | assert reader.read_string() == "hello-stream" |
| 400 | |
| 401 | |
| 402 | def test_stream_buffer_read_with_recv_into(): |
nothing calls this directly
no test coverage detected