%%% @doc Common Test suite for py_buffer API. %%% %%% Tests the zero-copy WSGI input buffer for streaming HTTP bodies. -module(py_buffer_SUITE). -include_lib("common_test/include/ct.hrl"). -export([ all/0, init_per_suite/1, end_per_suite/1, init_per_testcase/2, end_per_testcase/2 ]). -export([ create_buffer_test/1, create_buffer_with_size_test/1, write_read_test/1, readline_test/1, readlines_test/1, seek_tell_test/1, find_test/1, memoryview_test/1, iterator_test/1, closed_buffer_test/1, empty_buffer_test/1, pass_to_python_test/1, gc_refcount_test/1, nonblock_read_test/1, asyncio_read_test/1 ]). all() -> [ create_buffer_test, create_buffer_with_size_test, write_read_test, readline_test, readlines_test, seek_tell_test, find_test, memoryview_test, iterator_test, closed_buffer_test, empty_buffer_test, pass_to_python_test, gc_refcount_test, nonblock_read_test, asyncio_read_test ]. init_per_suite(Config) -> {ok, _} = application:ensure_all_started(erlang_python), {ok, _} = py:start_contexts(), Config. end_per_suite(_Config) -> ok = application:stop(erlang_python), ok. init_per_testcase(_TestCase, Config) -> Config. end_per_testcase(_TestCase, _Config) -> ok. %%% ============================================================================ %%% Test Cases %%% ============================================================================ %% @doc Test creating a buffer with default settings (chunked encoding) create_buffer_test(_Config) -> {ok, Buf} = py_buffer:new(), true = is_reference(Buf), ok = py_buffer:close(Buf). %% @doc Test creating a buffer with known content length create_buffer_with_size_test(_Config) -> {ok, Buf} = py_buffer:new(1024), true = is_reference(Buf), ok = py_buffer:close(Buf). %% @doc Test basic write and read cycle write_read_test(_Config) -> {ok, Buf} = py_buffer:new(), %% Write some data ok = py_buffer:write(Buf, <<"Hello, ">>), ok = py_buffer:write(Buf, <<"World!">>), ok = py_buffer:close(Buf), %% Read from Python Ctx = py:context(1), ok = py:exec(Ctx, <<"from erlang import PyBuffer">>), {ok, <<"Hello, World!">>} = py:eval(Ctx, <<"PyBuffer._test_create(b'Hello, World!').read()">>), ok. %% @doc Test readline method readline_test(_Config) -> Ctx = py:context(1), ok = py:exec(Ctx, <<"from erlang import PyBuffer">>), %% Create buffer with multiple lines and close it ok = py:exec(Ctx, <<" buf = PyBuffer._test_create(b'line1\\nline2\\nline3\\n') buf.close() ">>), {ok, Line1} = py:eval(Ctx, <<"buf.readline()">>), <<"line1\n">> = Line1, {ok, Line2} = py:eval(Ctx, <<"buf.readline()">>), <<"line2\n">> = Line2, {ok, Line3} = py:eval(Ctx, <<"buf.readline()">>), <<"line3\n">> = Line3, %% Empty at EOF {ok, <<>>} = py:eval(Ctx, <<"buf.readline()">>), ok. %% @doc Test readlines method readlines_test(_Config) -> Ctx = py:context(1), ok = py:exec(Ctx, <<"from erlang import PyBuffer">>), ok = py:exec(Ctx, <<" buf = PyBuffer._test_create(b'a\\nb\\nc\\n') buf.close() ">>), {ok, Lines} = py:eval(Ctx, <<"buf.readlines()">>), [<<"a\n">>, <<"b\n">>, <<"c\n">>] = Lines, ok. %% @doc Test seek and tell methods seek_tell_test(_Config) -> Ctx = py:context(1), ok = py:exec(Ctx, <<"from erlang import PyBuffer">>), %% Create buffer, close it, and check initial position ok = py:exec(Ctx, <<" buf = PyBuffer._test_create(b'0123456789') buf.close() ">>), {ok, 0} = py:eval(Ctx, <<"buf.tell()">>), %% Read 5 bytes, position should advance {ok, <<"01234">>} = py:eval(Ctx, <<"buf.read(5)">>), {ok, 5} = py:eval(Ctx, <<"buf.tell()">>), %% Seek back to beginning (SEEK_SET) {ok, 0} = py:eval(Ctx, <<"buf.seek(0)">>), {ok, 0} = py:eval(Ctx, <<"buf.tell()">>), %% Seek relative (SEEK_CUR) {ok, 3} = py:eval(Ctx, <<"buf.seek(3, 1)">>), {ok, 3} = py:eval(Ctx, <<"buf.tell()">>), %% Seek to position 7 {ok, 7} = py:eval(Ctx, <<"buf.seek(7)">>), {ok, <<"789">>} = py:eval(Ctx, <<"buf.read()">>), ok. %% @doc Test find method with memchr/memmem find_test(_Config) -> Ctx = py:context(1), ok = py:exec(Ctx, <<"from erlang import PyBuffer">>), %% Create buffer with test data ok = py:exec(Ctx, <<"buf = PyBuffer._test_create(b'hello world hello')">>), %% Find single byte (memchr path) {ok, 0} = py:eval(Ctx, <<"buf.find(b'h')">>), {ok, 4} = py:eval(Ctx, <<"buf.find(b'o')">>), %% Find substring (memmem path) {ok, 0} = py:eval(Ctx, <<"buf.find(b'hello')">>), {ok, 6} = py:eval(Ctx, <<"buf.find(b'world')">>), %% Find with start position {ok, 12} = py:eval(Ctx, <<"buf.find(b'hello', 1)">>), %% Not found {ok, -1} = py:eval(Ctx, <<"buf.find(b'xyz')">>), %% Empty substring returns start {ok, 0} = py:eval(Ctx, <<"buf.find(b'')">>), {ok, 5} = py:eval(Ctx, <<"buf.find(b'', 5)">>), ok. %% @doc Test buffer protocol (memoryview) memoryview_test(_Config) -> Ctx = py:context(1), ok = py:exec(Ctx, <<"from erlang import PyBuffer">>), %% Create buffer and get memoryview ok = py:exec(Ctx, <<" buf = PyBuffer._test_create(b'test data') mv = memoryview(buf) ">>), %% Check memoryview properties {ok, 9} = py:eval(Ctx, <<"len(mv)">>), {ok, true} = py:eval(Ctx, <<"mv.readonly">>), {ok, 1} = py:eval(Ctx, <<"mv.ndim">>), %% Access bytes via memoryview (zero-copy) {ok, <<"test data">>} = py:eval(Ctx, <<"bytes(mv)">>), %% Slice access {ok, <<"test">>} = py:eval(Ctx, <<"bytes(mv[:4])">>), {ok, <<"data">>} = py:eval(Ctx, <<"bytes(mv[5:])">>), %% Release memoryview ok = py:exec(Ctx, <<"mv.release()">>), ok. %% @doc Test iterator protocol for line-by-line reading iterator_test(_Config) -> Ctx = py:context(1), ok = py:exec(Ctx, <<"from erlang import PyBuffer">>), %% Create buffer and iterate ok = py:exec(Ctx, <<" buf = PyBuffer._test_create(b'line1\\nline2\\nline3\\n') buf.close() ">>), {ok, Lines} = py:eval(Ctx, <<"list(buf)">>), [<<"line1\n">>, <<"line2\n">>, <<"line3\n">>] = Lines, ok. %% @doc Test operations on closed buffer closed_buffer_test(_Config) -> Ctx = py:context(1), ok = py:exec(Ctx, <<"from erlang import PyBuffer">>), %% Create buffer and close it ok = py:exec(Ctx, <<" buf = PyBuffer._test_create(b'data') buf.close() ">>), %% Check closed property {ok, true} = py:eval(Ctx, <<"buf.closed">>), %% Reading should return remaining data then empty {ok, <<"data">>} = py:eval(Ctx, <<"buf.read()">>), {ok, <<>>} = py:eval(Ctx, <<"buf.read()">>), ok. %% @doc Test empty buffer behavior empty_buffer_test(_Config) -> Ctx = py:context(1), ok = py:exec(Ctx, <<"from erlang import PyBuffer">>), %% Create empty buffer and close immediately ok = py:exec(Ctx, <<" buf = PyBuffer._test_create(b'') buf.close() ">>), %% Read should return empty {ok, <<>>} = py:eval(Ctx, <<"buf.read()">>), {ok, <<>>} = py:eval(Ctx, <<"buf.readline()">>), {ok, []} = py:eval(Ctx, <<"buf.readlines()">>), %% Length should be 0 {ok, 0} = py:eval(Ctx, <<"len(buf)">>), %% File-like properties {ok, true} = py:eval(Ctx, <<"buf.readable()">>), {ok, false} = py:eval(Ctx, <<"buf.writable()">>), {ok, true} = py:eval(Ctx, <<"buf.seekable()">>), ok. %% @doc Test passing buffer ref to Python - auto-conversion via py_convert.c pass_to_python_test(_Config) -> {ok, Buf} = py_buffer:new(), %% Write data from Erlang ok = py_buffer:write(Buf, <<"chunk1">>), ok = py_buffer:write(Buf, <<"chunk2">>), ok = py_buffer:close(Buf), %% Pass buffer to Python via py:eval - should auto-convert to PyBuffer Ctx = py:context(1), %% Define a function that reads from a buffer ok = py:exec(Ctx, <<" def read_buffer(buf): return buf.read() ">>), %% Call with buffer ref - py_convert.c should wrap it as PyBuffer {ok, <<"chunk1chunk2">>} = py:eval(Ctx, <<"read_buffer(buf)">>, #{<<"buf">> => Buf}), ok. %% @doc Test that buffer resources are properly garbage collected %% Verifies reference counting between Erlang and Python gc_refcount_test(_Config) -> Ctx = py:context(1), ok = py:exec(Ctx, <<" import gc from erlang import PyBuffer # Track if we can create many buffers without memory issues def create_and_release_buffers(count): '''Create many buffers and let them be GC'd.''' for i in range(count): buf = PyBuffer._test_create(b'x' * 1000) buf.close() # buf goes out of scope, should be released gc.collect() return True def test_memoryview_refcount(): '''Test that memoryview keeps buffer alive.''' buf = PyBuffer._test_create(b'test data') mv = memoryview(buf) # Buffer should stay alive while memoryview exists data = bytes(mv[:4]) # Release memoryview mv.release() # Buffer should still be usable after memoryview release buf.close() remaining = buf.read() return data == b'test' and remaining == b'test data' def test_multiple_views(): '''Test multiple memoryviews on same buffer.''' buf = PyBuffer._test_create(b'hello world') mv1 = memoryview(buf) mv2 = memoryview(buf) result1 = bytes(mv1[:5]) result2 = bytes(mv2[6:]) mv1.release() mv2.release() buf.close() return result1 == b'hello' and result2 == b'world' ">>), %% Test 1: Create and release many buffers {ok, true} = py:eval(Ctx, <<"create_and_release_buffers(100)">>), %% Test 2: Memoryview reference counting {ok, true} = py:eval(Ctx, <<"test_memoryview_refcount()">>), %% Test 3: Multiple memoryviews {ok, true} = py:eval(Ctx, <<"test_multiple_views()">>), %% Test 4: Erlang-side reference counting %% Create buffer, pass to Python, let Erlang ref go out of scope {ok, Data} = begin {ok, TempBuf} = py_buffer:new(), ok = py_buffer:write(TempBuf, <<"erlang data">>), ok = py_buffer:close(TempBuf), %% Pass to Python - Python now holds a reference py:eval(Ctx, <<"buf.read()">>, #{<<"buf">> => TempBuf}) %% TempBuf goes out of scope here but Python read it first end, <<"erlang data">> = Data, %% Force Erlang GC erlang:garbage_collect(), %% Test 5: Verify no crashes after GC {ok, true} = py:eval(Ctx, <<"True">>), ok. %% @doc Test non-blocking read methods for async I/O nonblock_read_test(_Config) -> Ctx = py:context(1), ok = py:exec(Ctx, <<"from erlang import PyBuffer">>), %% Test read_nonblock returns available data immediately ok = py:exec(Ctx, <<" buf = PyBuffer._test_create(b'hello world') ">>), %% readable_amount should return 11 (length of 'hello world') {ok, 11} = py:eval(Ctx, <<"buf.readable_amount()">>), %% at_eof should be False (buffer not closed yet) {ok, false} = py:eval(Ctx, <<"buf.at_eof()">>), %% read_nonblock should return available data {ok, <<"hello">>} = py:eval(Ctx, <<"buf.read_nonblock(5)">>), %% readable_amount should now be 6 {ok, 6} = py:eval(Ctx, <<"buf.readable_amount()">>), %% read_nonblock with no size returns all remaining {ok, <<" world">>} = py:eval(Ctx, <<"buf.read_nonblock()">>), %% readable_amount should be 0 now {ok, 0} = py:eval(Ctx, <<"buf.readable_amount()">>), %% read_nonblock on empty buffer returns empty bytes (not blocking) {ok, <<>>} = py:eval(Ctx, <<"buf.read_nonblock()">>), %% Close buffer and check at_eof ok = py:exec(Ctx, <<"buf.close()">>), {ok, true} = py:eval(Ctx, <<"buf.at_eof()">>), %% Test async I/O pattern simulation ok = py:exec(Ctx, <<" def async_read_simulation(): '''Simulate async I/O read pattern.''' buf = PyBuffer._test_create(b'chunk1chunk2chunk3') buf.close() # EOF chunks = [] while not buf.at_eof(): available = buf.readable_amount() if available > 0: # Read in chunks of 6 chunk = buf.read_nonblock(6) chunks.append(chunk) else: # Would yield to event loop here in real async code break return chunks ">>), {ok, [<<"chunk1">>, <<"chunk2">>, <<"chunk3">>]} = py:eval(Ctx, <<"async_read_simulation()">>), ok. %% @doc Test PyBuffer with asyncio - Erlang fills buffer while Python reads asyncio_read_test(_Config) -> %% Create a buffer that Erlang will fill {ok, Buf} = py_buffer:new(), Ctx = py:context(1), Self = self(), %% Define async reader in Python ok = py:exec(Ctx, <<" import asyncio async def async_buffer_reader(buf): '''Read from buffer asynchronously as Erlang fills it.''' chunks = [] while not buf.at_eof(): available = buf.readable_amount() if available > 0: chunk = buf.read_nonblock(available) chunks.append(chunk) else: # Yield to event loop - Erlang is still writing await asyncio.sleep(0.005) return b''.join(chunks) def run_async_reader(buf): '''Run async reader.''' loop = asyncio.new_event_loop() try: return loop.run_until_complete(async_buffer_reader(buf)) finally: loop.close() ">>), %% Spawn a process to write data with delays (simulating streaming) spawn_link(fun() -> timer:sleep(10), ok = py_buffer:write(Buf, <<"chunk1:">>), timer:sleep(20), ok = py_buffer:write(Buf, <<"chunk2:">>), timer:sleep(20), ok = py_buffer:write(Buf, <<"chunk3">>), timer:sleep(10), ok = py_buffer:close(Buf), Self ! writer_done end), %% Read asynchronously while Erlang writes {ok, Result} = py:eval(Ctx, <<"run_async_reader(buf)">>, #{<<"buf">> => Buf}), %% Wait for writer to finish receive writer_done -> ok after 1000 -> ok end, %% Verify we got all the data <<"chunk1:chunk2:chunk3">> = Result, ok.