erlang_python

%%% @doc Common Test suite for Python thread callback support.
%%%
%%% Tests that Python threads (both threading.Thread and ThreadPoolExecutor)
%%% can call erlang.call() without blocking.
-module(py_thread_callback_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([
    threadpool_basic_test/1,
    threadpool_concurrent_test/1,
    threadpool_multiple_calls_test/1,
    threadpool_nested_callback_test/1,
    threadpool_error_handling_test/1,
    threadpool_thread_reuse_test/1,
    simple_thread_basic_test/1,
    simple_thread_multiple_calls_test/1,
    simple_thread_concurrent_test/1
]).

all() ->
    [
        threadpool_basic_test,
        threadpool_concurrent_test,
        threadpool_multiple_calls_test,
        threadpool_nested_callback_test,
        threadpool_error_handling_test,
        threadpool_thread_reuse_test,
        simple_thread_basic_test,
        simple_thread_multiple_calls_test,
        simple_thread_concurrent_test
    ].

init_per_suite(Config) ->
    {ok, _} = application:ensure_all_started(erlang_python),
    Config.

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

init_per_testcase(_TestCase, Config) ->
    Config.

end_per_testcase(_TestCase, _Config) ->
    %% Cleanup any registered functions
    catch py:unregister_function(double_it),
    catch py:unregister_function(add_one),
    catch py:unregister_function(call_python_square),
    catch py:unregister_function(square_in_erlang),
    catch py:unregister_function(maybe_fail),
    catch py:unregister_function(get_id),
    ok.

%%% ============================================================================
%%% Test Cases
%%% ============================================================================

%% @doc Basic: Single thread calling erlang.call()
threadpool_basic_test(_Config) ->
    py:register_function(double_it, fun([X]) -> X * 2 end),

    %% Use a lambda that creates executor, submits task, gets result
    %% Need fromlist to properly import concurrent.futures submodule
    Code = <<"(lambda cf, erl: cf.ThreadPoolExecutor(max_workers=1).__enter__().submit(lambda: erl.call('double_it', 5)).result())(__import__('concurrent.futures', fromlist=['ThreadPoolExecutor']), __import__('erlang'))">>,
    {ok, Result} = py:eval(Code),
    10 = Result,
    ok.

%% @doc Concurrent: Multiple threads calling simultaneously
threadpool_concurrent_test(_Config) ->
    py:register_function(double_it, fun([X]) -> X * 2 end),

    %% Use executor.map to run calls in parallel
    Code = <<"(lambda cf, erl: list(cf.ThreadPoolExecutor(max_workers=4).__enter__().map(lambda x: erl.call('double_it', x), range(20))))(__import__('concurrent.futures', fromlist=['ThreadPoolExecutor']), __import__('erlang'))">>,
    {ok, Results} = py:eval(Code),
    Expected = [X * 2 || X <- lists:seq(0, 19)],
    Expected = Results,
    ok.

%% @doc Multiple calls: Same thread makes multiple erlang.call() invocations
threadpool_multiple_calls_test(_Config) ->
    py:register_function(add_one, fun([X]) -> X + 1 end),

    %% Lambda that makes 3 sequential calls
    Code = <<"(lambda cf, erl: list(cf.ThreadPoolExecutor(max_workers=2).__enter__().map(lambda x: erl.call('add_one', erl.call('add_one', erl.call('add_one', x))), range(5))))(__import__('concurrent.futures', fromlist=['ThreadPoolExecutor']), __import__('erlang'))">>,
    {ok, Results} = py:eval(Code),
    Expected = [X + 3 || X <- lists:seq(0, 4)],
    Expected = Results,
    ok.

%% @doc Nested: erlang.call() result used in further computation
%% Note: Full nesting (erlang.call -> py:eval) is complex due to worker pool dynamics.
%% This test verifies the callback result can be used for further Python computation.
threadpool_nested_callback_test(_Config) ->
    %% Register Erlang function that computes square purely in Erlang
    py:register_function(square_in_erlang, fun([X]) -> X * X end),

    %% Call from thread pool, then use result in Python computation
    %% This tests that the result is properly returned and usable
    Code = <<"(lambda cf, erl: list(cf.ThreadPoolExecutor(max_workers=2).__enter__().map(lambda x: erl.call('square_in_erlang', x) + 1, range(5))))(__import__('concurrent.futures', fromlist=['ThreadPoolExecutor']), __import__('erlang'))">>,
    {ok, Results} = py:eval(Code),
    %% Results should be x^2 + 1 for x in 0..4
    Expected = [(X * X) + 1 || X <- lists:seq(0, 4)],
    Expected = Results,
    ok.

%% @doc Error handling: Callback raises exception
threadpool_error_handling_test(_Config) ->
    py:register_function(maybe_fail, fun([X]) ->
        case X rem 2 of
            0 -> X * 2;
            1 -> error(odd_number)
        end
    end),

    %% Lambda that catches exceptions - only calls maybe_fail for even numbers
    Code = <<"(lambda cf, erl: list(cf.ThreadPoolExecutor(max_workers=2).__enter__().map(lambda x: erl.call('maybe_fail', x) if x % 2 == 0 else 'error', range(4))))(__import__('concurrent.futures', fromlist=['ThreadPoolExecutor']), __import__('erlang'))">>,
    {ok, Results} = py:eval(Code),
    %% Even numbers succeed (doubled), odd numbers return 'error'
    [0, <<"error">>, 4, <<"error">>] = Results,
    ok.

%% @doc Thread reuse: Verify workers are reused correctly
threadpool_thread_reuse_test(_Config) ->
    py:register_function(get_id, fun([]) -> erlang:unique_integer() end),

    %% Get thread IDs and check that we have <= 2 unique threads
    Code = <<"(lambda cf, th: len(set(cf.ThreadPoolExecutor(max_workers=2).__enter__().map(lambda _: th.get_ident(), range(10)))))(__import__('concurrent.futures', fromlist=['ThreadPoolExecutor']), __import__('threading'))">>,
    {ok, ThreadCount} = py:eval(Code),
    true = (ThreadCount =< 2),
    ok.

%%% ============================================================================
%%% Simple threading.Thread Test Cases
%%% ============================================================================

%% @doc Simple thread calling erlang.call()
simple_thread_basic_test(_Config) ->
    py:register_function(double_it, fun([X]) -> X * 2 end),

    %% Create a threading.Thread subclass that stores its result
    Code = <<"
(lambda: (
    __import__('threading').Thread(target=lambda: setattr(__import__('sys').modules[__name__], '_result', __import__('erlang').call('double_it', 5))).start() or
    __import__('time').sleep(0.1) or
    getattr(__import__('sys').modules[__name__], '_result', None)
))()
">>,
    {ok, Result} = py:eval(Code),
    10 = Result,
    ok.

%% @doc Same simple thread makes multiple erlang.call() invocations
simple_thread_multiple_calls_test(_Config) ->
    py:register_function(add_one, fun([X]) -> X + 1 end),

    %% Thread that makes 3 sequential calls: add_one(add_one(add_one(0)))
    Code = <<"
(lambda: (
    __import__('threading').Thread(target=lambda: setattr(
        __import__('sys').modules[__name__], '_result',
        __import__('erlang').call('add_one',
            __import__('erlang').call('add_one',
                __import__('erlang').call('add_one', 0)))
    )).start() or
    __import__('time').sleep(0.1) or
    getattr(__import__('sys').modules[__name__], '_result', None)
))()
">>,
    {ok, Result} = py:eval(Code),
    3 = Result,
    ok.

%% @doc Multiple simple threads calling erlang.call() concurrently
simple_thread_concurrent_test(_Config) ->
    py:register_function(double_it, fun([X]) -> X * 2 end),

    %% Create 5 threads, each calling double_it with different values
    Code = <<"
(lambda: (
    (threads := [__import__('threading').Thread(target=lambda x=x: setattr(t, 'result', __import__('erlang').call('double_it', x))) for x in range(5) for t in [type('T', (), {'result': None})()]]),
    [setattr(threads[i], 'obj', type('T', (), {'result': None})()) for i in range(5)],
    (workers := []),
    [workers.append(type('Worker', (__import__('threading').Thread,), {'result': None, 'run': lambda self, x=x: setattr(self, 'result', __import__('erlang').call('double_it', x))})()) for x in range(5)],
    [w.start() for w in workers],
    [w.join() for w in workers],
    [w.result for w in workers]
)[-1])()
">>,
    {ok, Results} = py:eval(Code),
    [0, 2, 4, 6, 8] = Results,
    ok.