Function test_oob_buffer

python/pyfory/tests/test_cross_language.py:663–705

(in_band_file_path, out_of_band_file_path)

@cross_language_test
def test_oob_buffer(in_band_file_path, out_of_band_file_path):
    with open(in_band_file_path, "rb") as f:
        in_band_bytes = f.read()
    with open(out_of_band_file_path, "rb") as f:
        out_of_band_buffer = pyfory.Buffer(f.read())
    fory = pyfory.Fory(xlang=True, compatible=False, ref=True)
    n_buffers = out_of_band_buffer.read_int32()
    buffers = []
    for i in range(n_buffers):
        length = out_of_band_buffer.read_int32()
        reader_index = out_of_band_buffer.get_reader_index()
        buffers.append(out_of_band_buffer.slice(reader_index, length))
        out_of_band_buffer.set_reader_index(out_of_band_buffer.get_reader_index() + length)
    new_obj = fory.deserialize(in_band_bytes, buffers)
    obj = [bytes(bytearray([0, 1])) for _ in range(10)]
    assert new_obj == obj, (obj, new_obj)

    buffer_objects = []
    counter = 0

    def buffer_callback(binary_object):
        nonlocal counter
        counter += 1
        if counter % 2 == 0:
            buffer_objects.append(binary_object)
            return False
        else:
            return True

    serialized = fory.serialize(obj, buffer_callback=buffer_callback)
    # in_band_bytes size may be different because it may contain language-specific meta.
    debug_print(f"{len(serialized), len(in_band_bytes)}")
    debug_print(f"deserialized from other language {new_obj}")
    debug_print(f"deserialized from python {fory.deserialize(serialized, [o.getbuffer() for o in buffer_objects])}")
    fory.deserialize(serialized, [o.getbuffer() for o in buffer_objects])
    with open(in_band_file_path, "wb+") as f:
        f.write(serialized)
    out_of_band_buffer.write_int32(len(buffer_objects))
    for buffer_object in buffer_objects:
        out_of_band_buffer.write_int32(buffer_object.total_bytes())
        buffer_object.write_to(out_of_band_buffer)
    with open(out_of_band_file_path, "wb+") as f:
        f.write(out_of_band_buffer.to_bytes(0, out_of_band_buffer.get_writer_index()))


@cross_language_test