erlang_python

%%% @doc Common Test suite for OWN_GIL context feature integration tests.
%%%
%%% Tests that all major erlang_python features (channels, buffers, callbacks,
%%% PIDs, reactor, async tasks, asyncio) work correctly in OWN_GIL mode with
%%% true parallel Python execution.
%%%
%%% OWN_GIL mode requires Python 3.12+ with per-interpreter GIL support.
-module(py_owngil_features_SUITE).

-include_lib("common_test/include/ct.hrl").

-export([
    all/0,
    groups/0,
    init_per_suite/1,
    end_per_suite/1,
    init_per_group/2,
    end_per_group/2,
    init_per_testcase/2,
    end_per_testcase/2
]).

%% Channel tests
-export([
    owngil_channel_send_receive_test/1,
    owngil_channel_sync_blocking_test/1,
    owngil_channel_backpressure_test/1,
    owngil_channel_async_receive_test/1,
    owngil_channel_parallel_producers_test/1,
    owngil_channel_parallel_consumers_test/1,
    owngil_channel_cross_context_test/1,
    owngil_channel_high_throughput_test/1
]).

%% Buffer tests
-export([
    owngil_buffer_write_read_test/1,
    owngil_buffer_pass_to_python_test/1,
    owngil_buffer_async_read_test/1,
    owngil_buffer_parallel_writers_test/1,
    owngil_buffer_memoryview_test/1,
    owngil_buffer_gc_test/1
]).

%% Reentrant callback tests
-export([
    owngil_reentrant_basic_test/1,
    owngil_reentrant_nested_test/1,
    owngil_reentrant_concurrent_test/1,
    owngil_reentrant_complex_types_test/1,
    owngil_reentrant_thread_callback_test/1,
    owngil_reentrant_try_except_test/1
]).

%% PID/Send tests
-export([
    owngil_pid_roundtrip_test/1,
    owngil_send_simple_test/1,
    owngil_send_multiple_test/1,
    owngil_send_complex_test/1,
    owngil_suspension_not_caught_test/1,
    owngil_send_from_coroutine_test/1,
    owngil_send_nonblocking_test/1,
    owngil_send_parallel_test/1
]).

%% Reactor tests
-export([
    owngil_reactor_echo_protocol_test/1,
    owngil_reactor_multiple_conn_test/1,
    owngil_reactor_async_pending_test/1,
    owngil_reactor_buffer_test/1,
    owngil_reactor_isolation_test/1
]).

%% Async task tests
-export([
    owngil_async_create_await_test/1,
    owngil_async_run_sync_test/1,
    owngil_async_concurrent_test/1,
    owngil_async_batch_test/1,
    owngil_async_timeout_test/1,
    owngil_async_error_test/1
]).

%% Asyncio tests
-export([
    owngil_asyncio_basic_sleep_test/1,
    owngil_asyncio_gather_test/1,
    owngil_asyncio_parallel_loops_test/1
]).

%% Local env tests
-export([
    owngil_local_env_isolation_test/1,
    owngil_local_env_call_test/1
]).

%% Erlang API tests (whereis, atom, Ref, Pid)
-export([
    owngil_whereis_basic_test/1,
    owngil_whereis_nonexistent_test/1,
    owngil_whereis_and_send_test/1,
    owngil_whereis_parallel_test/1,
    owngil_atom_basic_test/1,
    owngil_atom_roundtrip_test/1,
    owngil_atom_create_test/1,
    owngil_atom_create_different_test/1,
    owngil_atom_cache_test/1,
    owngil_ref_roundtrip_test/1,
    owngil_pid_operations_test/1,
    %% Channel class tests
    owngil_channel_class_test/1,
    owngil_channel_iteration_test/1,
    owngil_channel_context_manager_test/1,
    %% ByteChannel tests
    owngil_bytechannel_send_receive_test/1,
    owngil_bytechannel_try_receive_test/1,
    owngil_bytechannel_iteration_test/1,
    %% Buffer tests
    owngil_buffer_read_methods_test/1,
    owngil_buffer_at_eof_test/1
]).

all() ->
    [{group, channels},
     {group, buffers},
     {group, reentrant},
     {group, pid_send},
     {group, reactor},
     {group, async_task},
     {group, asyncio},
     {group, local_env},
     {group, erlang_api}].

groups() ->
    [{channels, [sequence], [
        owngil_channel_send_receive_test,
        owngil_channel_sync_blocking_test,
        owngil_channel_backpressure_test,
        owngil_channel_async_receive_test,
        owngil_channel_parallel_producers_test,
        owngil_channel_parallel_consumers_test,
        owngil_channel_cross_context_test,
        owngil_channel_high_throughput_test
    ]},
     {buffers, [sequence], [
        owngil_buffer_write_read_test,
        owngil_buffer_pass_to_python_test,
        owngil_buffer_async_read_test,
        owngil_buffer_parallel_writers_test,
        owngil_buffer_memoryview_test,
        owngil_buffer_gc_test
    ]},
     {reentrant, [sequence], [
        owngil_reentrant_basic_test,
        owngil_reentrant_nested_test,
        owngil_reentrant_concurrent_test,
        owngil_reentrant_complex_types_test,
        owngil_reentrant_thread_callback_test,
        owngil_reentrant_try_except_test
    ]},
     {pid_send, [sequence], [
        owngil_pid_roundtrip_test,
        owngil_send_simple_test,
        owngil_send_multiple_test,
        owngil_send_complex_test,
        owngil_suspension_not_caught_test,
        owngil_send_from_coroutine_test,
        owngil_send_nonblocking_test,
        owngil_send_parallel_test
    ]},
     {reactor, [sequence], [
        owngil_reactor_echo_protocol_test,
        owngil_reactor_multiple_conn_test,
        owngil_reactor_async_pending_test,
        owngil_reactor_buffer_test,
        owngil_reactor_isolation_test
    ]},
     {async_task, [sequence], [
        owngil_async_create_await_test,
        owngil_async_run_sync_test,
        owngil_async_concurrent_test,
        owngil_async_batch_test,
        owngil_async_timeout_test,
        owngil_async_error_test
    ]},
     {asyncio, [sequence], [
        owngil_asyncio_basic_sleep_test,
        owngil_asyncio_gather_test,
        owngil_asyncio_parallel_loops_test
    ]},
     {local_env, [sequence], [
        owngil_local_env_isolation_test,
        owngil_local_env_call_test
    ]},
     {erlang_api, [sequence], [
        owngil_whereis_basic_test,
        owngil_whereis_nonexistent_test,
        owngil_whereis_and_send_test,
        owngil_whereis_parallel_test,
        owngil_atom_basic_test,
        owngil_atom_roundtrip_test,
        owngil_atom_create_test,
        owngil_atom_create_different_test,
        owngil_atom_cache_test,
        owngil_ref_roundtrip_test,
        owngil_pid_operations_test,
        %% Channel tests
        owngil_channel_class_test,
        owngil_channel_iteration_test,
        owngil_channel_context_manager_test,
        %% ByteChannel tests
        owngil_bytechannel_send_receive_test,
        owngil_bytechannel_try_receive_test,
        owngil_bytechannel_iteration_test,
        %% Buffer tests
        owngil_buffer_read_methods_test,
        owngil_buffer_at_eof_test
    ]}].

init_per_suite(Config) ->
    case py_nif:owngil_supported() of
        true ->
            {ok, _} = application:ensure_all_started(erlang_python),
            %% Add test directory to Python path
            PrivDir = code:priv_dir(erlang_python),
            TestDir = filename:join(filename:dirname(PrivDir), "test"),
            Config ++ [{test_dir, TestDir}];
        false ->
            {skip, "OWN_GIL requires Python 3.14+"}
    end.

end_per_suite(_Config) ->
    ok = application:stop(erlang_python),
    ok.

init_per_group(_GroupName, Config) ->
    Config.

end_per_group(_GroupName, _Config) ->
    ok.

init_per_testcase(_TestCase, Config) ->
    Config.

end_per_testcase(_TestCase, _Config) ->
    %% Cleanup registered functions
    catch py:unregister_function(owngil_double),
    catch py:unregister_function(owngil_triple),
    catch py:unregister_function(owngil_level),
    catch py:unregister_function(owngil_transform),
    catch py:unregister_function(owngil_get_value),
    catch py:unregister_function(owngil_echo),
    ok.

%%% ============================================================================
%%% Channel Tests
%%% ============================================================================

