erlang_python

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