%%% @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.