Current section
Files
Jump to
Current section
Files
test/py_channel_SUITE.erl
%%% @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
]).
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
].
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.