%%% @doc Common Test suite for Erlang-native asyncio event loop.

%%%

%%% Tests the enif_select-based event loop implementation that provides

%%% sub-millisecond latency for asyncio operations.

-module(py_event_loop_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([

test_event_loop_create_destroy/1,

test_event_loop_set_worker/1,

test_add_remove_reader/1,

test_add_remove_writer/1,

test_call_later_basic/1,

test_call_later_cancel/1,

test_multiple_timers/1,

test_fd_read_callback/1,

test_poll_events_timeout/1,

test_concurrent_fd_and_timer/1,

test_event_loop_wakeup/1,

test_get_pending_clears_queue/1,

%% TCP tests

test_tcp_listener_create/1,

test_tcp_connect_accept/1,

test_tcp_read_write_callback/1,

test_tcp_echo/1,

%% Lifecycle tests

test_cancel_reader_keeps_fd_open/1,

test_cancel_writer_keeps_fd_open/1,

test_double_close_is_safe/1,

test_cancel_then_reselect/1

]).

all() ->

[

test_event_loop_create_destroy,

test_event_loop_set_worker,

test_add_remove_reader,

test_add_remove_writer,

test_call_later_basic,

test_call_later_cancel,

test_multiple_timers,

test_fd_read_callback,

test_poll_events_timeout,

test_concurrent_fd_and_timer,

test_event_loop_wakeup,

test_get_pending_clears_queue,

%% TCP tests

test_tcp_listener_create,

test_tcp_connect_accept,

test_tcp_read_write_callback,

test_tcp_echo,

%% Lifecycle tests

test_cancel_reader_keeps_fd_open,

test_cancel_writer_keeps_fd_open,

test_double_close_is_safe,

test_cancel_then_reselect

].

init_per_suite(Config) ->

case application:ensure_all_started(erlang_python) of

{ok, _} ->

{ok, _} = py:start_contexts(),

%% Wait for event loop to be fully initialized

%% This is important for free-threaded Python where initialization

%% can race with test execution

case wait_for_event_loop(5000) of

ok ->

Config;

{error, Reason} ->

ct:fail({event_loop_not_ready, Reason})

end;

{error, {App, Reason}} ->

ct:fail({failed_to_start, App, Reason})

end.

%% Wait for the event loop to be fully initialized

wait_for_event_loop(Timeout) when Timeout =< 0 ->

{error, timeout};

wait_for_event_loop(Timeout) ->

case py_event_loop:get_loop() of

{ok, LoopRef} when is_reference(LoopRef) ->

%% Verify the event loop is actually functional by checking

%% we can create and immediately cancel a timer

case py_nif:event_loop_new() of

{ok, TestLoop} ->

py_nif:event_loop_destroy(TestLoop),

ok;

_ ->

timer:sleep(100),

wait_for_event_loop(Timeout - 100)

end;

_ ->

timer:sleep(100),

wait_for_event_loop(Timeout - 100)

end.

end_per_suite(_Config) ->

ok = application:stop(erlang_python),

ok.

init_per_testcase(_TestCase, Config) ->

Config.

end_per_testcase(_TestCase, _Config) ->

ok.

%% ============================================================================

%% Event Loop Lifecycle Tests

%% ============================================================================

test_event_loop_create_destroy(_Config) ->

%% Create event loop

{ok, LoopRef} = py_nif:event_loop_new(),

true = is_reference(LoopRef),

%% Destroy it

ok = py_nif:event_loop_destroy(LoopRef),

ok.

test_event_loop_set_worker(_Config) ->

{ok, LoopRef} = py_nif:event_loop_new(),

%% Start a worker process

{ok, WorkerPid} = py_event_worker:start_link(<<"test">>, LoopRef),

true = is_pid(WorkerPid),

%% Set worker

ok = py_nif:event_loop_set_worker(LoopRef, WorkerPid),

%% Cleanup

py_event_worker:stop(WorkerPid),

py_nif:event_loop_destroy(LoopRef),

ok.

%% ============================================================================

%% File Descriptor Monitoring Tests (using pipes - they work with enif_select)

%% ============================================================================

test_add_remove_reader(_Config) ->

{ok, LoopRef} = py_nif:event_loop_new(),

{ok, WorkerPid} = py_event_worker:start_link(<<"test">>, LoopRef),

ok = py_nif:event_loop_set_worker(LoopRef, WorkerPid),

%% Create a pipe for testing (pipes work properly with enif_select)

{ok, {ReadFd, WriteFd}} = py_nif:create_test_pipe(),

%% Add reader

CallbackId = 1,

{ok, FdRef} = py_nif:add_reader(LoopRef, ReadFd, CallbackId),

%% Remove reader (pass FdRef, not Fd)

ok = py_nif:remove_reader(LoopRef, FdRef),

%% Cleanup

py_nif:close_test_fd(ReadFd),

py_nif:close_test_fd(WriteFd),

py_event_worker:stop(WorkerPid),

py_nif:event_loop_destroy(LoopRef),

ok.

test_add_remove_writer(_Config) ->

{ok, LoopRef} = py_nif:event_loop_new(),

{ok, WorkerPid} = py_event_worker:start_link(<<"test">>, LoopRef),

ok = py_nif:event_loop_set_worker(LoopRef, WorkerPid),

{ok, {ReadFd, WriteFd}} = py_nif:create_test_pipe(),

%% Writer monitors the write end

{ok, FdRef} = py_nif:add_writer(LoopRef, WriteFd, 1),

%% Remove writer (pass FdRef, not Fd)

ok = py_nif:remove_writer(LoopRef, FdRef),

py_nif:close_test_fd(ReadFd),

py_nif:close_test_fd(WriteFd),

py_event_worker:stop(WorkerPid),

py_nif:event_loop_destroy(LoopRef),

ok.

test_fd_read_callback(_Config) ->

{ok, LoopRef} = py_nif:event_loop_new(),

{ok, WorkerPid} = py_event_worker:start_link(<<"test">>, LoopRef),

ok = py_nif:event_loop_set_worker(LoopRef, WorkerPid),

%% Create pipe - read end for monitoring, write end for triggering

{ok, {ReadFd, WriteFd}} = py_nif:create_test_pipe(),

%% Register reader callback on the read end

CallbackId = 42,

{ok, FdRef} = py_nif:add_reader(LoopRef, ReadFd, CallbackId),

%% Write data to the write end to trigger read readiness on read end

ok = py_nif:write_test_fd(WriteFd, <<"hello">>),

%% Wait for callback dispatch (worker receives {select, ...})

timer:sleep(100),

%% Get pending events

Pending = py_nif:get_pending(LoopRef),

%% Should have received the read callback

HasReadCallback = lists:any(

fun({CId, Type}) -> CId == CallbackId andalso Type == read end,

Pending

),

true = HasReadCallback,

%% Cleanup - remove reader before closing fd

ok = py_nif:remove_reader(LoopRef, FdRef),

py_nif:close_test_fd(ReadFd),

py_nif:close_test_fd(WriteFd),

py_event_worker:stop(WorkerPid),

py_nif:event_loop_destroy(LoopRef),

ok.

%% ============================================================================

%% Timer Tests

%% ============================================================================

test_call_later_basic(_Config) ->

{ok, LoopRef} = py_nif:event_loop_new(),

{ok, WorkerPid} = py_event_worker:start_link(<<"test">>, LoopRef),

ok = py_nif:event_loop_set_worker(LoopRef, WorkerPid),

CallbackId = 100,

DelayMs = 50,

T1 = erlang:monotonic_time(millisecond),

{ok, _TimerRef} = py_nif:call_later(LoopRef, DelayMs, CallbackId),

%% Wait for timer to fire

timer:sleep(DelayMs + 50),

T2 = erlang:monotonic_time(millisecond),

%% Get pending events

Pending = py_nif:get_pending(LoopRef),

%% Verify callback was triggered

HasTimerCallback = lists:any(

fun({CId, Type}) -> CId == CallbackId andalso Type == timer end,

Pending

),

true = HasTimerCallback,

%% Verify timing accuracy (within tolerance)

Elapsed = T2 - T1,

true = Elapsed >= DelayMs,

py_event_worker:stop(WorkerPid),

py_nif:event_loop_destroy(LoopRef),

ok.

test_call_later_cancel(_Config) ->

{ok, LoopRef} = py_nif:event_loop_new(),

{ok, WorkerPid} = py_event_worker:start_link(<<"test">>, LoopRef),

ok = py_nif:event_loop_set_worker(LoopRef, WorkerPid),

CallbackId = 101,

{ok, TimerRef} = py_nif:call_later(LoopRef, 200, CallbackId),

%% Cancel before it fires

ok = py_nif:cancel_timer(LoopRef, TimerRef),

%% Wait past the scheduled time

timer:sleep(300),

%% Verify callback was NOT triggered

Pending = py_nif:get_pending(LoopRef),

HasTimerCallback = lists:any(

fun({CId, Type}) -> CId == CallbackId andalso Type == timer end,

Pending

),

false = HasTimerCallback,

py_event_worker:stop(WorkerPid),

py_nif:event_loop_destroy(LoopRef),

ok.

test_multiple_timers(_Config) ->

{ok, LoopRef} = py_nif:event_loop_new(),

{ok, WorkerPid} = py_event_worker:start_link(<<"test">>, LoopRef),

ok = py_nif:event_loop_set_worker(LoopRef, WorkerPid),

%% Create timers with different delays

%% Use larger delays for CI reliability (especially on free-threaded Python)

{ok, _} = py_nif:call_later(LoopRef, 20, 1),

{ok, _} = py_nif:call_later(LoopRef, 50, 2),

{ok, _} = py_nif:call_later(LoopRef, 80, 3),

%% Wait for all with margin

timer:sleep(200),

Pending = py_nif:get_pending(LoopRef),

%% All three should have fired

CallbackIds = [CId || {CId, timer} <- Pending],

true = lists:member(1, CallbackIds),

true = lists:member(2, CallbackIds),

true = lists:member(3, CallbackIds),

py_event_worker:stop(WorkerPid),

py_nif:event_loop_destroy(LoopRef),

ok.

%% ============================================================================

%% Poll Events Tests

%% ============================================================================

test_poll_events_timeout(_Config) ->

{ok, LoopRef} = py_nif:event_loop_new(),

{ok, WorkerPid} = py_event_worker:start_link(<<"test">>, LoopRef),

ok = py_nif:event_loop_set_worker(LoopRef, WorkerPid),

%% Poll with short timeout, no events

T1 = erlang:monotonic_time(millisecond),

{ok, NumEvents} = py_nif:poll_events(LoopRef, 50),

T2 = erlang:monotonic_time(millisecond),

%% Should have 0 events and taken ~50ms

0 = NumEvents,

Elapsed = T2 - T1,

true = Elapsed >= 40, %% Allow some tolerance

py_event_worker:stop(WorkerPid),

py_nif:event_loop_destroy(LoopRef),

ok.

test_event_loop_wakeup(_Config) ->

{ok, LoopRef} = py_nif:event_loop_new(),

{ok, WorkerPid} = py_event_worker:start_link(<<"test">>, LoopRef),

ok = py_nif:event_loop_set_worker(LoopRef, WorkerPid),

%% Spawn a process to wait

Self = self(),

spawn(fun() ->

T1 = erlang:monotonic_time(millisecond),

{ok, _} = py_nif:poll_events(LoopRef, 5000),

T2 = erlang:monotonic_time(millisecond),

Self ! {poll_done, T2 - T1}

end),

%% Give it time to start waiting

timer:sleep(50),

%% Wake it up

ok = py_nif:event_loop_wakeup(LoopRef),

%% Should receive quickly (not 5 seconds)

receive

{poll_done, Elapsed} ->

true = Elapsed < 1000 %% Should be much less than 5000ms

after 2000 ->

ct:fail(poll_not_woken_up)

end,

py_event_worker:stop(WorkerPid),

py_nif:event_loop_destroy(LoopRef),

ok.

test_get_pending_clears_queue(_Config) ->

{ok, LoopRef} = py_nif:event_loop_new(),

{ok, WorkerPid} = py_event_worker:start_link(<<"test">>, LoopRef),

ok = py_nif:event_loop_set_worker(LoopRef, WorkerPid),

%% Create a timer that fires quickly

%% Use 20ms minimum for CI reliability

{ok, _} = py_nif:call_later(LoopRef, 20, 999),

timer:sleep(100),

%% First get_pending should return the event

Pending1 = py_nif:get_pending(LoopRef),

true = length(Pending1) > 0,

%% Second get_pending should return empty (queue was cleared)

Pending2 = py_nif:get_pending(LoopRef),

0 = length(Pending2),

py_event_worker:stop(WorkerPid),

py_nif:event_loop_destroy(LoopRef),

ok.

test_concurrent_fd_and_timer(_Config) ->

{ok, LoopRef} = py_nif:event_loop_new(),

{ok, WorkerPid} = py_event_worker:start_link(<<"test">>, LoopRef),

ok = py_nif:event_loop_set_worker(LoopRef, WorkerPid),

%% Set up pipe and timer

{ok, {ReadFd, WriteFd}} = py_nif:create_test_pipe(),

FdCallbackId = 1,

TimerCallbackId = 2,

%% Keep references to prevent GC

{ok, FdRef} = py_nif:add_reader(LoopRef, ReadFd, FdCallbackId),

{ok, TimerRef} = py_nif:call_later(LoopRef, 50, TimerCallbackId),

%% Trigger fd ready by writing to pipe

ok = py_nif:write_test_fd(WriteFd, <<"data">>),

%% Wait for both - use longer timeout for reliability

timer:sleep(200),

Pending = py_nif:get_pending(LoopRef),

%% Should have both callbacks

HasFdCallback = lists:any(

fun({CId, Type}) -> CId == FdCallbackId andalso Type == read end,

Pending

),

HasTimerCallback = lists:any(

fun({CId, Type}) -> CId == TimerCallbackId andalso Type == timer end,

Pending

),

%% Keep refs alive until here

_ = TimerRef,

true = HasFdCallback,

true = HasTimerCallback,

%% Cleanup - remove reader before closing fd

ok = py_nif:remove_reader(LoopRef, FdRef),

py_nif:close_test_fd(ReadFd),

py_nif:close_test_fd(WriteFd),

py_event_worker:stop(WorkerPid),

py_nif:event_loop_destroy(LoopRef),

ok.

%% ============================================================================

%% TCP Socket Tests

%% ============================================================================

test_tcp_listener_create(_Config) ->

%% Create a TCP listener on an ephemeral port

{ok, {ListenFd, Port}} = py_nif:create_test_tcp_listener(0),

true = is_integer(ListenFd),

true = ListenFd > 0,

true = is_integer(Port),

true = Port > 0,

%% Cleanup

py_nif:close_test_fd(ListenFd),

ok.

test_tcp_connect_accept(_Config) ->

%% Create listener

{ok, {ListenFd, Port}} = py_nif:create_test_tcp_listener(0),

%% Connect client

{ok, ClientFd} = py_nif:connect_test_tcp(<<"127.0.0.1">>, Port),

true = is_integer(ClientFd),

%% Give connection time to complete (non-blocking connect)

timer:sleep(50),

%% Accept on server side

{ok, ServerFd} = py_nif:accept_test_tcp(ListenFd),

true = is_integer(ServerFd),

%% Cleanup

py_nif:close_test_fd(ServerFd),

py_nif:close_test_fd(ClientFd),

py_nif:close_test_fd(ListenFd),

ok.

test_tcp_read_write_callback(_Config) ->

{ok, LoopRef} = py_nif:event_loop_new(),

{ok, WorkerPid} = py_event_worker:start_link(<<"test">>, LoopRef),

ok = py_nif:event_loop_set_worker(LoopRef, WorkerPid),

%% Create listener and connect

{ok, {ListenFd, Port}} = py_nif:create_test_tcp_listener(0),

{ok, ClientFd} = py_nif:connect_test_tcp(<<"127.0.0.1">>, Port),

timer:sleep(50),

{ok, ServerFd} = py_nif:accept_test_tcp(ListenFd),

%% Register server fd for read monitoring

ReadCallbackId = 100,

{ok, FdRef} = py_nif:add_reader(LoopRef, ServerFd, ReadCallbackId),

%% Write data from client

ok = py_nif:write_test_fd(ClientFd, <<"hello tcp">>),

%% Wait for callback dispatch

timer:sleep(100),

%% Get pending events

Pending = py_nif:get_pending(LoopRef),

%% Should have received the read callback

HasReadCallback = lists:any(

fun({CId, Type}) -> CId == ReadCallbackId andalso Type == read end,

Pending

),

true = HasReadCallback,

%% Cleanup - remove reader before closing fds

ok = py_nif:remove_reader(LoopRef, FdRef),

py_nif:close_test_fd(ServerFd),

py_nif:close_test_fd(ClientFd),

py_nif:close_test_fd(ListenFd),

py_event_worker:stop(WorkerPid),

py_nif:event_loop_destroy(LoopRef),

ok.

test_tcp_echo(_Config) ->

{ok, LoopRef} = py_nif:event_loop_new(),

{ok, WorkerPid} = py_event_worker:start_link(<<"test">>, LoopRef),

ok = py_nif:event_loop_set_worker(LoopRef, WorkerPid),

%% Create listener and connect

{ok, {ListenFd, Port}} = py_nif:create_test_tcp_listener(0),

{ok, ClientFd} = py_nif:connect_test_tcp(<<"127.0.0.1">>, Port),

timer:sleep(50),

{ok, ServerFd} = py_nif:accept_test_tcp(ListenFd),

%% Register both for read monitoring

ClientReadId = 200,

ServerReadId = 201,

{ok, ServerFdRef} = py_nif:add_reader(LoopRef, ServerFd, ServerReadId),

{ok, ClientFdRef} = py_nif:add_reader(LoopRef, ClientFd, ClientReadId),

%% Client sends data

TestData = <<"echo test data">>,

ok = py_nif:write_test_fd(ClientFd, TestData),

%% Wait for server to receive

timer:sleep(100),

%% Server reads and echoes back

{ok, ReceivedData} = py_nif:read_test_fd(ServerFd, 1024),

ok = py_nif:write_test_fd(ServerFd, ReceivedData),

%% Wait for client to receive echo

timer:sleep(100),

%% Client reads echo

{ok, EchoData} = py_nif:read_test_fd(ClientFd, 1024),

%% Verify echo matches original

TestData = EchoData,

%% Verify we got the callbacks

Pending = py_nif:get_pending(LoopRef),

HasServerCallback = lists:any(

fun({CId, Type}) -> CId == ServerReadId andalso Type == read end,

Pending

),

HasClientCallback = lists:any(

fun({CId, Type}) -> CId == ClientReadId andalso Type == read end,

Pending

),

%% At least server callback should have fired

true = HasServerCallback orelse HasClientCallback,

%% Cleanup - remove readers before closing fds

ok = py_nif:remove_reader(LoopRef, ServerFdRef),

ok = py_nif:remove_reader(LoopRef, ClientFdRef),

py_nif:close_test_fd(ServerFd),

py_nif:close_test_fd(ClientFd),

py_nif:close_test_fd(ListenFd),

py_event_worker:stop(WorkerPid),

py_nif:event_loop_destroy(LoopRef),

ok.

%% ============================================================================

%% FD Lifecycle Tests

%% ============================================================================

test_cancel_reader_keeps_fd_open(_Config) ->

%% Test that stop_reader stops monitoring but keeps FD open

{ok, LoopRef} = py_nif:event_loop_new(),

{ok, WorkerPid} = py_event_worker:start_link(<<"test">>, LoopRef),

ok = py_nif:event_loop_set_worker(LoopRef, WorkerPid),

%% Create a pipe

{ok, {ReadFd, WriteFd}} = py_nif:create_test_pipe(),

%% Add reader

CallbackId = 100,

{ok, FdRef} = py_nif:add_reader(LoopRef, ReadFd, CallbackId),

%% Stop reader (should NOT close FD) - using new API

ok = py_nif:stop_reader(FdRef),

%% Write to trigger readiness (but no callback since stopped)

ok = py_nif:write_test_fd(WriteFd, <<"test">>),

timer:sleep(50),

%% FD should still be readable since it wasn't closed

{ok, Data} = py_nif:read_test_fd(ReadFd, 1024),

<<"test">> = Data,

%% Verify no callback was dispatched

Pending = py_nif:get_pending(LoopRef),

HasCallback = lists:any(

fun({CId, _Type}) -> CId == CallbackId end,

Pending

),

false = HasCallback,

%% Cleanup - use close_fd since we stopped (remove_reader won't work)

ok = py_nif:close_fd(FdRef),

py_nif:close_test_fd(WriteFd),

py_event_worker:stop(WorkerPid),

py_nif:event_loop_destroy(LoopRef),

ok.

test_cancel_writer_keeps_fd_open(_Config) ->

%% Test that stop_writer stops monitoring but keeps FD open

{ok, LoopRef} = py_nif:event_loop_new(),

{ok, WorkerPid} = py_event_worker:start_link(<<"test">>, LoopRef),

ok = py_nif:event_loop_set_worker(LoopRef, WorkerPid),

%% Create a pipe

{ok, {ReadFd, WriteFd}} = py_nif:create_test_pipe(),

%% Add writer

CallbackId = 200,

{ok, FdRef} = py_nif:add_writer(LoopRef, WriteFd, CallbackId),

%% Stop writer (should NOT close FD) - using new API

ok = py_nif:stop_writer(FdRef),

%% FD should still be writable since it wasn't closed

ok = py_nif:write_test_fd(WriteFd, <<"test">>),

%% Verify we can read what was written

{ok, Data} = py_nif:read_test_fd(ReadFd, 1024),

<<"test">> = Data,

%% Cleanup

ok = py_nif:close_fd(FdRef),

py_nif:close_test_fd(ReadFd),

py_event_worker:stop(WorkerPid),

py_nif:event_loop_destroy(LoopRef),

ok.

test_double_close_is_safe(_Config) ->

%% Test that calling close_fd twice is idempotent (no crash)

{ok, LoopRef} = py_nif:event_loop_new(),

{ok, WorkerPid} = py_event_worker:start_link(<<"test">>, LoopRef),

ok = py_nif:event_loop_set_worker(LoopRef, WorkerPid),

%% Create a pipe

{ok, {ReadFd, _WriteFd}} = py_nif:create_test_pipe(),

%% Add reader

{ok, FdRef} = py_nif:add_reader(LoopRef, ReadFd, 300),

%% Close once

ok = py_nif:close_fd(FdRef),

%% Close again - should be safe (idempotent)

ok = py_nif:close_fd(FdRef),

%% Also test remove_reader after close is safe

ok = py_nif:remove_reader(LoopRef, FdRef),

%% Note: WriteFd is owned by us and should still work

%% But we won't test that since the pipe's read end is closed

py_event_worker:stop(WorkerPid),

py_nif:event_loop_destroy(LoopRef),

ok.

test_cancel_then_reselect(_Config) ->

%% Test that we can stop and then start (resume) monitoring

{ok, LoopRef} = py_nif:event_loop_new(),

{ok, WorkerPid} = py_event_worker:start_link(<<"test">>, LoopRef),

ok = py_nif:event_loop_set_worker(LoopRef, WorkerPid),

%% Create a pipe

{ok, {ReadFd, WriteFd}} = py_nif:create_test_pipe(),

%% Add reader

CallbackId = 400,

{ok, FdRef} = py_nif:add_reader(LoopRef, ReadFd, CallbackId),

%% Stop reader - using new API

ok = py_nif:stop_reader(FdRef),

%% Re-start using start_reader - using new API

ok = py_nif:start_reader(FdRef),

%% Now write to trigger the callback

ok = py_nif:write_test_fd(WriteFd, <<"hello">>),

timer:sleep(100),

%% Should have received the callback since we re-started

Pending = py_nif:get_pending(LoopRef),

HasCallback = lists:any(

fun({CId, Type}) -> CId == CallbackId andalso Type == read end,

Pending

),

true = HasCallback,

%% Cleanup

ok = py_nif:remove_reader(LoopRef, FdRef),

py_nif:close_test_fd(ReadFd),

py_nif:close_test_fd(WriteFd),

py_event_worker:stop(WorkerPid),

py_nif:event_loop_destroy(LoopRef),

ok.