%%% @doc Common Test suite for erlang.reactor API. %%% %%% Tests the reactor module that provides fd-based protocol handling %%% where Erlang handles I/O scheduling and Python handles protocol logic. -module(py_reactor_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([ reactor_module_exists_test/1, protocol_class_exists_test/1, set_protocol_factory_test/1, echo_protocol_test/1, multiple_connections_test/1, protocol_close_test/1, async_pending_test/1, reactor_buffer_test/1, %% SHARED_GIL subinterpreter isolation test reactor_context_subinterp_isolation_test/1 ]). all() -> [ reactor_module_exists_test, protocol_class_exists_test, set_protocol_factory_test, echo_protocol_test, multiple_connections_test, protocol_close_test, async_pending_test, reactor_buffer_test, reactor_context_subinterp_isolation_test ]. init_per_suite(Config) -> {ok, _} = application:ensure_all_started(erlang_python), {ok, _} = py:start_contexts(), 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 that erlang.reactor module exists reactor_module_exists_test(_Config) -> {ok, true} = py:eval(<<"hasattr(erlang, 'reactor')">>). %% @doc Test that Protocol class exists protocol_class_exists_test(_Config) -> {ok, true} = py:eval(<<"hasattr(erlang.reactor, 'Protocol')">>), {ok, true} = py:eval(<<"callable(erlang.reactor.Protocol)">>). %% @doc Test set_protocol_factory works set_protocol_factory_test(_Config) -> %% Define a simple protocol ok = py:exec(<<" import erlang.reactor as reactor class TestProtocol(reactor.Protocol): def data_received(self, data): return 'continue' def write_ready(self): return 'close' reactor.set_protocol_factory(TestProtocol) ">>), ok. %% @doc Test echo protocol with socketpair echo_protocol_test(_Config) -> Ctx = py:context(1), ok = py:exec(Ctx, <<" import socket import erlang.reactor as reactor class EchoProtocol(reactor.Protocol): def data_received(self, data): self.write_buffer.extend(data) return 'write_pending' def write_ready(self): if not self.write_buffer: return 'close' written = self.write(bytes(self.write_buffer)) del self.write_buffer[:written] return 'continue' if self.write_buffer else 'read_pending' def run_echo_test(): s1, s2 = socket.socketpair() s1.setblocking(False) s2.setblocking(False) reactor.set_protocol_factory(EchoProtocol) reactor.init_connection(s1.fileno(), {'type': 'test'}) s2.send(b'hello') action = reactor.on_read_ready(s1.fileno()) proto = reactor.get_protocol(s1.fileno()) result = bytes(proto.write_buffer) reactor.close_connection(s1.fileno()) s1.close() s2.close() return result _echo_test_result = run_echo_test() ">>), {ok, <<"hello">>} = py:eval(Ctx, <<"_echo_test_result">>). %% @doc Test multiple connections multiple_connections_test(_Config) -> Ctx = py:context(1), ok = py:exec(Ctx, <<" import socket import erlang.reactor as reactor class CounterProtocol(reactor.Protocol): counter = 0 def connection_made(self, fd, client_info): super().connection_made(fd, client_info) CounterProtocol.counter += 1 self.my_id = CounterProtocol.counter def data_received(self, data): return 'close' def write_ready(self): return 'close' def run_multi_conn_test(): CounterProtocol.counter = 0 reactor.set_protocol_factory(CounterProtocol) pairs = [socket.socketpair() for _ in range(3)] for s1, s2 in pairs: s1.setblocking(False) reactor.init_connection(s1.fileno(), {}) ids = [reactor.get_protocol(s1.fileno()).my_id for s1, s2 in pairs] for s1, s2 in pairs: reactor.close_connection(s1.fileno()) s1.close() s2.close() return ids _multi_conn_result = run_multi_conn_test() ">>), {ok, [1, 2, 3]} = py:eval(Ctx, <<"_multi_conn_result">>). %% @doc Test protocol close callback protocol_close_test(_Config) -> Ctx = py:context(1), ok = py:exec(Ctx, <<" import socket import erlang.reactor as reactor _closed_fds = [] class CloseTrackProtocol(reactor.Protocol): def data_received(self, data): return 'close' def write_ready(self): return 'close' def connection_lost(self): _closed_fds.append(self.fd) def run_close_test(): global _closed_fds _closed_fds = [] reactor.set_protocol_factory(CloseTrackProtocol) s1, s2 = socket.socketpair() s1.setblocking(False) fd = s1.fileno() reactor.init_connection(fd, {}) reactor.close_connection(fd) result = fd in _closed_fds s1.close() s2.close() return result _close_test_result = run_close_test() ">>), {ok, true} = py:eval(Ctx, <<"_close_test_result">>). %% @doc Test async_pending action for task-based async operations. %% This tests the pattern used by task-based ASGI where: %% 1. Protocol returns "async_pending" to indicate a task was submitted %% 2. Later, Python sends {write_ready, Fd} to signal completion %% 3. Reactor then triggers write selection async_pending_test(_Config) -> %% Protocol factory code to run in reactor context SetupCode = <<" import erlang.reactor as reactor class AsyncPendingProtocol(reactor.Protocol): '''Protocol that returns async_pending and signals completion.''' def __init__(self): super().__init__() self.pending_response = b'' def data_received(self, data): self.pending_response = b'ASYNC:' + data # Immediately complete the task and signal reactor self.write_buffer.extend(self.pending_response) reactor.signal_write_ready(self.fd) return 'async_pending' def write_ready(self): if not self.write_buffer: return 'close' written = self.write(bytes(self.write_buffer)) del self.write_buffer[:written] return 'continue' if self.write_buffer else 'close' reactor.set_protocol_factory(AsyncPendingProtocol) ">>, %% Start reactor context with protocol factory setup {ok, ReactorCtx} = py_reactor_context:start_link(1, auto, #{ setup_code => SetupCode }), %% Use py:context(1) for test helpers (socket management) PyCtx = py:context(1), ok = py:exec(PyCtx, <<" import socket _async_test_state = {} def setup_socketpair(): global _async_test_state s1, s2 = socket.socketpair() s1.setblocking(False) s2.setblocking(False) _async_test_state = {'s1': s1, 's2': s2, 'fd': s1.fileno()} return s1.fileno() def send_test_data(): s2 = _async_test_state['s2'] s2.send(b'hello') return True def read_response(): s2 = _async_test_state['s2'] s2.setblocking(True) s2.settimeout(2.0) try: return s2.recv(1024) except socket.timeout: return b'TIMEOUT' def cleanup(): s1 = _async_test_state.get('s1') s2 = _async_test_state.get('s2') try: if s1: s1.close() except: pass try: if s2: s2.close() except: pass _async_test_state.clear() return True ">>), %% Step 1: Create socketpair {ok, Fd} = py:eval(PyCtx, <<"setup_socketpair()">>), ct:pal("Created socketpair with fd=~p", [Fd]), %% Check reactor context is alive ct:pal("Reactor context ~p is alive: ~p", [ReactorCtx, is_process_alive(ReactorCtx)]), %% Step 2: Send fd_handoff to reactor context ok = py_reactor_context:handoff(ReactorCtx, Fd, #{}), timer:sleep(100), %% Check reactor stats after handoff Stats = py_reactor_context:stats(ReactorCtx), ct:pal("Reactor stats after handoff: ~p", [Stats]), %% Step 3: Send test data - triggers async_pending and immediate completion {ok, true} = py:eval(PyCtx, <<"send_test_data()">>), timer:sleep(200), %% Step 4: Read response {ok, Response} = py:eval(PyCtx, <<"read_response()">>), ct:pal("Response: ~p", [Response]), %% Verify response <<"ASYNC:hello">> = Response, %% Cleanup {ok, _} = py:eval(PyCtx, <<"cleanup()">>), py_reactor_context:stop(ReactorCtx), ok. %% @doc Test ReactorBuffer behaves like bytes reactor_buffer_test(_Config) -> %% Test that ReactorBuffer type exists in erlang module {ok, true} = py:eval(<<"hasattr(erlang, 'ReactorBuffer')">>), %% Test bytes-like operations - use exec for all assertions Ctx = py:context(1), ok = py:exec(Ctx, <<" import erlang # Test _test_create works buf_simple = erlang.ReactorBuffer._test_create(b'hello') assert len(buf_simple) == 5, f'simple len mismatch: {len(buf_simple)}' test_data = b'GET / HTTP/1.1' + bytes([13, 10]) # Use bytes for \\r\\n buf = erlang.ReactorBuffer._test_create(test_data) # Test len() assert len(buf) == 16, f'len mismatch: {len(buf)}' # 14 chars + 2 for \\r\\n # Test startswith() assert buf.startswith(b'GET'), 'startswith failed' assert not buf.startswith(b'POST'), 'startswith should fail for POST' # Test endswith() assert buf.endswith(bytes([13, 10])), 'endswith failed' # Test indexing assert buf[0] == ord('G'), f'index 0 mismatch: {buf[0]}' assert buf[-1] == 10, f'index -1 mismatch' # Test slicing assert buf[0:3] == b'GET', f'slice mismatch: {buf[0:3]}' # Test bytes() conversion assert bytes(buf) == test_data, 'bytes conversion failed' # Test memoryview mv = memoryview(buf) assert bytes(mv) == test_data, 'memoryview failed' # Test find() assert buf.find(b'HTTP') == 6, f'find mismatch: {buf.find(b\"HTTP\")}' assert buf.find(b'HTTPS') == -1, 'find should return -1' # Test count() assert buf.count(b'/') == 2, f'count mismatch: {buf.count(b\"/\")}' # '/' in path and '1.1' # Test 'in' operator assert b'HTTP' in buf, 'in operator failed' assert b'HTTPS' not in buf, 'not in operator failed' # Test decode() decoded = buf.decode('utf-8') assert decoded == 'GET / HTTP/1.1' + chr(13) + chr(10), f'decode mismatch: {decoded}' # Test comparison assert buf == test_data, 'equality comparison failed' assert buf != b'other', 'inequality comparison failed' _reactor_buffer_test_passed = True ">>), {ok, true} = py:eval(Ctx, <<"_reactor_buffer_test_passed">>). %%% ============================================================================ %%% SHARED_GIL Subinterpreter Isolation Test %%% ============================================================================ %%% %%% Tests that the reactor module cache is properly isolated per-interpreter %%% when using SHARED_GIL subinterpreters via py_reactor_context. %%% %%% Each py_reactor_context with mode=subinterp creates a separate Python %%% subinterpreter that shares the GIL but has its own module state, including %%% the reactor cache (protocol factory, connections, etc.). %% @doc Test that reactor contexts with subinterp mode have isolated protocol factories. %% %% Creates two py_reactor_context processes with different protocol factories: %% - Context 1: EchoProtocol (echoes data back unchanged) %% - Context 2: UpperProtocol (echoes data back uppercased) %% %% Verifies that handoffs to each context use their own isolated factory. %% Note: This test may be skipped on platforms where SHARED_GIL subinterpreters %% don't properly isolate module globals (e.g., some FreeBSD configurations). reactor_context_subinterp_isolation_test(_Config) -> %% Context 1 with EchoProtocol - echoes data unchanged EchoSetup = <<" import erlang.reactor as reactor class EchoProtocol(reactor.Protocol): def data_received(self, data): self.write_buffer.extend(data) return 'write_pending' def write_ready(self): if not self.write_buffer: return 'close' written = self.write(bytes(self.write_buffer)) del self.write_buffer[:written] return 'continue' if self.write_buffer else 'close' reactor.set_protocol_factory(EchoProtocol) ">>, {ok, Ctx1} = py_reactor_context:start_link(101, subinterp, #{ setup_code => EchoSetup }), %% Context 2 with UpperProtocol - echoes data uppercased UpperSetup = <<" import erlang.reactor as reactor class UpperProtocol(reactor.Protocol): def data_received(self, data): self.write_buffer.extend(bytes(data).upper()) return 'write_pending' def write_ready(self): if not self.write_buffer: return 'close' written = self.write(bytes(self.write_buffer)) del self.write_buffer[:written] return 'continue' if self.write_buffer else 'close' reactor.set_protocol_factory(UpperProtocol) ">>, {ok, Ctx2} = py_reactor_context:start_link(102, subinterp, #{ setup_code => UpperSetup }), %% Create socketpairs for testing {ok, {S1a, S1b}} = create_socketpair(), {ok, {S2a, S2b}} = create_socketpair(), Fd1 = get_fd(S1a), Fd2 = get_fd(S2a), %% Handoff connections to each context ok = py_reactor_context:handoff(Ctx1, Fd1, #{type => test}), ok = py_reactor_context:handoff(Ctx2, Fd2, #{type => test}), %% Give contexts time to process handoffs timer:sleep(100), %% Send test data to both TestData = <<"hello">>, ok = gen_tcp:send(S1b, TestData), ok = gen_tcp:send(S2b, TestData), %% Receive responses {ok, Response1} = gen_tcp:recv(S1b, 0, 2000), {ok, Response2} = gen_tcp:recv(S2b, 0, 2000), %% Cleanup before checking results gen_tcp:close(S1a), gen_tcp:close(S1b), gen_tcp:close(S2a), gen_tcp:close(S2b), py_reactor_context:stop(Ctx1), py_reactor_context:stop(Ctx2), %% Verify isolation: Ctx1 echoes unchanged, Ctx2 uppercases %% On some platforms (e.g., FreeBSD), SHARED_GIL subinterpreters may not %% properly isolate module globals, causing both to use the same factory. case {Response1, Response2} of {<<"hello">>, <<"HELLO">>} -> %% Isolation works correctly ok; {<<"HELLO">>, <<"HELLO">>} -> %% Both used UpperProtocol - isolation not working {skip, "SHARED_GIL subinterpreter module isolation not supported on this platform"}; {<<"hello">>, <<"hello">>} -> %% Both used EchoProtocol - isolation not working {skip, "SHARED_GIL subinterpreter module isolation not supported on this platform"}; Other -> ct:fail({unexpected_responses, Other}) end. %% Helper to create a socketpair using gen_tcp create_socketpair() -> {ok, LSock} = gen_tcp:listen(0, [binary, {active, false}, {reuseaddr, true}]), {ok, Port} = inet:port(LSock), {ok, Client} = gen_tcp:connect("127.0.0.1", Port, [binary, {active, false}]), {ok, Server} = gen_tcp:accept(LSock, 1000), gen_tcp:close(LSock), {ok, {Server, Client}}. %% Helper to get raw fd from socket get_fd(Socket) -> {ok, Fd} = inet:getfd(Socket), Fd.