()
| 608 | |
| 609 | |
| 610 | def test_pickle(): |
| 611 | buf = Buffer.allocate(32) |
| 612 | pickler = pickle.Pickler(buf) |
| 613 | pickler.dump(b"abc") |
| 614 | buf.write_int32(-1) |
| 615 | pickler.dump("abcd") |
| 616 | assert buf.get_writer_index() - 4 == len(pickle.dumps(b"abc")) + len(pickle.dumps("abcd")) |
| 617 | print(f"writer_index {buf.get_writer_index()}") |
| 618 | |
| 619 | bytes_io_ = io.BytesIO(buf) |
| 620 | unpickler = pickle.Unpickler(bytes_io_) |
| 621 | assert unpickler.load() == b"abc" |
| 622 | bytes_io_.seek(bytes_io_.tell() + 4) |
| 623 | assert unpickler.load() == "abcd" |
| 624 | print(f"reader_index {buf.get_reader_index()} {bytes_io_.tell()}") |
| 625 | |
| 626 | if pa: |
| 627 | pa_buf = pa.BufferReader(buf) |
| 628 | unpickler = pickle.Unpickler(pa_buf) |
| 629 | assert unpickler.load() == b"abc" |
| 630 | pa_buf.seek(pa_buf.tell() + 4) |
| 631 | assert unpickler.load() == "abcd" |
| 632 | print(f"reader_index {buf.get_reader_index()} {pa_buf.tell()} {buf.get_reader_index()}") |
| 633 | |
| 634 | unpickler = pickle.Unpickler(buf) |
| 635 | assert unpickler.load() == b"abc" |
| 636 | buf.set_reader_index(buf.get_reader_index() + 4) |
| 637 | assert unpickler.load() == "abcd" |
| 638 | print(f"reader_index {buf.get_reader_index()}") |
| 639 | |
| 640 | |
| 641 | @dataclass |
nothing calls this directly
no test coverage detected