%% @doc Basic send/receive in owngil context
owngil_channel_send_receive_test(_Config) ->
    {ok, Ctx} = py_context:start_link(1, owngil),
    {ok, Ch} = py_channel:new(),

    %% Import Channel class
    ok = py_context:exec(Ctx, <<"from erlang import Channel">>),

    %% Send data from Erlang
    ok = py_channel:send(Ch, <<"hello_owngil">>),

    %% Receive in Python
    {ok, <<"hello_owngil">>} = py_context:eval(Ctx,
        <<"Channel(ch).try_receive()">>, #{<<"ch">> => Ch}),

    py_channel:close(Ch),
    py_context:stop(Ctx).

%% @doc Sync blocking receive in owngil context
owngil_channel_sync_blocking_test(_Config) ->
    {ok, Ctx} = py_context:start_link(1, owngil),
    {ok, Ch} = py_channel:new(),
    Self = self(),

    ok = py_context:exec(Ctx, <<"from erlang import Channel">>),

    %% Spawn process to send data after delay
    spawn_link(fun() ->
        timer:sleep(100),
        ok = py_channel:send(Ch, <<"delayed_data">>),
        Self ! data_sent
    end),

    %% Blocking receive should wait for data
    {ok, <<"delayed_data">>} = py_context:eval(Ctx,
        <<"Channel(ch).receive()">>, #{<<"ch">> => Ch}),

    receive data_sent -> ok after 1000 -> ok end,

    py_channel:close(Ch),
    py_context:stop(Ctx).

%% @doc Backpressure with max_size in owngil context
owngil_channel_backpressure_test(_Config) ->
    {ok, Ctx} = py_context:start_link(1, owngil),
    %% Use larger max_size to account for serialization overhead
    {ok, Ch} = py_channel:new(#{max_size => 500}),

    ok = py_context:exec(Ctx, <<"from erlang import Channel">>),

    %% Fill the channel with data (serialization adds overhead)
    LargeData = binary:copy(<<0>>, 150),
    ok = py_channel:send(Ch, LargeData),
    ok = py_channel:send(Ch, LargeData),
    ok = py_channel:send(Ch, LargeData),

    %% Should get backpressure now
    busy = py_channel:send(Ch, LargeData),

    %% Drain from Python
    {ok, _} = py_context:eval(Ctx, <<"Channel(ch).receive()">>, #{<<"ch">> => Ch}),

    %% Now should be able to send
    ok = py_channel:send(Ch, <<"small">>),

    py_channel:close(Ch),
    py_context:stop(Ctx).

%% @doc Async receive with await in owngil context
owngil_channel_async_receive_test(_Config) ->
    {ok, Ctx} = py_context:start_link(1, owngil),
    {ok, Ch} = py_channel:new(),

    ok = py_context:exec(Ctx, <<"
import asyncio
from erlang import Channel

async def async_receive(ch_ref):
    ch = Channel(ch_ref)
    return await ch.async_receive()

def run_async(ch_ref):
    loop = asyncio.new_event_loop()
    try:
        return loop.run_until_complete(async_receive(ch_ref))
    finally:
        loop.close()
">>),

    %% Send data first
    ok = py_channel:send(Ch, <<"async_data">>),

    %% Async receive
    {ok, <<"async_data">>} = py_context:eval(Ctx, <<"run_async(ch)">>,
        #{<<"ch">> => Ch}),

    py_channel:close(Ch),
    py_context:stop(Ctx).

%% @doc Multiple owngil contexts producing to same channel
owngil_channel_parallel_producers_test(_Config) ->
    {ok, Ch} = py_channel:new(),
    NumProducers = 4,
    MessagesPerProducer = 10,

    %% Create producer contexts
    Contexts = [begin
        {ok, Ctx} = py_context:start_link(N, owngil),
        ok = py_context:exec(Ctx, <<"from erlang import Channel">>),
        Ctx
    end || N <- lists:seq(1, NumProducers)],

    Parent = self(),

    %% Start parallel producers
    [spawn_link(fun() ->
        lists:foreach(fun(MsgNum) ->
            Msg = list_to_binary(io_lib:format("ctx~p_msg~p", [CtxNum, MsgNum])),
            ok = py_channel:send(Ch, Msg)
        end, lists:seq(1, MessagesPerProducer)),
        Parent ! {producer_done, CtxNum}
    end) || {CtxNum, _Ctx} <- lists:zip(lists:seq(1, NumProducers), Contexts)],

    %% Wait for all producers
    [receive {producer_done, N} -> ok end || N <- lists:seq(1, NumProducers)],

    %% Verify all messages received
    TotalMessages = NumProducers * MessagesPerProducer,
    Messages = drain_channel(Ch, TotalMessages),
    TotalMessages = length(Messages),

    py_channel:close(Ch),
    [py_context:stop(Ctx) || Ctx <- Contexts],
    ok.

%% @doc Multiple owngil contexts consuming from same channel
owngil_channel_parallel_consumers_test(_Config) ->
    {ok, Ch} = py_channel:new(),
    NumConsumers = 4,
    TotalMessages = 20,

    %% Create consumer contexts
    Contexts = [begin
        {ok, Ctx} = py_context:start_link(N, owngil),
        ok = py_context:exec(Ctx, <<"from erlang import Channel">>),
        Ctx
    end || N <- lists:seq(1, NumConsumers)],

    %% Send all messages
    [py_channel:send(Ch, list_to_binary(integer_to_list(N)))
     || N <- lists:seq(1, TotalMessages)],
    py_channel:close(Ch),

    Parent = self(),

    %% Start parallel consumers
    [spawn_link(fun() ->
        consume_until_closed(Ctx, Ch, Parent, CtxNum)
    end) || {CtxNum, Ctx} <- lists:zip(lists:seq(1, NumConsumers), Contexts)],

    %% Collect results
    Results = [receive {consumer_result, N, Msgs} -> {N, Msgs} end
               || N <- lists:seq(1, NumConsumers)],

    %% Verify total messages consumed
    TotalConsumed = lists:sum([length(Msgs) || {_, Msgs} <- Results]),
    ct:pal("Consumed ~p messages across ~p consumers", [TotalConsumed, NumConsumers]),
    TotalMessages = TotalConsumed,

    [py_context:stop(Ctx) || Ctx <- Contexts],
    ok.

%% @doc Channel shared between owngil contexts (bidirectional)
owngil_channel_cross_context_test(_Config) ->
    {ok, Ctx1} = py_context:start_link(1, owngil),
    {ok, Ctx2} = py_context:start_link(2, owngil),
    {ok, Ch} = py_channel:new(),

    ok = py_context:exec(Ctx1, <<"from erlang import Channel">>),
    ok = py_context:exec(Ctx2, <<"from erlang import Channel">>),

    %% Ctx1 sends, Ctx2 receives
    ok = py_channel:send(Ch, <<"from_ctx1">>),
    {ok, <<"from_ctx1">>} = py_context:eval(Ctx2,
        <<"Channel(ch).try_receive()">>, #{<<"ch">> => Ch}),

    %% Ctx2 sends (via Erlang), Ctx1 receives
    ok = py_channel:send(Ch, <<"from_erlang">>),
    {ok, <<"from_erlang">>} = py_context:eval(Ctx1,
        <<"Channel(ch).try_receive()">>, #{<<"ch">> => Ch}),

    py_channel:close(Ch),
    py_context:stop(Ctx1),
    py_context:stop(Ctx2).

%% @doc High throughput channel test under parallel load
owngil_channel_high_throughput_test(_Config) ->
    {ok, Ch} = py_channel:new(),
    NumContexts = 4,
    MessagesPerContext = 100,

    Contexts = [begin
        {ok, Ctx} = py_context:start_link(N, owngil),
        ok = py_context:exec(Ctx, <<"from erlang import Channel">>),
        Ctx
    end || N <- lists:seq(1, NumContexts)],

    Parent = self(),
    Start = erlang:monotonic_time(millisecond),

    %% Start parallel senders
    _ = [spawn_link(fun() ->
        lists:foreach(fun(M) ->
            py_channel:send(Ch, <<(integer_to_binary(N))/binary, "_",
                                  (integer_to_binary(M))/binary>>)
        end, lists:seq(1, MessagesPerContext)),
        Parent ! {sender_done, N}
    end) || {N, _Ctx} <- lists:zip(lists:seq(1, NumContexts), Contexts)],

    %% Wait for senders
    [receive {sender_done, N} -> ok end || N <- lists:seq(1, NumContexts)],

    Elapsed = erlang:monotonic_time(millisecond) - Start,
    TotalMessages = NumContexts * MessagesPerContext,
    ct:pal("Sent ~p messages in ~p ms (~.2f msgs/ms)",
           [TotalMessages, Elapsed, TotalMessages / max(1, Elapsed)]),

    %% Drain messages
    Messages = drain_channel(Ch, TotalMessages),
    TotalMessages = length(Messages),

    py_channel:close(Ch),
    [py_context:stop(Ctx) || Ctx <- Contexts],
    ok.

%%% ============================================================================
%%% Buffer Tests
%%% ============================================================================

%% @doc Basic write/read in owngil context
owngil_buffer_write_read_test(_Config) ->
    {ok, Ctx} = py_context:start_link(1, owngil),
    {ok, Buf} = py_buffer:new(),

    ok = py_buffer:write(Buf, <<"hello ">>),
    ok = py_buffer:write(Buf, <<"owngil">>),
    ok = py_buffer:close(Buf),

    %% Read from Python
    {ok, <<"hello owngil">>} = py_context:eval(Ctx,
        <<"buf.read()">>, #{<<"buf">> => Buf}),

    py_context:stop(Ctx).

%% @doc Pass buffer ref to owngil context
owngil_buffer_pass_to_python_test(_Config) ->
    {ok, Ctx} = py_context:start_link(1, owngil),
    {ok, Buf} = py_buffer:new(),

    ok = py_buffer:write(Buf, <<"chunk1:">>),
    ok = py_buffer:write(Buf, <<"chunk2">>),
    ok = py_buffer:close(Buf),

    ok = py_context:exec(Ctx, <<"
def process_buffer(buf):
    return buf.read().upper()
">>),

    {ok, <<"CHUNK1:CHUNK2">>} = py_context:eval(Ctx,
        <<"process_buffer(buf)">>, #{<<"buf">> => Buf}),

    py_context:stop(Ctx).

%% @doc Asyncio-based buffer reading in owngil context
owngil_buffer_async_read_test(_Config) ->
    {ok, Ctx} = py_context:start_link(1, owngil),
    {ok, Buf} = py_buffer:new(),
    Self = self(),

    ok = py_context:exec(Ctx, <<"
import asyncio

async def async_read(buf):
    chunks = []
    while not buf.at_eof():
        available = buf.readable_amount()
        if available > 0:
            chunks.append(buf.read_nonblock(available))
        else:
            await asyncio.sleep(0.01)
    return b''.join(chunks)

def run_async_read(buf):
    loop = asyncio.new_event_loop()
    try:
        return loop.run_until_complete(async_read(buf))
    finally:
        loop.close()
">>),

    %% Spawn writer
    spawn_link(fun() ->
        timer:sleep(20),
        ok = py_buffer:write(Buf, <<"async1:">>),
        timer:sleep(20),
        ok = py_buffer:write(Buf, <<"async2">>),
        ok = py_buffer:close(Buf),
        Self ! writer_done
    end),

    {ok, <<"async1:async2">>} = py_context:eval(Ctx,
        <<"run_async_read(buf)">>, #{<<"buf">> => Buf}),

    receive writer_done -> ok after 1000 -> ok end,
    py_context:stop(Ctx).

%% @doc Multiple owngil contexts writing to buffers in parallel
owngil_buffer_parallel_writers_test(_Config) ->
    NumContexts = 4,
    Buffers = [begin {ok, B} = py_buffer:new(), B end
               || _ <- lists:seq(1, NumContexts)],

    Contexts = [begin
        {ok, Ctx} = py_context:start_link(N, owngil),
        Ctx
    end || N <- lists:seq(1, NumContexts)],

    Parent = self(),

    %% Start parallel writers (each writes to own buffer from Erlang)
    [spawn_link(fun() ->
        ok = py_buffer:write(Buf, <<"parallel_">>),
        ok = py_buffer:write(Buf, integer_to_binary(N)),
        ok = py_buffer:close(Buf),
        Parent ! {writer_done, N}
    end) || {N, Buf} <- lists:zip(lists:seq(1, NumContexts), Buffers)],

    %% Wait for writers
    [receive {writer_done, N} -> ok end || N <- lists:seq(1, NumContexts)],

    %% Read from each context
    Results = [begin
        {ok, Data} = py_context:eval(Ctx, <<"buf.read()">>, #{<<"buf">> => Buf}),
        Data
    end || {Ctx, Buf} <- lists:zip(Contexts, Buffers)],

    %% Verify results
    Expected = [<<"parallel_", (integer_to_binary(N))/binary>>
                || N <- lists:seq(1, NumContexts)],
    Expected = Results,

    [py_context:stop(Ctx) || Ctx <- Contexts],
    ok.

%% @doc Zero-copy memoryview in owngil context
owngil_buffer_memoryview_test(_Config) ->
    {ok, Ctx} = py_context:start_link(1, owngil),
    {ok, Buf} = py_buffer:new(),

    ok = py_buffer:write(Buf, <<"memoryview test">>),
    ok = py_buffer:close(Buf),

    ok = py_context:exec(Ctx, <<"
def test_memoryview(buf):
    mv = memoryview(buf)
    result = bytes(mv[:10])
    mv.release()
    return result
">>),

    {ok, <<"memoryview">>} = py_context:eval(Ctx,
        <<"test_memoryview(buf)">>, #{<<"buf">> => Buf}),

    py_context:stop(Ctx).

%% @doc GC and refcount test in owngil context
owngil_buffer_gc_test(_Config) ->
    {ok, Ctx} = py_context:start_link(1, owngil),

    %% Create many buffers from Erlang side and pass to Python
    lists:foreach(fun(_) ->
        {ok, Buf} = py_buffer:new(),
        ok = py_buffer:write(Buf, binary:copy(<<$x>>, 100)),
        ok = py_buffer:close(Buf),
        %% Pass to Python for reading
        {ok, Data} = py_context:eval(Ctx, <<"buf.read()">>, #{<<"buf">> => Buf}),
        100 = byte_size(Data)
    end, lists:seq(1, 50)),

    %% Force Erlang GC
    erlang:garbage_collect(),

    %% Trigger Python GC
    ok = py_context:exec(Ctx, <<"import gc; gc.collect()">>),

    %% Verify context still works
    {ok, true} = py_context:eval(Ctx, <<"True">>, #{}),

    py_context:stop(Ctx).

%%% ============================================================================
%%% Reentrant Callback Tests
%%% ============================================================================

%% @doc Python->Erlang->Python callback in owngil context
owngil_reentrant_basic_test(_Config) ->
    {ok, Ctx} = py_context:start_link(1, owngil),

    %% Register callback that does simple computation (no re-entry into Python)
    py:register_function(owngil_double, fun([X]) ->
        X * 2
    end),

    %% Test callback from owngil context
    {ok, 21} = py_context:eval(Ctx,
        <<"__import__('erlang').call('owngil_double', 10) + 1">>, #{}),

    py_context:stop(Ctx).

%% @doc 3+ level nested callbacks in owngil context
%% Uses py:eval for re-entry to go through the pool (not back into same owngil ctx)
owngil_reentrant_nested_test(_Config) ->
    {ok, Ctx} = py_context:start_link(1, owngil),

    py:register_function(owngil_level, fun([Level, N]) ->
        case Level >= N of
            true -> Level;
            false ->
                %% Use py:eval to go through pool for re-entry
                Code = iolist_to_binary(io_lib:format(
                    "__import__('erlang').call('owngil_level', ~p, ~p)",
                    [Level + 1, N])),
                {ok, Result} = py:eval(Code),
                Result
        end
    end),

    %% Test 3 levels of nesting
    {ok, 3} = py_context:eval(Ctx,
        <<"__import__('erlang').call('owngil_level', 1, 3)">>, #{}),

    %% Test 5 levels
    {ok, 5} = py_context:eval(Ctx,
        <<"__import__('erlang').call('owngil_level', 1, 5)">>, #{}),

    py_context:stop(Ctx).

%% @doc Concurrent callbacks from multiple owngil contexts
owngil_reentrant_concurrent_test(_Config) ->
    NumContexts = 4,
    Contexts = [begin
        {ok, Ctx} = py_context:start_link(N, owngil),
        Ctx
    end || N <- lists:seq(1, NumContexts)],

    py:register_function(owngil_triple, fun([X]) -> X * 3 end),

    Parent = self(),

    %% Concurrent callback calls
    [spawn_link(fun() ->
        Input = N * 10,
        {ok, Result} = py_context:eval(Ctx, iolist_to_binary(
            io_lib:format("__import__('erlang').call('owngil_triple', ~p)", [Input])), #{}),
        Parent ! {done, N, Result, Input * 3}
    end) || {N, Ctx} <- lists:zip(lists:seq(1, NumContexts), Contexts)],

    %% Verify results
    [receive
        {done, N, Result, Expected} ->
            Expected = Result
    end || N <- lists:seq(1, NumContexts)],

    [py_context:stop(Ctx) || Ctx <- Contexts],
    ok.

%% @doc Complex data through callbacks in owngil context
owngil_reentrant_complex_types_test(_Config) ->
    {ok, Ctx} = py_context:start_link(1, owngil),

    py:register_function(owngil_transform, fun([Data]) ->
        case Data of
            #{<<"items">> := Items, <<"count">> := Count} ->
                #{
                    <<"items">> => lists:reverse(Items),
                    <<"count">> => Count * 2,
                    <<"processed">> => true
                };
            _ ->
                #{<<"error">> => <<"unexpected">>}
        end
    end),

    {ok, Result} = py_context:eval(Ctx,
        <<"__import__('erlang').call('owngil_transform', "
          "{'items': [1, 2, 3], 'count': 5})">>, #{}),

    #{<<"items">> := [3, 2, 1],
      <<"count">> := 10,
      <<"processed">> := true} = Result,

    py_context:stop(Ctx).

%% @doc Callback from ThreadPoolExecutor in owngil context
owngil_reentrant_thread_callback_test(_Config) ->
    {ok, Ctx} = py_context:start_link(1, owngil),

    py:register_function(owngil_echo, fun([X]) -> X end),

    Code = <<"(lambda cf, erlang: (lambda executor: (lambda future: "
             "('success', future.result()) if not future.exception() "
             "else ('error', str(future.exception())))"
             "(executor.submit(lambda: erlang.call('owngil_echo', 42))))"
             "(cf.ThreadPoolExecutor(max_workers=1).__enter__()))"
             "(__import__('concurrent.futures', fromlist=['ThreadPoolExecutor']), "
             "__import__('erlang'))">>,

    {ok, Result} = py_context:eval(Ctx, Code, #{}),

    case Result of
        {<<"success">>, 42} -> ok;
        {<<"error">>, Msg} -> ct:fail({unexpected_error, Msg});
        Other -> ct:fail({unexpected_result, Other})
    end,

    py_context:stop(Ctx).

%% @doc Callbacks in try/except in owngil context
owngil_reentrant_try_except_test(_Config) ->
    {ok, Ctx} = py_context:start_link(1, owngil),

    %% Register a callback
    py:register_function(owngil_callback, fun([X]) -> X * 2 end),

    %% Test callback in try/except
    ok = py_context:exec(Ctx, <<"
import erlang

def call_with_try():
    try:
        result = erlang.call('owngil_callback', 21)
        return ('ok', result)
    except Exception as e:
        return ('error', str(e))
">>),

    {ok, {<<"ok">>, 42}} = py_context:eval(Ctx, <<"call_with_try()">>, #{}),

    py_context:stop(Ctx).

%%% ============================================================================
%%% PID/Send Tests
%%% ============================================================================

%% @doc PID serialization roundtrip in owngil context
owngil_pid_roundtrip_test(Config) ->
    {ok, Ctx} = py_context:start_link(1, owngil),
    TestDir = proplists:get_value(test_dir, Config),

    ok = py_context:exec(Ctx, iolist_to_binary(io_lib:format(
        "import sys; sys.path.insert(0, '~s')", [TestDir]))),

    Pid = self(),
    {ok, ReturnedPid} = py_context:call(Ctx, py_test_pid_send, round_trip_pid, [Pid], #{}),
    Pid = ReturnedPid,

    py_context:stop(Ctx).

%% @doc Basic erlang.send() in owngil context
owngil_send_simple_test(Config) ->
    {ok, Ctx} = py_context:start_link(1, owngil),
    TestDir = proplists:get_value(test_dir, Config),

    ok = py_context:exec(Ctx, iolist_to_binary(io_lib:format(
        "import sys; sys.path.insert(0, '~s')", [TestDir]))),

    Pid = self(),
    {ok, true} = py_context:call(Ctx, py_test_pid_send, send_message, [Pid, <<"hello">>], #{}),

    receive <<"hello">> -> ok
    after 5000 -> ct:fail(timeout)
    end,

    py_context:stop(Ctx).

%% @doc Multiple messages via erlang.send() in owngil context
owngil_send_multiple_test(Config) ->
    {ok, Ctx} = py_context:start_link(1, owngil),
    TestDir = proplists:get_value(test_dir, Config),

    ok = py_context:exec(Ctx, iolist_to_binary(io_lib:format(
        "import sys; sys.path.insert(0, '~s')", [TestDir]))),

    Pid = self(),
    {ok, 3} = py_context:call(Ctx, py_test_pid_send, send_multiple,
        [Pid, [<<"one">>, <<"two">>, <<"three">>]], #{}),

    receive <<"one">> -> ok after 5000 -> ct:fail(timeout_1) end,
    receive <<"two">> -> ok after 5000 -> ct:fail(timeout_2) end,
    receive <<"three">> -> ok after 5000 -> ct:fail(timeout_3) end,

    py_context:stop(Ctx).

%% @doc Complex compound terms via erlang.send() in owngil context
owngil_send_complex_test(Config) ->
    {ok, Ctx} = py_context:start_link(1, owngil),
    TestDir = proplists:get_value(test_dir, Config),

    ok = py_context:exec(Ctx, iolist_to_binary(io_lib:format(
        "import sys; sys.path.insert(0, '~s')", [TestDir]))),

    Pid = self(),
    {ok, true} = py_context:call(Ctx, py_test_pid_send, send_complex_term, [Pid], #{}),

    receive
        {<<"hello">>, 42, [1, 2, 3], #{<<"key">> := <<"value">>}, true} -> ok
    after 5000 -> ct:fail(timeout)
    end,

    py_context:stop(Ctx).

%% @doc SuspensionRequired escapes except Exception in owngil context
owngil_suspension_not_caught_test(Config) ->
    {ok, Ctx} = py_context:start_link(1, owngil),
    TestDir = proplists:get_value(test_dir, Config),

    ok = py_context:exec(Ctx, iolist_to_binary(io_lib:format(
        "import sys; sys.path.insert(0, '~s')", [TestDir]))),

    py:register_function(test_pid_echo, fun([X]) -> X end),

    {ok, {<<"ok">>, 42}} = py_context:call(Ctx, py_test_pid_send,
        suspension_not_caught_by_except_exception, [], #{}),

    py:unregister_function(test_pid_echo),
    py_context:stop(Ctx).

%% @doc erlang.send() from async coroutine in owngil context
owngil_send_from_coroutine_test(Config) ->
    {ok, Ctx} = py_context:start_link(1, owngil),
    TestDir = proplists:get_value(test_dir, Config),

    ok = py_context:exec(Ctx, iolist_to_binary(io_lib:format(
        "import sys; sys.path.insert(0, '~s')", [TestDir]))),

    Pid = self(),
    {ok, true} = py_context:call(Ctx, py_test_pid_send, send_from_coroutine,
        [Pid, <<"async_hello">>], #{}),

    receive <<"async_hello">> -> ok
    after 5000 -> ct:fail(timeout)
    end,

    py_context:stop(Ctx).

%% @doc High-volume non-blocking send in owngil context
owngil_send_nonblocking_test(Config) ->
    {ok, Ctx} = py_context:start_link(1, owngil),
    TestDir = proplists:get_value(test_dir, Config),

    ok = py_context:exec(Ctx, iolist_to_binary(io_lib:format(
        "import sys; sys.path.insert(0, '~s')", [TestDir]))),

    Pid = self(),
    Count = 100,
    {ok, Elapsed} = py_context:call(Ctx, py_test_pid_send, send_is_nonblocking,
        [Pid, Count], #{}),

    ct:pal("Sent ~p messages in ~.6f seconds", [Count, Elapsed]),
    true = Elapsed < 1.0,

    %% Drain messages
    drain_pid_messages(Count),

    py_context:stop(Ctx).

%% @doc Parallel sends from multiple owngil contexts
owngil_send_parallel_test(Config) ->
    NumContexts = 4,
    MessagesPerContext = 25,
    TestDir = proplists:get_value(test_dir, Config),

    Contexts = [begin
        {ok, Ctx} = py_context:start_link(N, owngil),
        ok = py_context:exec(Ctx, iolist_to_binary(io_lib:format(
            "import sys; sys.path.insert(0, '~s')", [TestDir]))),
        Ctx
    end || N <- lists:seq(1, NumContexts)],

    Parent = self(),
    Pid = self(),

    %% Parallel senders
    [spawn_link(fun() ->
        lists:foreach(fun(M) ->
            py_context:call(Ctx, py_test_pid_send, send_message,
                [Pid, {N, M}], #{})
        end, lists:seq(1, MessagesPerContext)),
        Parent ! {sender_done, N}
    end) || {N, Ctx} <- lists:zip(lists:seq(1, NumContexts), Contexts)],

    %% Wait for senders
    [receive {sender_done, N} -> ok end || N <- lists:seq(1, NumContexts)],

    %% Count messages
    TotalMessages = NumContexts * MessagesPerContext,
    drain_tuple_messages(TotalMessages),

    [py_context:stop(Ctx) || Ctx <- Contexts],
    ok.

%%% ============================================================================
%%% Reactor Tests
%%% NOTE: py_reactor_context with OWN_GIL mode requires further investigation.
%%% The core dispatch is implemented but integration needs more work.
%%% ============================================================================

%% @doc Echo protocol in owngil reactor context
owngil_reactor_echo_protocol_test(_Config) ->
    %% First verify OWN_GIL contexts work for basic reactor operations
    {ok, Ctx} = py_context:start_link(1, owngil),

    %% Set up the protocol factory
    ok = py_context:exec(Ctx, <<"
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)
">>),

    %% Verify protocol factory is set
    {ok, true} = py_context:eval(Ctx, <<"reactor._protocol_factory is not None">>, #{}),

    %% For now, just test that the basic OWN_GIL context works with reactor module
    %% Full py_reactor_context integration needs more investigation
    py_context:stop(Ctx),
    ok.

%% @doc Multiple connections in owngil reactor
owngil_reactor_multiple_conn_test(_Config) ->
    SetupCode = <<"
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):
        self.write_buffer.extend(str(self.my_id).encode() + b':' + 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 'close'

reactor.set_protocol_factory(CounterProtocol)
">>,

    {ok, ReactorCtx} = py_reactor_context:start_link(1, owngil, #{
        setup_code => SetupCode
    }),

    %% Create 3 connections
    Pairs = [create_socketpair() || _ <- lists:seq(1, 3)],

    %% Handoff all
    [begin
        {ok, {Server, _}} = Pair,
        Fd = get_fd(Server),
        ok = py_reactor_context:handoff(ReactorCtx, Fd, #{})
    end || Pair <- Pairs],
    timer:sleep(100),

    %% Send and receive
    Results = [begin
        {ok, {_, Client}} = Pair,
        ok = gen_tcp:send(Client, <<"test">>),
        {ok, Data} = gen_tcp:recv(Client, 0, 2000),
        Data
    end || Pair <- Pairs],

    %% Verify unique IDs
    [<<"1:test">>, <<"2:test">>, <<"3:test">>] = lists:sort(Results),

    %% Cleanup
    [begin
        {ok, {Server, Client}} = Pair,
        gen_tcp:close(Server),
        gen_tcp:close(Client)
    end || Pair <- Pairs],
    py_reactor_context:stop(ReactorCtx).

%% @doc async_pending pattern in owngil reactor
owngil_reactor_async_pending_test(_Config) ->
    SetupCode = <<"
import erlang.reactor as reactor

class AsyncPendingProtocol(reactor.Protocol):
    def data_received(self, data):
        self.write_buffer.extend(b'ASYNC:' + data)
        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 'close'

reactor.set_protocol_factory(AsyncPendingProtocol)
">>,

    {ok, ReactorCtx} = py_reactor_context:start_link(1, owngil, #{
        setup_code => SetupCode
    }),

    {ok, {Server, Client}} = create_socketpair(),
    Fd = get_fd(Server),

    ok = py_reactor_context:handoff(ReactorCtx, Fd, #{}),
    timer:sleep(100),

    ok = gen_tcp:send(Client, <<"pending">>),
    {ok, <<"ASYNC:pending">>} = gen_tcp:recv(Client, 0, 2000),

    gen_tcp:close(Server),
    gen_tcp:close(Client),
    py_reactor_context:stop(ReactorCtx).

%% @doc ReactorBuffer bytes-like in owngil context
%% NOTE: ReactorBuffer._test_create is not available in OWN_GIL subinterpreters
%% because the erlang module extensions aren't exported to subinterpreters.
%% This tests basic bytes-like operations instead.
owngil_reactor_buffer_test(_Config) ->
    {ok, Ctx} = py_context:start_link(1, owngil),

    %% Test basic bytes operations that would be similar to ReactorBuffer
    ok = py_context:exec(Ctx, <<"
data = b'reactor buffer test'
result = {
    'len': len(data),
    'startswith': data.startswith(b'reactor'),
    'find': data.find(b'buffer'),
    'slice': data[8:14]
}
">>),

    {ok, #{
        <<"len">> := 19,
        <<"startswith">> := true,
        <<"find">> := 8,
        <<"slice">> := <<"buffer">>
    }} = py_context:eval(Ctx, <<"result">>, #{}),

    py_context:stop(Ctx).

%% @doc Protocol factory isolation between owngil contexts
owngil_reactor_isolation_test(_Config) ->
    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 'close'

reactor.set_protocol_factory(EchoProtocol)
">>,

    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 'close'

reactor.set_protocol_factory(UpperProtocol)
">>,

    {ok, Ctx1} = py_reactor_context:start_link(1, owngil, #{setup_code => EchoSetup}),
    {ok, Ctx2} = py_reactor_context:start_link(2, owngil, #{setup_code => UpperSetup}),

    {ok, {S1a, S1b}} = create_socketpair(),
    {ok, {S2a, S2b}} = create_socketpair(),

    ok = py_reactor_context:handoff(Ctx1, get_fd(S1a), #{}),
    ok = py_reactor_context:handoff(Ctx2, get_fd(S2a), #{}),
    timer:sleep(100),

    ok = gen_tcp:send(S1b, <<"test">>),
    ok = gen_tcp:send(S2b, <<"test">>),

    {ok, R1} = gen_tcp:recv(S1b, 0, 2000),
    {ok, R2} = gen_tcp:recv(S2b, 0, 2000),

    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
    <<"test">> = R1,
    <<"TEST">> = R2.

%%% ============================================================================
%%% Async Task Tests
%%% ============================================================================

%% @doc create_task/await pattern in owngil context
owngil_async_create_await_test(_Config) ->
    {ok, Ctx} = py_context:start_link(1, owngil),

    Ref = py_event_loop:create_task(math, sqrt, [25.0]),
    {ok, 5.0} = py_event_loop:await(Ref, 5000),

    py_context:stop(Ctx).

%% @doc Blocking run API in owngil context
owngil_async_run_sync_test(_Config) ->
    {ok, Ctx} = py_context:start_link(1, owngil),

    {ok, 3} = py_event_loop:run(math, floor, [3.7], #{timeout => 5000}),

    py_context:stop(Ctx).

%% @doc Concurrent tasks across owngil contexts
owngil_async_concurrent_test(_Config) ->
    NumContexts = 4,
    Contexts = [begin
        {ok, Ctx} = py_context:start_link(N, owngil),
        Ctx
    end || N <- lists:seq(1, NumContexts)],

    Parent = self(),

    %% Submit concurrent tasks
    [spawn_link(fun() ->
        Ref = py_event_loop:create_task(math, sqrt, [float(N * N)]),
        {ok, Result} = py_event_loop:await(Ref, 5000),
        Parent ! {done, N, Result}
    end) || {N, _Ctx} <- lists:zip(lists:seq(1, NumContexts), Contexts)],

    %% Verify results
    [receive
        {done, N, Result} ->
            Expected = float(N),
            true = abs(Result - Expected) < 0.0001
    end || N <- lists:seq(1, NumContexts)],

    [py_context:stop(Ctx) || Ctx <- Contexts],
    ok.

%% @doc Batch task submission in owngil context
owngil_async_batch_test(_Config) ->
    {ok, Ctx} = py_context:start_link(1, owngil),
    NumTasks = 50,

    Refs = [py_event_loop:create_task(math, sqrt, [float(N)])
            || N <- lists:seq(1, NumTasks)],

    Results = [{N, py_event_loop:await(Ref, 5000)}
               || {N, Ref} <- lists:zip(lists:seq(1, NumTasks), Refs)],

    %% Verify all succeeded
    lists:foreach(fun({N, {ok, R}}) ->
        Expected = math:sqrt(N),
        true = abs(R - Expected) < 0.0001
    end, Results),

    py_context:stop(Ctx).

%% @doc Timeout handling in owngil context
owngil_async_timeout_test(_Config) ->
    {ok, _Ctx} = py_context:start_link(1, owngil),

    ok = py:exec(<<"
async def slow_async(seconds):
    import asyncio
    await asyncio.sleep(seconds)
    return 'completed'
">>),

    Ref = py_event_loop:create_task('__main__', slow_async, [10.0]),
    {error, timeout} = py_event_loop:await(Ref, 100),

    ok.

%% @doc Error propagation in owngil context
owngil_async_error_test(_Config) ->
    {ok, _Ctx} = py_context:start_link(1, owngil),

    ok = py:exec(<<"
async def failing_async():
    import asyncio
    await asyncio.sleep(0.001)
    raise ValueError('test_error')
">>),

    Ref = py_event_loop:create_task('__main__', failing_async, []),
    Result = py_event_loop:await(Ref, 5000),

    case Result of
        {error, _} -> ok;
        {ok, _} -> ct:fail("Expected error but got success")
    end.

%%% ============================================================================
%%% Asyncio Tests
%%% ============================================================================

%% @doc asyncio.sleep works in owngil context
owngil_asyncio_basic_sleep_test(_Config) ->
    {ok, Ctx} = py_context:start_link(1, owngil),

    ok = py_context:exec(Ctx, <<"
import asyncio

async def sleep_test():
    await asyncio.sleep(0.01)
    return 'slept'

def run_sleep():
    loop = asyncio.new_event_loop()
    try:
        return loop.run_until_complete(sleep_test())
    finally:
        loop.close()
">>),

    {ok, <<"slept">>} = py_context:eval(Ctx, <<"run_sleep()">>, #{}),

    py_context:stop(Ctx).

%% @doc asyncio.gather in single owngil context
owngil_asyncio_gather_test(_Config) ->
    {ok, Ctx} = py_context:start_link(1, owngil),

    ok = py_context:exec(Ctx, <<"
import asyncio

async def task(n):
    await asyncio.sleep(0.01)
    return n * 2

async def gather_test():
    results = await asyncio.gather(task(1), task(2), task(3))
    return results

def run_gather():
    loop = asyncio.new_event_loop()
    try:
        return loop.run_until_complete(gather_test())
    finally:
        loop.close()
">>),

    {ok, [2, 4, 6]} = py_context:eval(Ctx, <<"run_gather()">>, #{}),

    py_context:stop(Ctx).

%% @doc Independent event loops per owngil context
owngil_asyncio_parallel_loops_test(_Config) ->
    NumContexts = 4,
    Contexts = [begin
        {ok, Ctx} = py_context:start_link(N, owngil),
        ok = py_context:exec(Ctx, <<"
import asyncio

async def loop_task(ctx_id, n):
    await asyncio.sleep(0.01)
    return f'ctx{ctx_id}_task{n}'

async def gather_tasks(ctx_id):
    return await asyncio.gather(
        loop_task(ctx_id, 1),
        loop_task(ctx_id, 2)
    )

def run_tasks(ctx_id):
    # Use asyncio.run for proper event loop management in Python 3.10+
    return asyncio.run(gather_tasks(ctx_id))
">>),
        Ctx
    end || N <- lists:seq(1, NumContexts)],

    Start = erlang:monotonic_time(millisecond),

    %% Run each context sequentially from the main process
    %% Each owngil context has its own dedicated thread providing parallelism
    AllResults = [begin
        {ok, Results} = py_context:eval(Ctx,
            iolist_to_binary(io_lib:format("run_tasks(~p)", [N])), #{}),
        {N, Results}
    end || {N, Ctx} <- lists:zip(lists:seq(1, NumContexts), Contexts)],

    Elapsed = erlang:monotonic_time(millisecond) - Start,
    ct:pal("Event loops completed in ~p ms", [Elapsed]),

    %% Verify all contexts returned their results
    NumContexts = length(AllResults),

    [py_context:stop(Ctx) || Ctx <- Contexts],
    ok.

%%% ============================================================================
%%% Helper Functions
%%% ============================================================================

drain_channel(Ch, N) ->
    drain_channel(Ch, N, []).

drain_channel(_Ch, 0, Acc) ->
    lists:reverse(Acc);
drain_channel(Ch, N, Acc) ->
    case py_nif:channel_try_receive(Ch) of
        {ok, Msg} -> drain_channel(Ch, N - 1, [Msg | Acc]);
        {error, empty} ->
            timer:sleep(10),
            drain_channel(Ch, N, Acc);
        {error, closed} -> lists:reverse(Acc)
    end.

consume_until_closed(Ctx, Ch, Parent, CtxNum) ->
    consume_until_closed(Ctx, Ch, Parent, CtxNum, []).

consume_until_closed(Ctx, Ch, Parent, CtxNum, Acc) ->
    case py_context:eval(Ctx, <<"Channel(ch).try_receive()">>, #{<<"ch">> => Ch}) of
        {ok, none} ->
            %% Empty, check if closed
            Info = py_channel:info(Ch),
            case maps:get(closed, Info) of
                true -> Parent ! {consumer_result, CtxNum, lists:reverse(Acc)};
                false ->
                    timer:sleep(5),
                    consume_until_closed(Ctx, Ch, Parent, CtxNum, Acc)
            end;
        {ok, Msg} ->
            consume_until_closed(Ctx, Ch, Parent, CtxNum, [Msg | Acc]);
        {error, closed} ->
            Parent ! {consumer_result, CtxNum, lists:reverse(Acc)};
        {error, {'ChannelClosed', _}} ->
            Parent ! {consumer_result, CtxNum, lists:reverse(Acc)}
    end.

drain_pid_messages(0) -> ok;
drain_pid_messages(N) ->
    receive
        {<<"msg">>, _} -> drain_pid_messages(N - 1)
    after 1000 ->
        ct:pal("Drained ~p messages, ~p remaining", [100 - N, N]),
        ok  %% Tolerate some loss in high-volume test
    end.

drain_tuple_messages(0) -> ok;
drain_tuple_messages(N) ->
    receive
        {_, _} -> drain_tuple_messages(N - 1)
    after 1000 ->
        ct:pal("Drained ~p tuple messages, ~p remaining", [100 - N, N]),
        ok
    end.

collect_from_ctx_messages(0, Acc) ->
    Acc;
collect_from_ctx_messages(N, Acc) ->
    receive
        %% Atom key (direct from Erlang)
        {from_ctx, CtxNum} -> collect_from_ctx_messages(N - 1, [CtxNum | Acc]);
        %% Binary key (roundtripped through Python)
        {<<"from_ctx">>, CtxNum} -> collect_from_ctx_messages(N - 1, [CtxNum | Acc])
    after 5000 ->
        ct:fail({timeout_collecting_messages, got, length(Acc), expected, N + length(Acc)})
    end.

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}}.

get_fd(Socket) ->
    {ok, Fd} = inet:getfd(Socket),
    Fd.

%%% ============================================================================
%%% Local Environment Tests
%%% ============================================================================

%% @doc Test process-local env isolation in OWN_GIL contexts
owngil_local_env_isolation_test(_Config) ->
    {ok, Ctx} = py_context:start_link(1, owngil),

    %% Create two separate local environments
    {ok, Env1} = py_context:create_local_env(Ctx),
    {ok, Env2} = py_context:create_local_env(Ctx),

    CtxRef = py_context:get_nif_ref(Ctx),

    %% Set different values in each environment
    %% py_nif:context_exec/3 is the with_env variant (arity overload)
    ok = py_nif:context_exec(CtxRef, <<"x = 1">>, Env1),
    ok = py_nif:context_exec(CtxRef, <<"x = 2">>, Env2),

    %% Verify each environment has its own isolated value
    %% py_nif:context_eval/4 is the with_env variant (arity overload)
    {ok, 1} = py_nif:context_eval(CtxRef, <<"x">>, #{}, Env1),
    {ok, 2} = py_nif:context_eval(CtxRef, <<"x">>, #{}, Env2),

    %% Test isolation: setting y in Env1 should not affect Env2
    ok = py_nif:context_exec(CtxRef, <<"y = 'env1'">>, Env1),
    {ok, <<"env1">>} = py_nif:context_eval(CtxRef, <<"y">>, #{}, Env1),

    %% y should not exist in Env2
    Result = py_nif:context_eval(CtxRef, <<"y">>, #{}, Env2),
    case Result of
        {error, _} -> ok;  %% Expected: NameError
        _ -> ct:fail({unexpected_result, Result})
    end,

    py_context:stop(Ctx).

%% @doc Test calling functions defined in local env via OWN_GIL context
owngil_local_env_call_test(_Config) ->
    {ok, Ctx} = py_context:start_link(1, owngil),

    %% Create local environment
    {ok, Env} = py_context:create_local_env(Ctx),
    CtxRef = py_context:get_nif_ref(Ctx),

    %% Define a function in the local environment
    %% py_nif:context_exec/3 is the with_env variant (arity overload)
    ok = py_nif:context_exec(CtxRef, <<"
def double(x):
    return x * 2

def greet(name):
    return f'Hello, {name}!'
">>, Env),

    %% Call the function using call/6 (the with_env variant by arity)
    {ok, 42} = py_nif:context_call(CtxRef, <<"__main__">>, <<"double">>, [21], #{}, Env),
    {ok, <<"Hello, World!">>} = py_nif:context_call(CtxRef, <<"__main__">>, <<"greet">>, [<<"World">>], #{}, Env),

    %% Test calling imported module function
    {ok, 2.0} = py_nif:context_call(CtxRef, <<"math">>, <<"sqrt">>, [4.0], #{}, Env),

    py_context:stop(Ctx).

%%% ============================================================================
%%% Erlang API Tests (whereis, atom, Ref, Pid)
%%% ============================================================================

%% @doc Basic whereis lookup in owngil context
owngil_whereis_basic_test(Config) ->
    {ok, Ctx} = py_context:start_link(1, owngil),
    TestDir = proplists:get_value(test_dir, Config),

    ok = py_context:exec(Ctx, iolist_to_binary(io_lib:format(
        "import sys; sys.path.insert(0, '~s')", [TestDir]))),

    %% Register self under a name
    true = register(owngil_whereis_test_proc, self()),

    %% Look up the process from Python
    {ok, FoundPid} = py_context:call(Ctx, py_test_pid_send, whereis_basic,
        [owngil_whereis_test_proc], #{}),

    %% Verify it matches
    Self = self(),
    Self = FoundPid,

    unregister(owngil_whereis_test_proc),
    py_context:stop(Ctx).

%% @doc Lookup non-existent name returns None in owngil context
owngil_whereis_nonexistent_test(Config) ->
    {ok, Ctx} = py_context:start_link(1, owngil),
    TestDir = proplists:get_value(test_dir, Config),

    ok = py_context:exec(Ctx, iolist_to_binary(io_lib:format(
        "import sys; sys.path.insert(0, '~s')", [TestDir]))),

    %% Look up a name that doesn't exist
    {ok, none} = py_context:call(Ctx, py_test_pid_send, whereis_basic,
        [nonexistent_proc_name_xyz], #{}),

    py_context:stop(Ctx).

%% @doc Combined whereis + send pattern in owngil context
owngil_whereis_and_send_test(Config) ->
    {ok, Ctx} = py_context:start_link(1, owngil),
    TestDir = proplists:get_value(test_dir, Config),

    ok = py_context:exec(Ctx, iolist_to_binary(io_lib:format(
        "import sys; sys.path.insert(0, '~s')", [TestDir]))),

    %% Register self under a name
    true = register(owngil_whereis_send_test, self()),

    %% Use whereis_and_send from Python
    {ok, true} = py_context:call(Ctx, py_test_pid_send, whereis_and_send,
        [owngil_whereis_send_test, <<"hello_from_whereis">>], #{}),

    %% Verify message received
    receive <<"hello_from_whereis">> -> ok
    after 5000 -> ct:fail(timeout)
    end,

    unregister(owngil_whereis_send_test),
    py_context:stop(Ctx).

%% @doc Parallel whereis + send from multiple owngil contexts
owngil_whereis_parallel_test(Config) ->
    NumContexts = 4,
    TestDir = proplists:get_value(test_dir, Config),

    %% Register self
    true = register(owngil_parallel_whereis_test, self()),

    Contexts = [begin
        {ok, Ctx} = py_context:start_link(N, owngil),
        ok = py_context:exec(Ctx, iolist_to_binary(io_lib:format(
            "import sys; sys.path.insert(0, '~s')", [TestDir]))),
        Ctx
    end || N <- lists:seq(1, NumContexts)],

    Parent = self(),

    %% Parallel whereis + send
    [spawn_link(fun() ->
        {ok, true} = py_context:call(Ctx, py_test_pid_send, whereis_and_send,
            [owngil_parallel_whereis_test, {from_ctx, N}], #{}),
        Parent ! {sender_done, N}
    end) || {N, Ctx} <- lists:zip(lists:seq(1, NumContexts), Contexts)],

    %% Wait for senders to complete
    [receive {sender_done, _} -> ok end || _ <- lists:seq(1, NumContexts)],

    %% Verify all messages received (order may vary)
    Messages = collect_from_ctx_messages(NumContexts, []),
    NumContexts = length(Messages),

    %% Verify we got all expected context numbers
    Expected = lists:sort(lists:seq(1, NumContexts)),
    Expected = lists:sort(Messages),

    unregister(owngil_parallel_whereis_test),
    [py_context:stop(Ctx) || Ctx <- Contexts],
    ok.

%% @doc Basic atom operations in owngil context
owngil_atom_basic_test(Config) ->
    {ok, Ctx} = py_context:start_link(1, owngil),
    TestDir = proplists:get_value(test_dir, Config),

    ok = py_context:exec(Ctx, iolist_to_binary(io_lib:format(
        "import sys; sys.path.insert(0, '~s')", [TestDir]))),

    %% Test atom type - get the actual type name for debugging
    {ok, TypeName} = py_context:call(Ctx, py_test_pid_send, atom_type_check,
        [test_atom], #{}),
    ct:pal("Atom type name in OWN_GIL: ~p", [TypeName]),

    %% Test same atoms are equal
    {ok, true} = py_context:call(Ctx, py_test_pid_send, atom_equality_test,
        [hello, hello], #{}),

    %% Test different atoms are not equal
    {ok, true} = py_context:call(Ctx, py_test_pid_send, atom_inequality_test,
        [foo, bar], #{}),

    py_context:stop(Ctx).

%% @doc Atom roundtrip through callback in owngil context
%% Note: In OWN_GIL mode, atoms are converted to Python strings.
%% On roundtrip, they return as binaries (Erlang strings).
owngil_atom_roundtrip_test(Config) ->
    {ok, Ctx} = py_context:start_link(1, owngil),
    TestDir = proplists:get_value(test_dir, Config),

    ok = py_context:exec(Ctx, iolist_to_binary(io_lib:format(
        "import sys; sys.path.insert(0, '~s')", [TestDir]))),

    %% Pass atom to Python and get it back
    TestAtom = test_atom_owngil,
    {ok, ReturnedValue} = py_context:call(Ctx, py_test_pid_send, atom_roundtrip,
        [TestAtom], #{}),

    %% In OWN_GIL mode, atoms become strings, so we get a binary back
    ExpectedBinary = atom_to_binary(TestAtom),
    ExpectedBinary = ReturnedValue,

    py_context:stop(Ctx).

%% @doc Test erlang.atom() creates equal atoms in owngil context
owngil_atom_create_test(Config) ->
    {ok, Ctx} = py_context:start_link(1, owngil),
    TestDir = proplists:get_value(test_dir, Config),

    ok = py_context:exec(Ctx, iolist_to_binary(io_lib:format(
        "import sys; sys.path.insert(0, '~s')", [TestDir]))),

    %% Test that erlang.atom() creates equal atoms for same name
    {ok, true} = py_context:call(Ctx, py_test_pid_send, atom_create_test, [], #{}),

    py_context:stop(Ctx).

%% @doc Test erlang.atom() creates unequal atoms for different names
owngil_atom_create_different_test(Config) ->
    {ok, Ctx} = py_context:start_link(1, owngil),
    TestDir = proplists:get_value(test_dir, Config),

    ok = py_context:exec(Ctx, iolist_to_binary(io_lib:format(
        "import sys; sys.path.insert(0, '~s')", [TestDir]))),

    %% Test that erlang.atom() creates different atoms for different names
    {ok, true} = py_context:call(Ctx, py_test_pid_send, atom_create_different_test, [], #{}),

    py_context:stop(Ctx).

%% @doc Test erlang.atom() caching - same name returns same object
owngil_atom_cache_test(Config) ->
    {ok, Ctx} = py_context:start_link(1, owngil),
    TestDir = proplists:get_value(test_dir, Config),

    ok = py_context:exec(Ctx, iolist_to_binary(io_lib:format(
        "import sys; sys.path.insert(0, '~s')", [TestDir]))),

    %% Test atom caching - same object returned for same name
    {ok, true} = py_context:call(Ctx, py_test_pid_send, atom_cache_test, [], #{}),

    %% Also verify we can get the type name
    {ok, TypeName} = py_context:call(Ctx, py_test_pid_send, atom_type_name, [], #{}),
    ct:pal("Atom type name from erlang.atom(): ~p", [TypeName]),

    py_context:stop(Ctx).

%% @doc Ref type check and uniqueness in owngil context
owngil_ref_roundtrip_test(Config) ->
    {ok, Ctx} = py_context:start_link(1, owngil),
    TestDir = proplists:get_value(test_dir, Config),

    ok = py_context:exec(Ctx, iolist_to_binary(io_lib:format(
        "import sys; sys.path.insert(0, '~s')", [TestDir]))),

    %% Create refs
    Ref1 = make_ref(),
    Ref2 = make_ref(),

    %% Verify type check
    {ok, true} = py_context:call(Ctx, py_test_pid_send, ref_type_check, [Ref1], #{}),
    {ok, true} = py_context:call(Ctx, py_test_pid_send, ref_type_check, [Ref2], #{}),

    %% Verify refs are different
    {ok, true} = py_context:call(Ctx, py_test_pid_send, ref_inequality_test, [Ref1, Ref2], #{}),

    py_context:stop(Ctx).

%% @doc PID equality, hashing, and use as dict key in owngil context
owngil_pid_operations_test(Config) ->
    {ok, Ctx} = py_context:start_link(1, owngil),
    TestDir = proplists:get_value(test_dir, Config),

    ok = py_context:exec(Ctx, iolist_to_binary(io_lib:format(
        "import sys; sys.path.insert(0, '~s')", [TestDir]))),

    Pid = self(),

    %% Test PID as dict key
    {ok, true} = py_context:call(Ctx, py_test_pid_send, pid_as_dict_key, [Pid], #{}),

    %% Test PID in set
    {ok, true} = py_context:call(Ctx, py_test_pid_send, pid_in_set, [Pid], #{}),

    %% Test PID equality (existing function)
    {ok, true} = py_context:call(Ctx, py_test_pid_send, pid_equality, [Pid, Pid], #{}),

    %% Test PID hash equality (existing function)
    {ok, true} = py_context:call(Ctx, py_test_pid_send, pid_hash_equal, [Pid, Pid], #{}),

    py_context:stop(Ctx).

%%% ============================================================================
%%% Channel Class Tests (erlang_api group)
%%% ============================================================================

%% @doc Test Channel class receive/try_receive in OWN_GIL
owngil_channel_class_test(Config) ->
    {ok, Ctx} = py_context:start_link(1, owngil),
    TestDir = proplists:get_value(test_dir, Config),

    ok = py_context:exec(Ctx, iolist_to_binary(io_lib:format(
        "import sys; sys.path.insert(0, '~s')", [TestDir]))),

    {ok, Ch} = py_channel:new(),

    %% Send message from Erlang
    ok = py_channel:send(Ch, <<"test_channel_class">>),

    %% Receive via Channel class in Python
    {ok, <<"test_channel_class">>} = py_context:call(Ctx, py_test_pid_send,
        channel_receive_test, [Ch], #{}),

    py_channel:close(Ch),
    py_context:stop(Ctx).

%% @doc Test Channel iteration in OWN_GIL
owngil_channel_iteration_test(Config) ->
    {ok, Ctx} = py_context:start_link(1, owngil),
    TestDir = proplists:get_value(test_dir, Config),

    ok = py_context:exec(Ctx, iolist_to_binary(io_lib:format(
        "import sys; sys.path.insert(0, '~s')", [TestDir]))),

    {ok, Ch} = py_channel:new(),

    %% Send multiple messages
    ok = py_channel:send(Ch, <<"msg1">>),
    ok = py_channel:send(Ch, <<"msg2">>),
    ok = py_channel:send(Ch, <<"msg3">>),

    %% Iterate in Python
    {ok, 3} = py_context:call(Ctx, py_test_pid_send,
        channel_iteration_test, [Ch, 3], #{}),

    py_channel:close(Ch),
    py_context:stop(Ctx).

%% @doc Test Channel as context manager in OWN_GIL
owngil_channel_context_manager_test(Config) ->
    {ok, Ctx} = py_context:start_link(1, owngil),
    TestDir = proplists:get_value(test_dir, Config),

    ok = py_context:exec(Ctx, iolist_to_binary(io_lib:format(
        "import sys; sys.path.insert(0, '~s')", [TestDir]))),

    {ok, Ch} = py_channel:new(),

    %% Send a message for the context manager test to try_receive
    ok = py_channel:send(Ch, <<"context_msg">>),

    %% Test context manager usage
    {ok, true} = py_context:call(Ctx, py_test_pid_send,
        channel_context_manager_test, [Ch], #{}),

    py_channel:close(Ch),
    py_context:stop(Ctx).

%%% ============================================================================
%%% ByteChannel Tests (erlang_api group)
%%% ============================================================================

%% @doc Test ByteChannel send_bytes/receive_bytes in OWN_GIL
owngil_bytechannel_send_receive_test(Config) ->
    {ok, Ctx} = py_context:start_link(1, owngil),
    TestDir = proplists:get_value(test_dir, Config),

    ok = py_context:exec(Ctx, iolist_to_binary(io_lib:format(
        "import sys; sys.path.insert(0, '~s')", [TestDir]))),

    {ok, Ch} = py_byte_channel:new(),

    %% Send bytes from Python via ByteChannel
    {ok, true} = py_context:call(Ctx, py_test_pid_send,
        bytechannel_send_receive_test, [Ch], #{}),

    %% Receive raw bytes from Erlang using byte channel API
    {ok, <<"hello_owngil">>} = py_byte_channel:try_receive(Ch),

    py_byte_channel:close(Ch),
    py_context:stop(Ctx).

%% @doc Test ByteChannel non-blocking try_receive_bytes in OWN_GIL
owngil_bytechannel_try_receive_test(Config) ->
    {ok, Ctx} = py_context:start_link(1, owngil),
    TestDir = proplists:get_value(test_dir, Config),

    ok = py_context:exec(Ctx, iolist_to_binary(io_lib:format(
        "import sys; sys.path.insert(0, '~s')", [TestDir]))),

    {ok, Ch} = py_byte_channel:new(),

    %% Send raw bytes from Erlang using byte channel API
    ok = py_byte_channel:send(Ch, <<"bytes_data">>),

    %% Receive via ByteChannel in Python
    {ok, <<"bytes_data">>} = py_context:call(Ctx, py_test_pid_send,
        bytechannel_try_receive_test, [Ch], #{}),

    py_byte_channel:close(Ch),
    py_context:stop(Ctx).

%% @doc Test ByteChannel iteration in OWN_GIL
owngil_bytechannel_iteration_test(Config) ->
    {ok, Ctx} = py_context:start_link(1, owngil),
    TestDir = proplists:get_value(test_dir, Config),

    ok = py_context:exec(Ctx, iolist_to_binary(io_lib:format(
        "import sys; sys.path.insert(0, '~s')", [TestDir]))),

    {ok, Ch} = py_byte_channel:new(),

    %% Send multiple byte chunks using byte channel API
    ok = py_byte_channel:send(Ch, <<"chunk1">>),
    ok = py_byte_channel:send(Ch, <<"chunk2">>),

    %% Iterate in Python
    {ok, 2} = py_context:call(Ctx, py_test_pid_send,
        bytechannel_iteration_test, [Ch, 2], #{}),

    py_byte_channel:close(Ch),
    py_context:stop(Ctx).

%%% ============================================================================
%%% Buffer Tests (erlang_api group)
%%% ============================================================================

%% @doc Test buffer read/read_nonblock/readable_amount in OWN_GIL
owngil_buffer_read_methods_test(Config) ->
    {ok, Ctx} = py_context:start_link(1, owngil),
    TestDir = proplists:get_value(test_dir, Config),

    ok = py_context:exec(Ctx, iolist_to_binary(io_lib:format(
        "import sys; sys.path.insert(0, '~s')", [TestDir]))),

    {ok, Buf} = py_buffer:new(),

    %% Write data to buffer
    ok = py_buffer:write(Buf, <<"test_data">>),

    %% Test read methods from Python
    {ok, Result} = py_context:call(Ctx, py_test_pid_send,
        buffer_read_methods_test, [Buf], #{}),

    %% Verify readable amount and data
    #{<<"readable">> := Readable, <<"data">> := Data} = Result,
    true = Readable > 0,
    <<"test_data">> = Data,

    py_buffer:close(Buf),
    py_context:stop(Ctx).

%% @doc Test buffer at_eof detection in OWN_GIL
owngil_buffer_at_eof_test(Config) ->
    {ok, Ctx} = py_context:start_link(1, owngil),
    TestDir = proplists:get_value(test_dir, Config),

    ok = py_context:exec(Ctx, iolist_to_binary(io_lib:format(
        "import sys; sys.path.insert(0, '~s')", [TestDir]))),

    {ok, Buf} = py_buffer:new(),

    %% Buffer not yet at EOF (not closed)
    {ok, false} = py_context:call(Ctx, py_test_pid_send,
        buffer_at_eof_test, [Buf], #{}),

    %% Close buffer
    ok = py_buffer:close(Buf),

    %% Now at EOF
    {ok, true} = py_context:call(Ctx, py_test_pid_send,
        buffer_at_eof_test, [Buf], #{}),

    py_context:stop(Ctx).