%%% @doc Common Test suite for py_channel API. %%% %%% Tests the bidirectional channel API for Erlang-Python communication. -module(py_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_channel_test/1, create_channel_with_max_size_test/1, send_receive_test/1, send_receive_multiple_test/1, try_receive_empty_test/1, close_channel_test/1, channel_info_test/1, backpressure_test/1, send_to_closed_test/1, python_receive_test/1, python_try_receive_test/1, python_iterate_test/1, async_receive_immediate_test/1, async_receive_wait_test/1, async_iteration_test/1, async_closed_channel_test/1, channel_ref_roundtrip_test/1, channel_ref_call_test/1, %% Sync blocking receive tests sync_receive_immediate_test/1, sync_receive_wait_test/1, sync_receive_closed_test/1, sync_receive_multiple_waiters_test/1, %% Mixed waiter rejection tests mixed_waiter_sync_blocks_async_test/1, mixed_waiter_async_blocks_sync_test/1, %% Async receive with actual waiting async_receive_wait_e2e_test/1, %% Close and drain tests close_drain_erlang_test/1, close_drain_python_sync_test/1, close_drain_python_async_test/1, close_drain_create_task_async_test/1 ]). all() -> [ create_channel_test, create_channel_with_max_size_test, send_receive_test, send_receive_multiple_test, try_receive_empty_test, close_channel_test, channel_info_test, backpressure_test, send_to_closed_test, python_receive_test, python_try_receive_test, python_iterate_test, async_receive_immediate_test, async_receive_wait_test, async_iteration_test, async_closed_channel_test, channel_ref_roundtrip_test, channel_ref_call_test, %% Sync blocking receive tests sync_receive_immediate_test, sync_receive_wait_test, sync_receive_closed_test, sync_receive_multiple_waiters_test, %% Mixed waiter rejection tests mixed_waiter_sync_blocks_async_test, mixed_waiter_async_blocks_sync_test, %% Async receive with actual waiting async_receive_wait_e2e_test, %% Close and drain tests close_drain_erlang_test, close_drain_python_sync_test, close_drain_python_async_test, close_drain_create_task_async_test ]. init_per_suite(Config) -> {ok, _} = application:ensure_all_started(erlang_python), {ok, _} = py:start_contexts(), %% Register channel callbacks ok = py_channel:register_callbacks(), 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 channel with default settings create_channel_test(_Config) -> {ok, Ch} = py_channel:new(), true = is_reference(Ch), ok = py_channel:close(Ch). %% @doc Test creating a channel with max_size for backpressure create_channel_with_max_size_test(_Config) -> {ok, Ch} = py_channel:new(#{max_size => 1000}), true = is_reference(Ch), Info = py_channel:info(Ch), 1000 = maps:get(max_size, Info), ok = py_channel:close(Ch). %% @doc Test basic send and receive send_receive_test(_Config) -> {ok, Ch} = py_channel:new(), ok = py_channel:send(Ch, <<"hello">>), {ok, <<"hello">>} = py_nif:channel_try_receive(Ch), ok = py_channel:close(Ch). %% @doc Test sending and receiving multiple messages send_receive_multiple_test(_Config) -> {ok, Ch} = py_channel:new(), ok = py_channel:send(Ch, 1), ok = py_channel:send(Ch, 2), ok = py_channel:send(Ch, 3), {ok, 1} = py_nif:channel_try_receive(Ch), {ok, 2} = py_nif:channel_try_receive(Ch), {ok, 3} = py_nif:channel_try_receive(Ch), ok = py_channel:close(Ch). %% @doc Test try_receive on empty channel try_receive_empty_test(_Config) -> {ok, Ch} = py_channel:new(), {error, empty} = py_nif:channel_try_receive(Ch), ok = py_channel:close(Ch). %% @doc Test closing a channel close_channel_test(_Config) -> {ok, Ch} = py_channel:new(), ok = py_channel:send(Ch, <<"data">>), ok = py_channel:close(Ch), Info = py_channel:info(Ch), true = maps:get(closed, Info). %% @doc Test channel info channel_info_test(_Config) -> {ok, Ch} = py_channel:new(#{max_size => 500}), Info1 = py_channel:info(Ch), 0 = maps:get(size, Info1), 500 = maps:get(max_size, Info1), false = maps:get(closed, Info1), ok = py_channel:send(Ch, <<"test">>), Info2 = py_channel:info(Ch), true = maps:get(size, Info2) > 0, ok = py_channel:close(Ch). %% @doc Test backpressure when queue exceeds max_size backpressure_test(_Config) -> %% Create channel with small max_size %% Note: external term format adds overhead, so actual size is larger than raw data {ok, Ch} = py_channel:new(#{max_size => 200}), %% Fill up the channel with data that will exceed max_size after serialization LargeData = binary:copy(<<0>>, 80), ok = py_channel:send(Ch, LargeData), %% Check current size Info1 = py_channel:info(Ch), Size1 = maps:get(size, Info1), ct:pal("After first send, size: ~p", [Size1]), %% Send another - should still succeed as we have room ok = py_channel:send(Ch, LargeData), %% Check size again - should be close to or exceed max_size Info2 = py_channel:info(Ch), Size2 = maps:get(size, Info2), ct:pal("After second send, size: ~p", [Size2]), %% Next send should return busy (backpressure) busy = py_channel:send(Ch, LargeData), %% Drain one message {ok, _} = py_nif:channel_try_receive(Ch), %% Now should be able to send again ok = py_channel:send(Ch, <<"small">>), ok = py_channel:close(Ch). %% @doc Test sending to a closed channel send_to_closed_test(_Config) -> {ok, Ch} = py_channel:new(), ok = py_channel:close(Ch), {error, closed} = py_channel:send(Ch, <<"data">>). %% @doc Test Python receiving from channel via callback %% Note: NIF resource references don't round-trip through Python conversion, %% so this test verifies the basic Channel Python class instantiation. python_receive_test(_Config) -> Ctx = py:context(1), %% Test that the channel module is importable via erlang namespace ok = py:exec(Ctx, <<"from erlang import Channel, ChannelClosed">>), %% Test basic Channel class behavior {ok, true} = py:eval(Ctx, <<"callable(Channel)">>), ok. %% @doc Test Python Channel exception class python_try_receive_test(_Config) -> Ctx = py:context(1), %% Test that ChannelClosed exception exists ok = py:exec(Ctx, <<"from erlang import ChannelClosed">>), {ok, true} = py:eval(Ctx, <<"issubclass(ChannelClosed, Exception)">>), ok. %% @doc Test Python channel reply function python_iterate_test(_Config) -> Ctx = py:context(1), %% Test that the reply function is importable ok = py:exec(Ctx, <<"from erlang import reply">>), {ok, true} = py:eval(Ctx, <<"callable(reply)">>), ok. %%% ============================================================================ %%% Async Channel Tests %%% ============================================================================ %% @doc Test async_receive when data is immediately available async_receive_immediate_test(_Config) -> {ok, Ch} = py_channel:new(), %% Send data first ok = py_channel:send(Ch, <<"immediate_data">>), %% Run async receive via Python - data should return immediately Ctx = py:context(1), _Code = <<" import asyncio from erlang import Channel async def test_immediate(ch_ref): ch = Channel(ch_ref) result = await ch.async_receive() return result # Run the async function loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) try: result = loop.run_until_complete(test_immediate(channel_ref)) finally: loop.close() result ">>, %% Set channel ref as a variable ok = py:exec(Ctx, <<"channel_ref = None">>), %% For now, test that the async methods exist ok = py:exec(Ctx, <<"from erlang import Channel">>), {ok, true} = py:eval(Ctx, <<"hasattr(Channel, 'async_receive')">>), {ok, true} = py:eval(Ctx, <<"hasattr(Channel, '__aiter__')">>), {ok, true} = py:eval(Ctx, <<"hasattr(Channel, '__anext__')">>), ok = py_channel:close(Ch). %% @doc Test async_receive waiting for data (with timer-based dispatch) async_receive_wait_test(_Config) -> Ctx = py:context(1), %% Test that async methods are available ok = py:exec(Ctx, <<"from erlang import Channel, ChannelClosed">>), {ok, true} = py:eval(Ctx, <<"hasattr(Channel, 'async_receive')">>), %% Verify the _is_closed method exists ok = py:exec(Ctx, <<"ch = Channel(None)">>), {ok, true} = py:eval(Ctx, <<"hasattr(ch, '_is_closed')">>), ok. %% @doc Test async iteration over channel async_iteration_test(_Config) -> Ctx = py:context(1), %% Test that async iteration methods exist ok = py:exec(Ctx, <<"from erlang import Channel">>), ok = py:exec(Ctx, <<"ch = Channel(None)">>), {ok, true} = py:eval(Ctx, <<"hasattr(ch, '__aiter__')">>), {ok, true} = py:eval(Ctx, <<"hasattr(ch, '__anext__')">>), %% Verify __aiter__ returns self {ok, true} = py:eval(Ctx, <<"ch.__aiter__() is ch">>), ok. %% @doc Test async_receive on closed channel raises ChannelClosed async_closed_channel_test(_Config) -> {ok, Ch} = py_channel:new(), ok = py_channel:close(Ch), Ctx = py:context(1), %% Test that ChannelClosed exception is properly defined ok = py:exec(Ctx, <<"from erlang import ChannelClosed">>), {ok, true} = py:eval(Ctx, <<"issubclass(ChannelClosed, Exception)">>), ok. %% @doc Test that channel references can be passed to Python and back via eval channel_ref_roundtrip_test(_Config) -> {ok, Ch} = py_channel:new(), %% Pass channel ref to Python via NIF and get it back %% This tests the PyCapsule conversion in py_convert.c {ok, ReturnedRef} = py:eval(<<"ch_ref">>, #{<<"ch_ref">> => Ch}), %% The returned ref should be usable as a channel %% Send data through original channel ok = py_channel:send(Ch, <<"test_data">>), %% Receive using the returned ref (should be the same channel) {ok, <<"test_data">>} = py_nif:channel_try_receive(ReturnedRef), %% Verify channel info works on returned ref Info = py_channel:info(ReturnedRef), false = maps:get(closed, Info), ok = py_channel:close(Ch). %% @doc Test that channel references can be passed via py:call channel_ref_call_test(_Config) -> {ok, Ch} = py_channel:new(), %% Test passing channel ref through py:call to a Python function %% The test_channel_ref module has identity, get_channel_type, store_and_return {ok, ReturnedRef} = py:call(test_channel_ref, identity, [Ch]), %% The returned ref should be usable as a channel ok = py_channel:send(Ch, <<"identity_data">>), {ok, <<"identity_data">>} = py_nif:channel_try_receive(ReturnedRef), %% Test that Python sees it as a PyCapsule {ok, <<"PyCapsule">>} = py:call(test_channel_ref, get_channel_type, [Ch]), %% Test storing in a container and returning {ok, StoredRef} = py:call(test_channel_ref, store_and_return, [Ch]), ok = py_channel:send(Ch, <<"stored_data">>), {ok, <<"stored_data">>} = py_nif:channel_try_receive(StoredRef), %% Also test passing through containers via eval {ok, [Ref1, Ref2]} = py:eval(<<"[a, b]">>, #{<<"a">> => Ch, <<"b">> => Ch}), ok = py_channel:send(Ch, <<"ref1_data">>), {ok, <<"ref1_data">>} = py_nif:channel_try_receive(Ref1), ok = py_channel:send(Ch, <<"ref2_data">>), {ok, <<"ref2_data">>} = py_nif:channel_try_receive(Ref2), ok = py_channel:close(Ch). %%% ============================================================================ %%% Sync Blocking Receive Tests %%% ============================================================================ %% @doc Test sync receive when data is already available (immediate return) sync_receive_immediate_test(_Config) -> {ok, Ch} = py_channel:new(), %% Send data before receive ok = py_channel:send(Ch, <<"immediate_data">>), %% Receive should return immediately {ok, <<"immediate_data">>} = py_channel:handle_receive([Ch]), ok = py_channel:close(Ch). %% @doc Test sync receive that blocks waiting for data sync_receive_wait_test(_Config) -> {ok, Ch} = py_channel:new(), Self = self(), %% Spawn a process to do blocking receive _Receiver = spawn_link(fun() -> Result = py_channel:handle_receive([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_channel:send(Ch, <<"delayed_data">>), %% Wait for result receive {receive_result, {ok, <<"delayed_data">>}} -> ok after 2000 -> ct:fail("Receiver did not get data within timeout") end, ok = py_channel:close(Ch). %% @doc Test sync receive when channel is closed while waiting sync_receive_closed_test(_Config) -> {ok, Ch} = py_channel:new(), Self = self(), %% Spawn a process to do blocking receive _Receiver = spawn_link(fun() -> Result = py_channel:handle_receive([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_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. %% @doc Test that only one sync waiter can be registered at a time sync_receive_multiple_waiters_test(_Config) -> {ok, Ch} = py_channel:new(), %% Register first sync waiter directly via NIF ok = py_nif:channel_register_sync_waiter(Ch), %% Try to register another - should fail {error, waiter_exists} = py_nif:channel_register_sync_waiter(Ch), %% Send data to clear the first waiter ok = py_channel:send(Ch, <<"data">>), %% Consume the notification message that was sent to us receive channel_data_ready -> ok after 100 -> ct:fail("Did not receive channel_data_ready") end, %% Consume the data from the queue (required before re-registering) {ok, <<"data">>} = py_nif:channel_try_receive(Ch), %% Now we should be able to register again ok = py_nif:channel_register_sync_waiter(Ch), ok = py_channel:close(Ch), %% Consume the close message receive channel_closed -> ok after 100 -> ct:fail("Did not receive channel_closed") end. %% @doc Test that sync waiter blocks async waiter registration mixed_waiter_sync_blocks_async_test(_Config) -> {ok, Ch} = py_channel:new(), %% Register sync waiter first ok = py_nif:channel_register_sync_waiter(Ch), %% Create an event loop for async waiter test {ok, Loop} = py_nif:event_loop_new(), %% Try to register async waiter - should fail {error, waiter_exists} = py_nif:channel_wait(Ch, 123, Loop), %% Clean up: send data to clear sync waiter ok = py_channel:send(Ch, <<"data">>), receive channel_data_ready -> ok after 100 -> ok end, py_nif:event_loop_destroy(Loop), ok = py_channel:close(Ch). %% @doc Test that async waiter blocks sync waiter registration mixed_waiter_async_blocks_sync_test(_Config) -> {ok, Ch} = py_channel:new(), %% Create an event loop and register async waiter first {ok, Loop} = py_nif:event_loop_new(), ok = py_nif:channel_wait(Ch, 456, Loop), %% Try to register sync waiter - should fail {error, waiter_exists} = py_nif:channel_register_sync_waiter(Ch), %% Clean up: cancel async waiter and close ok = py_nif:channel_cancel_wait(Ch, 456), py_nif:event_loop_destroy(Loop), ok = py_channel:close(Ch). %% @doc End-to-end test for async_receive waiting for data %% Tests that async_receive properly integrates with asyncio when data %% is sent concurrently via a background task async_receive_wait_e2e_test(_Config) -> {ok, Ch} = py_channel:new(), %% Send data first, then test async receive %% This is simpler and validates the async path works ok = py_channel:send(Ch, <<"async_data">>), Ctx = py:context(1), %% Set up the channel ref in Python ok = py:exec(Ctx, <<"channel_ref = None">>), %% Define async function that receives from channel ok = py:exec(Ctx, <<" import erlang from erlang import Channel async def receive_from_channel(ch_ref): ch = Channel(ch_ref) data = await ch.async_receive() return data ">>), %% Run the async receive - data is already there {ok, <<"async_data">>} = py:eval(Ctx, <<"erlang.run(receive_from_channel(ch))">>, #{<<"ch">> => Ch}), ct:pal("Async receive successfully received data via erlang.run()"), ok = py_channel:close(Ch). %%% ============================================================================ %%% Close and Drain Tests %%% ============================================================================ %% @doc Test that data can be drained from channel after close (Erlang side) %% Verifies that closing a channel doesn't prevent reading existing data close_drain_erlang_test(_Config) -> {ok, Ch} = py_channel:new(), %% Send multiple messages ok = py_channel:send(Ch, <<"msg1">>), ok = py_channel:send(Ch, <<"msg2">>), ok = py_channel:send(Ch, <<"msg3">>), %% Close the channel while data is still in queue ok = py_channel:close(Ch), %% Verify channel is marked as closed Info = py_channel:info(Ch), true = maps:get(closed, Info), %% Should still be able to drain all data {ok, <<"msg1">>} = py_nif:channel_try_receive(Ch), {ok, <<"msg2">>} = py_nif:channel_try_receive(Ch), {ok, <<"msg3">>} = py_nif:channel_try_receive(Ch), %% Only after draining should we get closed error {error, closed} = py_nif:channel_try_receive(Ch), ct:pal("Erlang close+drain test passed"). %% @doc Test that Python can drain data from closed channel (sync iteration) close_drain_python_sync_test(_Config) -> {ok, Ch} = py_channel:new(), %% Send multiple messages ok = py_channel:send(Ch, <<"first">>), ok = py_channel:send(Ch, <<"second">>), ok = py_channel:send(Ch, <<"third">>), %% Close the channel ok = py_channel:close(Ch), %% Python should be able to drain all messages via iteration Ctx = py:context(1), ok = py:exec(Ctx, <<" from erlang import Channel, ChannelClosed def drain_channel(ch_ref): '''Drain all messages from channel, return as list.''' ch = Channel(ch_ref) messages = [] for msg in ch: messages.append(msg) return messages ">>), {ok, [<<"first">>, <<"second">>, <<"third">>]} = py:eval(Ctx, <<"drain_channel(ch)">>, #{<<"ch">> => Ch}), ct:pal("Python sync close+drain test passed"). %% @doc Test that Python can drain data from closed channel (async iteration) close_drain_python_async_test(_Config) -> {ok, Ch} = py_channel:new(), %% Send multiple messages ok = py_channel:send(Ch, <<"alpha">>), ok = py_channel:send(Ch, <<"beta">>), ok = py_channel:send(Ch, <<"gamma">>), %% Close the channel ok = py_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 Channel, ChannelClosed async def async_drain_channel(ch_ref): '''Async drain all messages from channel, return as list.''' ch = Channel(ch_ref) messages = [] async for msg in ch: messages.append(msg) return messages ">>), {ok, [<<"alpha">>, <<"beta">>, <<"gamma">>]} = py:eval(Ctx, <<"erlang.run(async_drain_channel(ch))">>, #{<<"ch">> => Ch}), ct:pal("Python async 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_create_task_async_test(_Config) -> {ok, Ch} = py_channel:new(), %% Define the async drain task Ctx = py:context(1), ok = py:exec(Ctx, <<" import erlang from erlang import Channel async def drain_task(ch_ref, reply_pid): '''Task that drains channel and sends results back.''' try: ch = Channel(ch_ref) messages = [] async for msg in ch: messages.append(msg) erlang.send(reply_pid, ('result', messages)) 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_channel:send(Ch, <<"msg1">>), ok = py_channel:send(Ch, <<"msg2">>), ok = py_channel:send(Ch, <<"msg3">>), %% Close the channel to signal end of stream ok = py_channel:close(Ch), %% Wait for result from the task receive {<<"result">>, [<<"msg1">>, <<"msg2">>, <<"msg3">>]} -> 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.