%%% @doc Common Test suite for UDP/Datagram support in the event loop.

%%%

%%% Tests the UDP socket NIFs and integration with the Erlang-native

%%% asyncio event loop.

-module(py_udp_e2e_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_udp_socket_create/1,

test_udp_send_recv/1,

test_udp_echo/1,

test_udp_multiple_clients/1,

test_udp_broadcast_option/1,

test_udp_read_callback/1

]).

all() ->

[

test_udp_socket_create,

test_udp_send_recv,

test_udp_echo,

test_udp_multiple_clients,

test_udp_broadcast_option,

test_udp_read_callback

].

init_per_suite(Config) ->

{ok, _} = application:ensure_all_started(erlang_python),

Config.

end_per_suite(_Config) ->

ok = application:stop(erlang_python),

ok.

init_per_testcase(_TestCase, Config) ->

Config.

end_per_testcase(_TestCase, _Config) ->

ok.

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

%% UDP Socket Creation Tests

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

test_udp_socket_create(_Config) ->

%% Create a UDP socket on an ephemeral port

{ok, {Fd, Port}} = py_nif:create_test_udp_socket(0),

true = is_integer(Fd),

true = Fd > 0,

true = is_integer(Port),

true = Port > 0,

%% Cleanup

py_nif:close_test_fd(Fd),

ok.

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

%% UDP Send/Receive Tests

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

test_udp_send_recv(_Config) ->

%% Create two UDP sockets

{ok, {ServerFd, ServerPort}} = py_nif:create_test_udp_socket(0),

{ok, {ClientFd, _ClientPort}} = py_nif:create_test_udp_socket(0),

%% Send data from client to server

TestData = <<"hello udp">>,

ok = py_nif:sendto_test_udp(ClientFd, TestData, <<"127.0.0.1">>, ServerPort),

%% Give time for data to arrive

timer:sleep(50),

%% Receive on server

{ok, {ReceivedData, {SourceHost, _SourcePort}}} = py_nif:recvfrom_test_udp(ServerFd, 1024),

%% Verify data

TestData = ReceivedData,

<<"127.0.0.1">> = SourceHost,

%% Cleanup

py_nif:close_test_fd(ServerFd),

py_nif:close_test_fd(ClientFd),

ok.

test_udp_echo(_Config) ->

%% Create server and client sockets

{ok, {ServerFd, ServerPort}} = py_nif:create_test_udp_socket(0),

{ok, {ClientFd, _ClientPort}} = py_nif:create_test_udp_socket(0),

%% Client sends data to server

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

ok = py_nif:sendto_test_udp(ClientFd, TestData, <<"127.0.0.1">>, ServerPort),

%% Give time for data to arrive

timer:sleep(50),

%% Server receives and echoes back

{ok, {ReceivedData, {SourceHost, SourcePort}}} = py_nif:recvfrom_test_udp(ServerFd, 1024),

TestData = ReceivedData,

%% Echo back to client

ok = py_nif:sendto_test_udp(ServerFd, ReceivedData, SourceHost, SourcePort),

%% Give time for echo to arrive

timer:sleep(50),

%% Client receives echo

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

%% Verify echo matches original

TestData = EchoData,

%% Cleanup

py_nif:close_test_fd(ServerFd),

py_nif:close_test_fd(ClientFd),

ok.

test_udp_multiple_clients(_Config) ->

%% Create server socket

{ok, {ServerFd, ServerPort}} = py_nif:create_test_udp_socket(0),

%% Create multiple client sockets

{ok, {Client1Fd, _Client1Port}} = py_nif:create_test_udp_socket(0),

{ok, {Client2Fd, _Client2Port}} = py_nif:create_test_udp_socket(0),

{ok, {Client3Fd, _Client3Port}} = py_nif:create_test_udp_socket(0),

%% Send data from each client

ok = py_nif:sendto_test_udp(Client1Fd, <<"client1">>, <<"127.0.0.1">>, ServerPort),

ok = py_nif:sendto_test_udp(Client2Fd, <<"client2">>, <<"127.0.0.1">>, ServerPort),

ok = py_nif:sendto_test_udp(Client3Fd, <<"client3">>, <<"127.0.0.1">>, ServerPort),

%% Receive all messages on server with retry for slow CI

{ok, {Data1, {_, Port1}}} = recv_with_retry(ServerFd, 1024, 10),

{ok, {Data2, {_, Port2}}} = recv_with_retry(ServerFd, 1024, 10),

{ok, {Data3, {_, Port3}}} = recv_with_retry(ServerFd, 1024, 10),

%% Verify we received all three messages (order may vary)

ReceivedData = lists:sort([Data1, Data2, Data3]),

ExpectedData = lists:sort([<<"client1">>, <<"client2">>, <<"client3">>]),

ExpectedData = ReceivedData,

%% Verify source ports are different (each client has unique port)

Ports = lists:sort([Port1, Port2, Port3]),

3 = length(lists:usort(Ports)),

%% Cleanup

py_nif:close_test_fd(ServerFd),

py_nif:close_test_fd(Client1Fd),

py_nif:close_test_fd(Client2Fd),

py_nif:close_test_fd(Client3Fd),

ok.

%% Helper to retry recvfrom on would_block (for slow CI)

recv_with_retry(Fd, Size, Retries) ->

recv_with_retry(Fd, Size, Retries, 50).

recv_with_retry(_Fd, _Size, 0, _Delay) ->

{error, timeout};

recv_with_retry(Fd, Size, Retries, Delay) ->

case py_nif:recvfrom_test_udp(Fd, Size) of

{error, would_block} ->

timer:sleep(Delay),

recv_with_retry(Fd, Size, Retries - 1, Delay);

Result ->

Result

end.

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

%% UDP Socket Options Tests

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

test_udp_broadcast_option(_Config) ->

%% Create a UDP socket

{ok, {Fd, _Port}} = py_nif:create_test_udp_socket(0),

%% Enable broadcast

ok = py_nif:set_udp_broadcast(Fd, true),

%% Disable broadcast

ok = py_nif:set_udp_broadcast(Fd, false),

%% Test with integer values

ok = py_nif:set_udp_broadcast(Fd, 1),

ok = py_nif:set_udp_broadcast(Fd, 0),

%% Cleanup

py_nif:close_test_fd(Fd),

ok.

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

%% UDP Event Loop Integration Tests

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

test_udp_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 UDP sockets

{ok, {ServerFd, ServerPort}} = py_nif:create_test_udp_socket(0),

{ok, {ClientFd, _ClientPort}} = py_nif:create_test_udp_socket(0),

%% Register server fd for read monitoring

ReadCallbackId = 500,

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

%% Send data from client

ok = py_nif:sendto_test_udp(ClientFd, <<"callback test">>, <<"127.0.0.1">>, ServerPort),

%% 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_event_worker:stop(WorkerPid),

py_nif:event_loop_destroy(LoopRef),

ok.