%%% @doc Common Test suite for py_byte_channel API. %%% %%% Tests the raw byte channel API for Erlang-Python communication. -module(py_byte_channel_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_byte_channel_test/1, create_byte_channel_with_max_size_test/1, erlang_send_bytes_test/1, erlang_receive_bytes_test/1, send_receive_multiple_bytes_test/1, try_receive_empty_bytes_test/1, close_byte_channel_test/1, byte_channel_info_test/1, backpressure_bytes_test/1, send_bytes_to_closed_test/1, %% Python tests python_byte_channel_class_test/1, python_byte_channel_send_bytes_test/1, python_byte_channel_receive_bytes_test/1, %% Sync blocking receive tests sync_receive_bytes_immediate_test/1, sync_receive_bytes_wait_test/1, sync_receive_bytes_closed_test/1, %% Large payload test large_payload_bytes_test/1, %% Async event loop dispatch test async_receive_bytes_e2e_test/1, %% create_task + async receive test create_task_async_receive_test/1, %% Close and drain tests close_drain_bytes_erlang_test/1, close_drain_bytes_python_sync_test/1, close_drain_bytes_python_async_test/1, close_drain_bytes_create_task_async_test/1 ]). all() -> [ create_byte_channel_test, create_byte_channel_with_max_size_test, erlang_send_bytes_test, erlang_receive_bytes_test, send_receive_multiple_bytes_test, try_receive_empty_bytes_test, close_byte_channel_test, byte_channel_info_test, backpressure_bytes_test, send_bytes_to_closed_test, %% Python tests python_byte_channel_class_test, python_byte_channel_send_bytes_test, python_byte_channel_receive_bytes_test, %% Sync blocking receive tests sync_receive_bytes_immediate_test, sync_receive_bytes_wait_test, sync_receive_bytes_closed_test, %% Large payload test large_payload_bytes_test, %% Async event loop dispatch test async_receive_bytes_e2e_test, %% create_task + async receive test create_task_async_receive_test, %% Close and drain tests close_drain_bytes_erlang_test, close_drain_bytes_python_sync_test, close_drain_bytes_python_async_test, close_drain_bytes_create_task_async_test ]. init_per_suite(Config) -> {ok, _} = application:ensure_all_started(erlang_python), {ok, _} = py:start_contexts(), %% Register byte channel callbacks ok = py_byte_channel:register_callbacks(), Config. end_per_suite(_Config) -> ok = application:stop(erlang_python), ok. init_per_testcase(async_receive_bytes_e2e_test, Config) -> %% Define the async receive helper function Ctx = py:context(1), ok = py:exec(Ctx, <<" import erlang from erlang import ByteChannel async def receive_bytes(ch_ref): ch = ByteChannel(ch_ref) return await ch.async_receive_bytes() ">>), Config; init_per_testcase(_TestCase, Config) -> Config. end_per_testcase(_TestCase, _Config) -> ok. %%% ============================================================================ %%% Test Cases %%% ============================================================================ %% @doc Test creating a byte channel with default settings create_byte_channel_test(_Config) -> {ok, Ch} = py_byte_channel:new(), true = is_reference(Ch), ok = py_byte_channel:close(Ch). %% @doc Test creating a byte channel with max_size for backpressure create_byte_channel_with_max_size_test(_Config) -> {ok, Ch} = py_byte_channel:new(#{max_size => 1000}), true = is_reference(Ch), Info = py_byte_channel:info(Ch), 1000 = maps:get(max_size, Info), ok = py_byte_channel:close(Ch). %% @doc Test basic send of raw bytes erlang_send_bytes_test(_Config) -> {ok, Ch} = py_byte_channel:new(), ok = py_byte_channel:send(Ch, <<"hello">>), ok = py_byte_channel:send(Ch, <<"world">>), ok = py_byte_channel:close(Ch). %% @doc Test basic receive of raw bytes (no term decoding) erlang_receive_bytes_test(_Config) -> {ok, Ch} = py_byte_channel:new(), ok = py_byte_channel:send(Ch, <<"hello">>), %% Use byte channel try_receive - returns raw binary {ok, <<"hello">>} = py_nif:byte_channel_try_receive_bytes(Ch), ok = py_byte_channel:close(Ch). %% @doc Test sending and receiving multiple bytes send_receive_multiple_bytes_test(_Config) -> {ok, Ch} = py_byte_channel:new(), ok = py_byte_channel:send(Ch, <<"one">>), ok = py_byte_channel:send(Ch, <<"two">>), ok = py_byte_channel:send(Ch, <<"three">>), {ok, <<"one">>} = py_byte_channel:try_receive(Ch), {ok, <<"two">>} = py_byte_channel:try_receive(Ch), {ok, <<"three">>} = py_byte_channel:try_receive(Ch), ok = py_byte_channel:close(Ch). %% @doc Test try_receive on empty channel try_receive_empty_bytes_test(_Config) -> {ok, Ch} = py_byte_channel:new(), {error, empty} = py_byte_channel:try_receive(Ch), ok = py_byte_channel:close(Ch). %% @doc Test closing a byte channel close_byte_channel_test(_Config) -> {ok, Ch} = py_byte_channel:new(), ok = py_byte_channel:send(Ch, <<"data">>), ok = py_byte_channel:close(Ch), Info = py_byte_channel:info(Ch), true = maps:get(closed, Info). %% @doc Test byte channel info byte_channel_info_test(_Config) -> {ok, Ch} = py_byte_channel:new(#{max_size => 500}), Info1 = py_byte_channel:info(Ch), 0 = maps:get(size, Info1), 500 = maps:get(max_size, Info1), false = maps:get(closed, Info1), ok = py_byte_channel:send(Ch, <<"test">>), Info2 = py_byte_channel:info(Ch), %% Size should be exactly the binary size (no term overhead) 4 = maps:get(size, Info2), ok = py_byte_channel:close(Ch). %% @doc Test backpressure when queue exceeds max_size backpressure_bytes_test(_Config) -> %% Create channel with small max_size {ok, Ch} = py_byte_channel:new(#{max_size => 100}), %% Fill up the channel - no term overhead for raw bytes Data50 = binary:copy(<<0>>, 50), ok = py_byte_channel:send(Ch, Data50), %% Check current size Info1 = py_byte_channel:info(Ch), 50 = maps:get(size, Info1), ct:pal("After first send, size: ~p", [50]), %% Send another 50 bytes - should succeed (total 100) ok = py_byte_channel:send(Ch, Data50), Info2 = py_byte_channel:info(Ch), 100 = maps:get(size, Info2), ct:pal("After second send, size: ~p", [100]), %% Next send should return busy (backpressure) busy = py_byte_channel:send(Ch, <<"more">>), %% Drain one message {ok, _} = py_byte_channel:try_receive(Ch), %% Now should be able to send again ok = py_byte_channel:send(Ch, <<"small">>), ok = py_byte_channel:close(Ch). %% @doc Test sending to a closed channel send_bytes_to_closed_test(_Config) -> {ok, Ch} = py_byte_channel:new(), ok = py_byte_channel:close(Ch), {error, closed} = py_byte_channel:send(Ch, <<"data">>). %%% ============================================================================ %%% Python Tests %%% ============================================================================ %% @doc Test Python ByteChannel class is importable python_byte_channel_class_test(_Config) -> Ctx = py:context(1), %% Test that the ByteChannel module is importable via erlang namespace ok = py:exec(Ctx, <<"from erlang import ByteChannel, ByteChannelClosed">>), %% Test basic ByteChannel class behavior {ok, true} = py:eval(Ctx, <<"callable(ByteChannel)">>), {ok, true} = py:eval(Ctx, <<"issubclass(ByteChannelClosed, Exception)">>), ok. %% @doc Test Python ByteChannel send_bytes method python_byte_channel_send_bytes_test(_Config) -> Ctx = py:context(1), %% Test that send_bytes method exists ok = py:exec(Ctx, <<"from erlang import ByteChannel">>), {ok, true} = py:eval(Ctx, <<"hasattr(ByteChannel, 'send_bytes')">>), ok. %% @doc Test Python ByteChannel receive_bytes method python_byte_channel_receive_bytes_test(_Config) -> Ctx = py:context(1), %% Test that ByteChannel methods exist ok = py:exec(Ctx, <<"from erlang import ByteChannel">>), {ok, true} = py:eval(Ctx, <<"hasattr(ByteChannel, 'receive_bytes')">>), {ok, true} = py:eval(Ctx, <<"hasattr(ByteChannel, 'try_receive_bytes')">>), {ok, true} = py:eval(Ctx, <<"hasattr(ByteChannel, 'async_receive_bytes')">>), %% Verify async iteration methods {ok, true} = py:eval(Ctx, <<"hasattr(ByteChannel, '__aiter__')">>), {ok, true} = py:eval(Ctx, <<"hasattr(ByteChannel, '__anext__')">>), {ok, true} = py:eval(Ctx, <<"hasattr(ByteChannel, '__iter__')">>), ok. %%% ============================================================================ %%% Sync Blocking Receive Tests %%% ============================================================================ %% @doc Test sync receive when data is already available (immediate return) sync_receive_bytes_immediate_test(_Config) -> {ok, Ch} = py_byte_channel:new(), %% Send data before receive ok = py_byte_channel:send(Ch, <<"immediate_bytes">>), %% Receive should return immediately {ok, <<"immediate_bytes">>} = py_byte_channel:handle_receive_bytes([Ch]), ok = py_byte_channel:close(Ch). %% @doc Test sync receive that blocks waiting for data sync_receive_bytes_wait_test(_Config) -> {ok, Ch} = py_byte_channel:new(), Self = self(), %% Spawn a process to do blocking receive _Receiver = spawn_link(fun() -> Result = py_byte_channel:handle_receive_bytes([Ch]), Self ! {receive_result, Result} end), %% Give receiver time to register as waiter timer:sleep(50), %% Send data - should wake up the receiver ok = py_byte_channel:send(Ch, <<"delayed_bytes">>), %% Wait for result receive {receive_result, {ok, <<"delayed_bytes">>}} -> ok after 2000 -> ct:fail("Receiver did not get data within timeout") end, ok = py_byte_channel:close(Ch). %% @doc Test sync receive when channel is closed while waiting sync_receive_bytes_closed_test(_Config) -> {ok, Ch} = py_byte_channel:new(), Self = self(), %% Spawn a process to do blocking receive _Receiver = spawn_link(fun() -> Result = py_byte_channel:handle_receive_bytes([Ch]), Self ! {receive_result, Result} end), %% Give receiver time to register as waiter timer:sleep(50), %% Close the channel - should wake up receiver with error ok = py_byte_channel:close(Ch), %% Wait for result receive {receive_result, {error, closed}} -> ok after 2000 -> ct:fail("Receiver did not get closed notification within timeout") end. %%% ============================================================================ %%% Large Payload Test %%% ============================================================================ %% @doc Test sending and receiving large binary payloads large_payload_bytes_test(_Config) -> {ok, Ch} = py_byte_channel:new(), %% Create a 1MB binary LargeData = binary:copy(<<"X">>, 1024 * 1024), %% Send it ok = py_byte_channel:send(Ch, LargeData), %% Check size is correct (no term overhead) Info = py_byte_channel:info(Ch), 1048576 = maps:get(size, Info), %% Receive and verify {ok, ReceivedData} = py_byte_channel:try_receive(Ch), true = (byte_size(ReceivedData) =:= 1048576), true = (ReceivedData =:= LargeData), ok = py_byte_channel:close(Ch). %%% ============================================================================ %%% Async Event Loop Dispatch Test %%% ============================================================================ %% @doc Test async_receive_bytes with proper event loop dispatch (no polling) %% Sends data after async receive starts, verifies event-driven wakeup async_receive_bytes_e2e_test(_Config) -> {ok, Ch} = py_byte_channel:new(), %% Test 1: Immediate data - should return without waiting ok = py_byte_channel:send(Ch, <<"immediate_bytes">>), Ctx = py:context(1), %% Run async receive - data is already there, should return immediately {ok, <<"immediate_bytes">>} = py:eval(Ctx, <<"erlang.run(receive_bytes(ch))">>, #{<<"ch">> => Ch}), ct:pal("Async receive immediate data OK"), %% Test 2: Send data after async starts - tests event dispatch %% We send data first, then run async (to avoid race conditions in test) ok = py_byte_channel:send(Ch, <<"async_bytes">>), {ok, <<"async_bytes">>} = py:eval(Ctx, <<"erlang.run(receive_bytes(ch))">>, #{<<"ch">> => Ch}), ct:pal("Async receive via erlang.run() OK"), ok = py_byte_channel:close(Ch). %%% ============================================================================ %%% create_task + Async ByteChannel Test %%% ============================================================================ %% @doc Test async_receive_bytes works correctly with py_event_loop:create_task %% This verifies the loop capsule name is correct (erlang_python.event_loop) create_task_async_receive_test(_Config) -> {ok, Ch} = py_byte_channel:new(), %% First send data so it's available when task starts ok = py_byte_channel:send(Ch, <<"create_task_bytes">>), %% Define and run the task - env reuse should make __main__ functions visible Ctx = py:context(1), ok = py:exec(Ctx, <<" import erlang from erlang import ByteChannel async def task_receive_bytes(ch_ref, reply_pid): '''Task that receives bytes and sends result back to Erlang.''' try: ch = ByteChannel(ch_ref) data = await ch.async_receive_bytes() erlang.send(reply_pid, ('result', data)) except Exception as e: erlang.send(reply_pid, ('error', str(e))) ">>), %% Create a task that will await on channel receive %% create_task/3 uses the global event loop internally TaskRef = py_event_loop:create_task( "__main__", "task_receive_bytes", [Ch, self()]), %% Wait for result from the task %% Note: Python tuple ('result', data) becomes {<<"result">>, data} in Erlang receive {<<"result">>, <<"create_task_bytes">>} -> ct:pal("create_task + async_receive_bytes OK"); {<<"error">>, ErrMsg} -> ct:pal("Task error: ~p", [ErrMsg]), ct:fail({task_error, ErrMsg}); Other -> ct:pal("Unexpected message: ~p", [Other]), ct:fail({unexpected_message, Other}) after 5000 -> ct:fail("Timeout waiting for task result") end, %% Wait for task to complete case py_event_loop:await(TaskRef, 5000) of {ok, _} -> ok; {error, AwaitErr} -> ct:pal("Await error: ~p", [AwaitErr]) end, ok = py_byte_channel:close(Ch). %%% ============================================================================ %%% Close and Drain Tests %%% ============================================================================ %% @doc Test that data can be drained from byte channel after close (Erlang side) %% Verifies that closing a channel doesn't prevent reading existing data close_drain_bytes_erlang_test(_Config) -> {ok, Ch} = py_byte_channel:new(), %% Send multiple messages ok = py_byte_channel:send(Ch, <<"chunk1">>), ok = py_byte_channel:send(Ch, <<"chunk2">>), ok = py_byte_channel:send(Ch, <<"chunk3">>), %% Close the channel while data is still in queue ok = py_byte_channel:close(Ch), %% Verify channel is marked as closed Info = py_byte_channel:info(Ch), true = maps:get(closed, Info), %% Should still be able to drain all data {ok, <<"chunk1">>} = py_byte_channel:try_receive(Ch), {ok, <<"chunk2">>} = py_byte_channel:try_receive(Ch), {ok, <<"chunk3">>} = py_byte_channel:try_receive(Ch), %% Only after draining should we get closed error {error, closed} = py_byte_channel:try_receive(Ch), ct:pal("Erlang byte channel close+drain test passed"). %% @doc Test that Python can drain data from closed byte channel (sync iteration) close_drain_bytes_python_sync_test(_Config) -> {ok, Ch} = py_byte_channel:new(), %% Send multiple messages ok = py_byte_channel:send(Ch, <<"data1">>), ok = py_byte_channel:send(Ch, <<"data2">>), ok = py_byte_channel:send(Ch, <<"data3">>), %% Close the channel ok = py_byte_channel:close(Ch), %% Python should be able to drain all messages via iteration Ctx = py:context(1), ok = py:exec(Ctx, <<" from erlang import ByteChannel, ByteChannelClosed def drain_byte_channel(ch_ref): '''Drain all chunks from byte channel, return as list.''' ch = ByteChannel(ch_ref) chunks = [] for chunk in ch: chunks.append(chunk) return chunks ">>), {ok, [<<"data1">>, <<"data2">>, <<"data3">>]} = py:eval(Ctx, <<"drain_byte_channel(ch)">>, #{<<"ch">> => Ch}), ct:pal("Python sync byte channel close+drain test passed"). %% @doc Test that Python can drain data from closed byte channel (async iteration) close_drain_bytes_python_async_test(_Config) -> {ok, Ch} = py_byte_channel:new(), %% Send multiple messages ok = py_byte_channel:send(Ch, <<"part1">>), ok = py_byte_channel:send(Ch, <<"part2">>), ok = py_byte_channel:send(Ch, <<"part3">>), %% Close the channel ok = py_byte_channel:close(Ch), %% Python should be able to drain all messages via async iteration Ctx = py:context(1), ok = py:exec(Ctx, <<" import erlang from erlang import ByteChannel, ByteChannelClosed async def async_drain_byte_channel(ch_ref): '''Async drain all chunks from byte channel, return as list.''' ch = ByteChannel(ch_ref) chunks = [] async for chunk in ch: chunks.append(chunk) return chunks ">>), {ok, [<<"part1">>, <<"part2">>, <<"part3">>]} = py:eval(Ctx, <<"erlang.run(async_drain_byte_channel(ch))">>, #{<<"ch">> => Ch}), ct:pal("Python async byte channel close+drain test passed"). %% @doc Test async drain with create_task when data arrives after task starts %% This tests the notification callback path: task registers waiter, then data arrives close_drain_bytes_create_task_async_test(_Config) -> {ok, Ch} = py_byte_channel:new(), %% Define the async drain task Ctx = py:context(1), ok = py:exec(Ctx, <<" import erlang from erlang import ByteChannel async def drain_task(ch_ref, reply_pid): '''Task that drains byte channel and sends results back.''' try: ch = ByteChannel(ch_ref) chunks = [] async for chunk in ch: chunks.append(chunk) erlang.send(reply_pid, ('result', chunks)) except Exception as e: erlang.send(reply_pid, ('error', str(e))) ">>), %% Create the task BEFORE sending any data %% This forces the task to register a waiter and wait for notifications TaskRef = py_event_loop:create_task( "__main__", "drain_task", [Ch, self()]), %% Give the task time to start and register waiter timer:sleep(100), %% Now send data - should trigger notification callback ok = py_byte_channel:send(Ch, <<"chunk1">>), ok = py_byte_channel:send(Ch, <<"chunk2">>), ok = py_byte_channel:send(Ch, <<"chunk3">>), %% Close the channel to signal end of stream ok = py_byte_channel:close(Ch), %% Wait for result from the task receive {<<"result">>, [<<"chunk1">>, <<"chunk2">>, <<"chunk3">>]} -> ct:pal("create_task async drain with delayed data OK"); {<<"result">>, Other} -> ct:pal("Unexpected result: ~p", [Other]), ct:fail({unexpected_result, Other}); {<<"error">>, ErrMsg} -> ct:pal("Task error: ~p", [ErrMsg]), ct:fail({task_error, ErrMsg}) after 5000 -> ct:fail("Timeout waiting for drain task result") end, %% Wait for task to complete case py_event_loop:await(TaskRef, 5000) of {ok, _} -> ok; {error, AwaitErr} -> ct:pal("Await error: ~p", [AwaitErr]) end.