erlang_python

%%% @doc Common Test suite for py module.
-module(py_SUITE).

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

-export([
    all/0,
    init_per_suite/1,
    end_per_suite/1
]).

-export([
    test_call_math/1,
    test_call_json/1,
    test_call_kwargs/1,
    test_eval/1,
    test_eval_complex_locals/1,
    test_exec/1,
    test_cast/1,
    test_cast_kwargs/1,
    test_spawn_call/1,
    test_type_conversions/1,
    test_nested_types/1,
    test_timeout/1,
    test_special_floats/1,
    test_streaming/1,
    test_stream_function/1,
    test_stream_iterators/1,
    test_error_handling/1,
    test_version/1,
    test_memory_stats/1,
    test_gc/1,
    test_erlang_callback/1,
    test_erlang_callback_mfa/1,
    test_erlang_import_syntax/1,
    test_erlang_attr_syntax/1,
    test_asyncio_call/1,
    test_asyncio_gather/1,
    test_asyncio_gather_timeout/1,
    test_subinterp_supported/1,
    test_parallel_execution/1,
    test_venv/1,
    test_venv_pth/1,
    %% New scalability tests
    test_execution_mode/1,
    test_semaphore_basic/1,
    test_semaphore_acquire_release/1,
    test_semaphore_concurrent/1,
    test_semaphore_timeout/1,
    test_semaphore_rate_limiting/1,
    test_overload_protection/1,
    test_shared_state/1,
    test_reload/1,
    %% Process-bound Python environment tests
    test_process_env_isolation/1,
    test_process_env_main_function/1,
    test_process_env_state_persistence/1,
    test_process_env_cleanup/1,
    %% SharedDict tests
    test_shared_dict_basic/1,
    test_shared_dict_types/1,
    test_shared_dict_process_death/1,
    test_shared_dict_python_access/1,
    test_shared_dict_destroy/1,
    %% SharedDict concurrent tests
    test_shared_dict_concurrent_same_key/1,
    test_shared_dict_concurrent_different_keys/1,
    test_shared_dict_concurrent_mixed/1,
    test_shared_dict_benchmark/1
]).

all() ->
    [
        test_call_math,
        test_call_json,
        test_call_kwargs,
        test_eval,
        test_eval_complex_locals,
        test_exec,
        test_cast,
        test_cast_kwargs,
        test_spawn_call,
        test_type_conversions,
        test_nested_types,
        test_timeout,
        test_special_floats,
        test_streaming,
        test_stream_function,
        test_stream_iterators,
        test_error_handling,
        test_version,
        test_memory_stats,
        test_gc,
        test_erlang_callback,
        test_erlang_callback_mfa,
        test_erlang_import_syntax,
        test_erlang_attr_syntax,
        test_asyncio_call,
        test_asyncio_gather,
        test_asyncio_gather_timeout,
        test_subinterp_supported,
        test_parallel_execution,
        test_venv,
        test_venv_pth,
        %% Scalability tests
        test_execution_mode,
        test_semaphore_basic,
        test_semaphore_acquire_release,
        test_semaphore_concurrent,
        test_semaphore_timeout,
        test_semaphore_rate_limiting,
        test_overload_protection,
        test_shared_state,
        test_reload,
        %% Process-bound Python environment tests
        test_process_env_isolation,
        test_process_env_main_function,
        test_process_env_state_persistence,
        test_process_env_cleanup,
        %% SharedDict tests
        test_shared_dict_basic,
        test_shared_dict_types,
        test_shared_dict_process_death,
        test_shared_dict_python_access,
        test_shared_dict_destroy,
        %% SharedDict concurrent tests
        test_shared_dict_concurrent_same_key,
        test_shared_dict_concurrent_different_keys,
        test_shared_dict_concurrent_mixed,
        test_shared_dict_benchmark
    ].

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

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

test_call_math(_Config) ->
    {ok, 4.0} = py:call(math, sqrt, [16]),
    {ok, 120} = py:call(math, factorial, [5]),
    %% Test accessing module attribute via eval - use __import__
    {ok, Pi} = py:eval(<<"__import__('math').pi">>),
    true = Pi > 3.14 andalso Pi < 3.15,
    ok.

test_call_json(_Config) ->
    %% Encode
    {ok, Json} = py:call(json, dumps, [#{foo => bar}]),
    true = is_binary(Json),

    %% Decode
    {ok, Decoded} = py:call(json, loads, [<<"{\"a\": 1, \"b\": [1,2,3]}">>]),
    #{<<"a">> := 1, <<"b">> := [1, 2, 3]} = Decoded,
    ok.

test_call_kwargs(_Config) ->
    %% Test keyword arguments with json.dumps indent parameter
    {ok, JsonIndented} = py:call(json, dumps, [#{a => 1}], #{indent => 2}),
    true = is_binary(JsonIndented),
    %% Indented JSON should contain newlines
    true = binary:match(JsonIndented, <<"\n">>) =/= nomatch,

    %% Test keyword arguments with sort_keys
    {ok, JsonSorted} = py:call(json, dumps, [#{z => 1, a => 2}], #{sort_keys => true}),
    true = is_binary(JsonSorted),

    %% Test multiple kwargs
    {ok, JsonMulti} = py:call(json, dumps, [#{b => 1, a => 2}], #{indent => 4, sort_keys => true}),
    true = is_binary(JsonMulti),
    true = binary:match(JsonMulti, <<"\n">>) =/= nomatch,

    %% Test with string functions - str.split with maxsplit kwarg
    {ok, Split} = py:eval(<<"'a-b-c-d'.split('-', maxsplit=2)">>),
    [<<"a">>, <<"b">>, <<"c-d">>] = Split,

    ok.

test_eval(_Config) ->
    {ok, 45} = py:eval(<<"sum(range(10))">>),
    {ok, 25} = py:eval(<<"x * y">>, #{x => 5, y => 5}),
    {ok, [0, 1, 4, 9, 16]} = py:eval(<<"[x**2 for x in range(5)]">>),
    ok.

test_eval_complex_locals(_Config) ->
    %% Test eval with list local
    {ok, 15} = py:eval(<<"sum(numbers)">>, #{numbers => [1, 2, 3, 4, 5]}),

    %% Test eval with map local
    {ok, <<"Alice">>} = py:eval(<<"data['name']">>, #{data => #{name => <<"Alice">>, age => 30}}),

    %% Test eval with nested structures
    {ok, 42} = py:eval(<<"config['settings']['value']">>,
                       #{config => #{settings => #{value => 42}}}),

    %% Test eval with multiple complex locals using default argument capture
    %% Note: Python's eval() locals aren't accessible in nested scopes (lambda/comprehension)
    %% Use default argument to capture the value: lambda x, m=multiplier: x * m
    {ok, [2, 4, 6]} = py:eval(<<"list(map(lambda x, m=multiplier: x * m, items))">>,
                              #{items => [1, 2, 3], multiplier => 2}),

    %% Test eval with multiple variables
    {ok, 6} = py:eval(<<"a + b + c">>, #{a => 1, b => 2, c => 3}),

    %% Test eval using local in function call
    {ok, 3} = py:eval(<<"len(word)">>, #{word => <<"foo">>}),

    %% Test eval with nested map access
    {ok, <<"value">>} = py:eval(<<"obj['nested']['key']">>,
                                #{obj => #{nested => #{key => <<"value">>}}}),

    ok.

test_exec(_Config) ->
    %% Test exec - just ensure it doesn't fail
    %% Note: exec runs on any worker, subsequent calls may go to different workers
    ok = py:exec(<<"x = 42">>),
    ok = py:exec(<<"
def my_func():
    pass
">>),
    ok.

test_cast(_Config) ->
    %% Test fire-and-forget cast with erlang.send
    Self = erlang:self(),
    ok = py:cast(erlang, send, [Self, {<<"result">>, <<"msg1">>}]),
    ok = py:cast(erlang, send, [Self, {<<"result">>, <<"msg2">>}]),
    %% Wait for results (order may vary)
    R1 = receive {<<"result">>, V1} -> V1 after 5000 -> ct:fail(timeout1) end,
    R2 = receive {<<"result">>, V2} -> V2 after 5000 -> ct:fail(timeout2) end,
    true = lists:sort([R1, R2]) =:= [<<"msg1">>, <<"msg2">>],
    ok.

test_cast_kwargs(_Config) ->
    %% py:cast/4 must accept and forward a kwargs map without crashing.
    %% The kwargs marshaling itself is exercised by test_call_kwargs/1; this
    %% covers the cast/4 wrapper whose only difference from cast/3 is the
    %% Kwargs map.
    ok = py:cast(json, dumps, [#{<<"a">> => 1}], #{indent => 2}),
    ok = py:cast(json, dumps, [[1, 2, 3]], #{separators => [<<", ">>, <<": ">>]}),
    %% Allow the spawned executors to actually run the calls.
    timer:sleep(100),
    ok.

test_spawn_call(_Config) ->
    %% Test spawn_call with await
    Ref1 = py:spawn_call(math, sqrt, [100]),
    Ref2 = py:spawn_call(math, sqrt, [144]),
    {ok, 10.0} = py:await(Ref1),
    {ok, 12.0} = py:await(Ref2),
    ok.

test_type_conversions(_Config) ->
    %% Integer
    {ok, 42} = py:eval(<<"42">>),

    %% Float
    {ok, 3.14} = py:eval(<<"3.14">>),

    %% String -> binary
    {ok, <<"hello">>} = py:eval(<<"'hello'">>),

    %% Boolean
    {ok, true} = py:eval(<<"True">>),
    {ok, false} = py:eval(<<"False">>),

    %% None
    {ok, none} = py:eval(<<"None">>),

    %% List
    {ok, [1, 2, 3]} = py:eval(<<"[1, 2, 3]">>),

    %% Tuple
    {ok, {1, 2, 3}} = py:eval(<<"(1, 2, 3)">>),

    %% Dict -> map
    {ok, #{<<"a">> := 1}} = py:eval(<<"{'a': 1}">>),

    ok.

test_nested_types(_Config) ->
    %% Test nested list
    {ok, [[1, 2], [3, 4]]} = py:eval(<<"[[1, 2], [3, 4]]">>),

    %% Test nested dict
    {ok, #{<<"outer">> := #{<<"inner">> := 42}}} =
        py:eval(<<"{'outer': {'inner': 42}}">>),

    %% Test list of dicts
    {ok, [#{<<"a">> := 1}, #{<<"b">> := 2}]} =
        py:eval(<<"[{'a': 1}, {'b': 2}]">>),

    %% Test dict with list values
    {ok, #{<<"items">> := [1, 2, 3]}} = py:eval(<<"{'items': [1, 2, 3]}">>),

    %% Test tuple containing lists
    {ok, {[1, 2], [3, 4]}} = py:eval(<<"([1, 2], [3, 4])">>),

    %% Test list containing tuple
    {ok, [{1, 2}, {3, 4}]} = py:eval(<<"[(1, 2), (3, 4)]">>),

    %% Test deep nesting
    {ok, #{<<"level1">> := #{<<"level2">> := #{<<"level3">> := <<"deep">>}}}} =
        py:eval(<<"{'level1': {'level2': {'level3': 'deep'}}}">>),

    %% Test mixed nesting
    {ok, #{<<"data">> := [{1, <<"a">>}, {2, <<"b">>}]}} =
        py:eval(<<"{'data': [(1, 'a'), (2, 'b')]}">>),

    ok.

test_timeout(_Config) ->
    %% Test that timeout works - use time.sleep which guarantees delay
    %% time.sleep(1) will definitely exceed 100ms timeout
    {error, timeout} = py:eval(<<"__import__('time').sleep(1)">>, #{}, 100),

    %% Test that normal operations complete within timeout
    {ok, 45} = py:eval(<<"sum(range(10))">>, #{}, 5000),

    %% Test call with timeout
    {ok, 4.0} = py:call(math, sqrt, [16], #{}, 5000),

    ok.

test_special_floats(_Config) ->
    %% Test NaN
    {ok, nan} = py:eval(<<"float('nan')">>),

    %% Test Infinity
    {ok, infinity} = py:eval(<<"float('inf')">>),

    %% Test negative infinity
    {ok, neg_infinity} = py:eval(<<"float('-inf')">>),

    ok.

test_streaming(_Config) ->
    %% Test generator expression streaming
    {ok, [0, 1, 4, 9, 16]} = py:stream_eval(<<"(x**2 for x in range(5))">>),

    %% Test range iterator
    {ok, [0, 1, 2, 3, 4]} = py:stream_eval(<<"iter(range(5))">>),

    %% Test map streaming
    {ok, [<<"A">>, <<"B">>, <<"C">>]} = py:stream_eval(<<"(c.upper() for c in 'abc')">>),

    %% Test filtered generator expression (documented in streaming.md)
    {ok, Evens} = py:stream_eval(<<"(x for x in range(10) if x % 2 == 0)">>),
    [0, 2, 4, 6, 8] = Evens,

    ok.

test_stream_function(_Config) ->
    %% Test streaming from generator functions
    %% As documented in streaming.md, you can define generator functions
    %% and stream from them. We test with inline generators using lambda.

    %% Fibonacci generator - compute fibonacci sequence using lambda wrapper
    %% (as documented pattern for inline generators)
    {ok, Fib} = py:stream_eval(<<"(lambda: ((fib := [0, 1]), [fib.append(fib[-1] + fib[-2]) for _ in range(8)], iter(fib))[-1])()">>),
    ct:pal("Fibonacci result: ~p~n", [Fib]),
    [0, 1, 1, 2, 3, 5, 8, 13, 21, 34] = Fib,

    %% Test py:stream/3 with builtins that return iterators
    {ok, Result} = py:stream(builtins, iter, [[1, 2, 3, 4, 5]]),
    [1, 2, 3, 4, 5] = Result,

    %% Test stream with enumerate and kwargs (start parameter)
    {ok, EnumResult} = py:stream(builtins, enumerate, [[<<"a">>, <<"b">>, <<"c">>]], #{start => 1}),
    [{1, <<"a">>}, {2, <<"b">>}, {3, <<"c">>}] = EnumResult,

    %% Test stream_eval with filter (returns iterator)
    {ok, FilterResult} = py:stream_eval(<<"filter(lambda x: x > 2, [1, 2, 3, 4, 5])">>),
    [3, 4, 5] = FilterResult,

    %% Test batching generator (as documented in streaming.md)
    {ok, Batches} = py:stream_eval(<<"(lambda data, size: (data[i:i+size] for i in range(0, len(data), size)))([1,2,3,4,5,6,7], 3)">>),
    ct:pal("Batches: ~p~n", [Batches]),
    [[1, 2, 3], [4, 5, 6], [7]] = Batches,

    ok.

test_stream_iterators(_Config) ->
    %% Test dictionary items iterator (documented in streaming.md)
    {ok, Items} = py:stream_eval(<<"iter({'a': 1, 'b': 2, 'c': 3}.items())">>),
    ct:pal("Dict items: ~p~n", [Items]),
    %% Items should be tuples
    true = lists:all(fun(I) -> is_tuple(I) end, Items),
    3 = length(Items),

    %% Test dictionary keys iterator
    {ok, Keys} = py:stream_eval(<<"iter({'x': 10, 'y': 20}.keys())">>),
    true = lists:all(fun(K) -> is_binary(K) end, Keys),

    %% Test dictionary values iterator
    {ok, Values} = py:stream_eval(<<"iter({'x': 10, 'y': 20}.values())">>),
    true = lists:all(fun(V) -> is_integer(V) end, Values),

    %% Test enumerate iterator
    {ok, Enumerated} = py:stream_eval(<<"enumerate(['a', 'b', 'c'])">>),
    [{0, <<"a">>}, {1, <<"b">>}, {2, <<"c">>}] = Enumerated,

    %% Test zip iterator
    {ok, Zipped} = py:stream_eval(<<"zip([1, 2, 3], ['a', 'b', 'c'])">>),
    [{1, <<"a">>}, {2, <<"b">>}, {3, <<"c">>}] = Zipped,

    ok.

test_error_handling(_Config) ->
    %% Test Python exception
    {error, {'NameError', _}} = py:eval(<<"undefined_variable">>),

    %% Test syntax error
    {error, {'SyntaxError', _}} = py:eval(<<"if True">>),

    %% Test division by zero
    {error, {'ZeroDivisionError', _}} = py:eval(<<"1/0">>),

    %% Test import error
    {error, {'ModuleNotFoundError', _}} = py:call(nonexistent_module, func, []),

    ok.

test_version(_Config) ->
    {ok, Version} = py:version(),
    true = is_binary(Version),
    true = byte_size(Version) > 0,
    ok.

test_memory_stats(_Config) ->
    %% tracemalloc doesn't support subinterpreters
    case py_nif:subinterp_supported() of
        true ->
            {skip, "tracemalloc not supported in subinterpreters"};
        false ->
            {ok, Stats} = py:memory_stats(),
            true = is_map(Stats),
            %% Check that we have GC stats
            true = maps:is_key(gc_stats, Stats),
            true = maps:is_key(gc_count, Stats),
            true = maps:is_key(gc_threshold, Stats),

            %% Test tracemalloc
            ok = py:tracemalloc_start(),
            %% Allocate some memory
            {ok, _} = py:eval(<<"[x**2 for x in range(1000)]">>),
            {ok, StatsWithTrace} = py:memory_stats(),
            true = maps:is_key(traced_memory_current, StatsWithTrace),
            true = maps:is_key(traced_memory_peak, StatsWithTrace),
            ok = py:tracemalloc_stop(),

            ok
    end.

test_gc(_Config) ->
    %% Test basic GC
    {ok, Collected} = py:gc(),
    true = is_integer(Collected),
    true = Collected >= 0,

    %% Test generation-specific GC
    {ok, Collected0} = py:gc(0),
    true = is_integer(Collected0),

    {ok, Collected1} = py:gc(1),
    true = is_integer(Collected1),

    {ok, Collected2} = py:gc(2),
    true = is_integer(Collected2),

    ok.

test_erlang_callback(_Config) ->
    %% Register a simple function that adds two numbers
    py:register_function(add, fun([A, B]) -> A + B end),

    %% Call from Python - erlang module is auto-imported
    {ok, Result1} = py:eval(<<"erlang.call('add', 5, 7)">>),
    12 = Result1,

    %% Register function with multiple return types
    py:register_function(get_info, fun([]) -> #{name => <<"test">>, count => 42} end),
    {ok, Info} = py:eval(<<"erlang.call('get_info')">>),
    #{<<"name">> := <<"test">>, <<"count">> := 42} = Info,

    %% Register function that works with lists
    py:register_function(reverse_list, fun([L]) -> lists:reverse(L) end),
    {ok, Reversed} = py:eval(<<"erlang.call('reverse_list', [1, 2, 3])">>),
    [3, 2, 1] = Reversed,

    %% Cleanup
    py:unregister_function(add),
    py:unregister_function(get_info),
    py:unregister_function(reverse_list),

    ok.

test_erlang_callback_mfa(_Config) ->
    %% Test registering an MFA (Module:Function style)
    %%
    %% Note: MFA callbacks receive ALL arguments as a single list [Args].
    %% So erlang.call('foo', 1, 2, 3) passes Args = [1, 2, 3] to the function.
    %% The function is called as Module:Function([1, 2, 3]).
    %%
    %% This means functions that work with MFA must accept a list argument.

    %% lists:sum expects a list - works perfectly with MFA
    py:register_function(sum_list, lists, sum),
    {ok, 15} = py:eval(<<"erlang.call('sum_list', 1, 2, 3, 4, 5)">>),

    %% lists:reverse expects a list - works with MFA
    py:register_function(reverse_list, lists, reverse),
    {ok, [3, 2, 1]} = py:eval(<<"erlang.call('reverse_list', 1, 2, 3)">>),

    %% erlang:hd gets the first element of the args list
    py:register_function(first_arg, erlang, hd),
    {ok, <<"hello">>} = py:eval(<<"erlang.call('first_arg', 'hello', 'world')">>),

    %% erlang:length gets the count of arguments
    py:register_function(arg_count, erlang, length),
    {ok, 4} = py:eval(<<"erlang.call('arg_count', 'a', 'b', 'c', 'd')">>),

    %% Cleanup
    py:unregister_function(sum_list),
    py:unregister_function(reverse_list),
    py:unregister_function(first_arg),
    py:unregister_function(arg_count),

    ok.

test_erlang_import_syntax(_Config) ->
    %% Test "from erlang import func_name" syntax
    %% This test mirrors test_erlang_callback but uses the new import syntax
    %%
    %% Note: We use assert statements to verify results within the exec block
    %% because workers may change between py:exec and py:eval calls.
    py:register_function(add, fun([A, B]) -> A + B end),
    py:register_function(greet, fun([Name]) -> <<"Hello, ", Name/binary>> end),

    %% Test basic import - will fail with AssertionError if result is wrong
    ok = py:exec(<<"
from erlang import add
result = add(10, 20)
assert result == 30, f'Expected 30, got {result}'
">>),

    %% Test import with alias
    ok = py:exec(<<"
from erlang import add as my_add
result = my_add(40, 60)
assert result == 100, f'Expected 100, got {result}'
">>),

    %% Test importing and using a string function
    %% Note: Erlang binary comes back as Python str (due to conversion)
    ok = py:exec(<<"
from erlang import greet
result = greet('World')
assert result == 'Hello, World' or result == b'Hello, World', f'Expected \"Hello, World\", got {result}'
">>),

    %% Cleanup
    py:unregister_function(add),
    py:unregister_function(greet),

    ok.

test_erlang_attr_syntax(_Config) ->
    %% Test "erlang.func_name()" syntax (via __getattr__)
    %% This test mirrors test_erlang_callback but uses the new attribute syntax
    %% Note: erlang module is auto-imported, no need for "import erlang"
    py:register_function(multiply, fun([A, B]) -> A * B end),
    py:register_function(get_data, fun([]) -> #{value => 42} end),

    %% Test direct attribute access call (erlang is auto-imported)
    {ok, 50} = py:eval(<<"erlang.multiply(5, 10)">>),

    %% Test no-arg function via attr syntax
    {ok, #{<<"value">> := 42}} = py:eval(<<"erlang.get_data()">>),

    %% Verify backward compatibility - erlang.call() still works
    {ok, 100} = py:eval(<<"erlang.call('multiply', 10, 10)">>),

    %% Cleanup
    py:unregister_function(multiply),
    py:unregister_function(get_data),

    ok.

test_asyncio_call(_Config) ->
    %% Sync function dispatched through the event loop short-circuits to
    %% the result, so async_call/async_await round-trips end-to-end.
    Ref = py:async_call(math, sqrt, [16]),
    true = is_reference(Ref),
    {ok, 4.0} = py:async_await(Ref, 5000),
    ok.

test_asyncio_gather(_Config) ->
    Calls = [
        {math, sqrt, [16]},
        {math, sqrt, [25]},
        {math, sqrt, [36]}
    ],
    {ok, [4.0, 5.0, 6.0]} = py:async_gather(Calls),
    ok.

test_asyncio_gather_timeout(_Config) ->
    %% py:async_gather/2 with an explicit timeout returns inside the budget
    %% on success.
    Calls = [
        {math, sqrt, [4]},
        {math, sqrt, [9]},
        {math, sqrt, [16]}
    ],
    {ok, [2.0, 3.0, 4.0]} = py:async_gather(Calls, 5000),
    ok.

test_subinterp_supported(_Config) ->
    %% Test that subinterp_supported returns a boolean
    Result = py:subinterp_supported(),
    true = is_boolean(Result),
    %% Log the result
    ct:pal("Sub-interpreter support: ~p~n", [Result]),
    ok.

test_parallel_execution(_Config) ->
    %% Test parallel execution using sub-interpreters (Python 3.12+)
    case py:subinterp_supported() of
        true ->
            Calls = [
                {math, sqrt, [16]},
                {math, sqrt, [25]},
                {math, sqrt, [36]},
                {math, sqrt, [49]}
            ],
            Result = py:parallel(Calls),
            ct:pal("Parallel result: ~p~n", [Result]),
            %% parallel/1 should return {ok, Results} on success
            %% or {error, Reason} on failure
            case Result of
                {ok, Results} when is_list(Results) ->
                    4 = length(Results),
                    ok;
                {error, Reason} ->
                    ct:pal("Parallel execution failed: ~p (may be expected on some setups)~n", [Reason]),
                    ok
            end;
        false ->
            ct:pal("Sub-interpreters not supported, skipping parallel test~n"),
            ok
    end.

test_venv(_Config) ->
    %% Test venv_info when no venv is active
    {ok, #{<<"active">> := false}} = py:venv_info(),

    %% Create a fake venv structure for testing (faster than real venv)
    TmpDir = <<"/tmp/erlang_python_test_venv">>,

    %% Get Python version for site-packages path
    {ok, PyVer} = py:eval(<<"f'python{__import__(\"sys\").version_info.major}.{__import__(\"sys\").version_info.minor}'">>),

    %% Create minimal venv structure using os.makedirs
    SitePackages = <<TmpDir/binary, "/lib/", PyVer/binary, "/site-packages">>,
    {ok, _} = py:call(os, makedirs, [SitePackages], #{exist_ok => true}),

    %% Activate it
    ok = py:activate_venv(TmpDir),

    %% Check venv_info
    {ok, Info} = py:venv_info(),
    true = maps:get(<<"active">>, Info),
    true = is_binary(maps:get(<<"venv_path">>, Info)),
    true = is_binary(maps:get(<<"site_packages">>, Info)),

    %% site-packages must be in sys.path and at position 0
    {ok, true} = py:eval(<<"sp in __import__('sys').path">>, #{sp => SitePackages}),
    {ok, 0} = py:eval(<<"__import__('sys').path.index(sp)">>, #{sp => SitePackages}),

    %% Deactivate
    ok = py:deactivate_venv(),

    %% Verify deactivated: venv_info and sys.path both restored
    {ok, #{<<"active">> := false}} = py:venv_info(),
    {ok, false} = py:eval(<<"sp in __import__('sys').path">>, #{sp => SitePackages}),

    %% Test error case - invalid path
    {error, _} = py:activate_venv(<<"/nonexistent/path">>),

    %% Cleanup
    {ok, _} = py:call(shutil, rmtree, [TmpDir], #{ignore_errors => true}),

    ok.

test_venv_pth(_Config) ->
    %% Verify .pth files are processed (editable installs)
    TmpDir = <<"/tmp/erlang_python_test_venv_pth">>,
    PkgSrc = <<"/tmp/erlang_python_test_pth_src">>,

    {ok, PyVer} = py:eval(<<"f'python{__import__(\"sys\").version_info.major}.{__import__(\"sys\").version_info.minor}'">>),
    SitePackages = <<TmpDir/binary, "/lib/", PyVer/binary, "/site-packages">>,

    %% Create fake venv with a .pth file pointing at PkgSrc
    {ok, _} = py:call(os, makedirs, [SitePackages], #{exist_ok => true}),
    {ok, _} = py:call(os, makedirs, [PkgSrc], #{exist_ok => true}),
    PthFile = <<SitePackages/binary, "/test_editable.pth">>,
    {ok, _} = py:eval(<<"open(pf, 'w').write(pd + '\\n')">>, #{pf => PthFile, pd => PkgSrc}),

    %% Drop a module in PkgSrc so we can verify it's importable
    ModFile = <<PkgSrc/binary, "/ep_test_editable_mod.py">>,
    {ok, _} = py:eval(<<"open(f, 'w').write('answer = 42\\n')">>, #{f => ModFile}),

    {ok, false} = py:eval(<<"pd in __import__('sys').path">>, #{pd => PkgSrc}),

    %% Activate and verify paths and import
    ok = py:activate_venv(TmpDir),
    {ok, 0} = py:eval(<<"__import__('sys').path.index(sp)">>, #{sp => SitePackages}),
    {ok, 1} = py:eval(<<"__import__('sys').path.index(pd)">>, #{pd => PkgSrc}),
    {ok, 42} = py:eval(<<"__import__('ep_test_editable_mod').answer">>),

    %% Deactivate and verify cleanup
    ok = py:deactivate_venv(),
    {ok, false} = py:eval(<<"pd in __import__('sys').path">>, #{pd => PkgSrc}),
    {ok, false} = py:eval(<<"sp in __import__('sys').path">>, #{sp => SitePackages}),

    {ok, _} = py:call(shutil, rmtree, [TmpDir], #{ignore_errors => true}),
    {ok, _} = py:call(shutil, rmtree, [PkgSrc], #{ignore_errors => true}),
    ok.

%%% ============================================================================
%%% Scalability Tests
%%% ============================================================================

test_execution_mode(_Config) ->
    %% Test that execution_mode returns a valid mode
    Mode = py:execution_mode(),
    ct:pal("Execution mode: ~p~n", [Mode]),
    true = lists:member(Mode, [worker, owngil]),
    ok.

test_semaphore_basic(_Config) ->
    %% Test semaphore initialization and basic getters
    Max = py_semaphore:max_concurrent(),
    Current = py_semaphore:current(),

    ct:pal("Semaphore - Max: ~p, Current: ~p~n", [Max, Current]),

    true = is_integer(Max),
    true = Max > 0,
    true = is_integer(Current),
    true = Current >= 0,
    ok.

test_semaphore_acquire_release(_Config) ->
    %% Test basic acquire/release cycle
    Initial = py_semaphore:current(),

    %% Acquire
    ok = py_semaphore:acquire(1000),
    After1 = py_semaphore:current(),
    true = After1 > Initial,

    %% Release
    ok = py_semaphore:release(),
    After2 = py_semaphore:current(),
    After2 = Initial,

    ok.

test_semaphore_concurrent(_Config) ->
    %% Test concurrent acquire/release from multiple processes
    Parent = self(),
    NumProcs = 20,

    %% Spawn processes that each acquire, sleep, and release
    Pids = [spawn_link(fun() ->
        ok = py_semaphore:acquire(5000),
        timer:sleep(10),
        ok = py_semaphore:release(),
        Parent ! {done, self()}
    end) || _ <- lists:seq(1, NumProcs)],

    %% Wait for all to complete
    [receive {done, Pid} -> ok after 10000 -> ct:fail({timeout, Pid}) end || Pid <- Pids],

    %% Current should be back to 0 (or initial state)
    timer:sleep(50),
    Current = py_semaphore:current(),
    ct:pal("After concurrent test - Current: ~p~n", [Current]),
    true = Current =:= 0,

    ok.

test_semaphore_timeout(_Config) ->
    %% Save original max
    OrigMax = py_semaphore:max_concurrent(),

    %% Set a very low limit
    ok = py_semaphore:set_max_concurrent(1),
    1 = py_semaphore:max_concurrent(),

    %% First acquire should succeed
    ok = py_semaphore:acquire(1000),

    %% Second acquire should timeout (we're at max)
    Parent = self(),
    spawn_link(fun() ->
        Result = py_semaphore:acquire(100),
        Parent ! {acquire_result, Result}
    end),

    Result = receive
        {acquire_result, R} -> R
    after 2000 ->
        ct:fail(timeout_waiting_for_result)
    end,

    %% Should have gotten error
    {error, max_concurrent} = Result,

    %% Release and restore
    ok = py_semaphore:release(),
    ok = py_semaphore:set_max_concurrent(OrigMax),

    ok.

test_semaphore_rate_limiting(_Config) ->
    %% Test that the semaphore properly limits concurrent operations
    OrigMax = py_semaphore:max_concurrent(),

    %% Set limit to 3
    ok = py_semaphore:set_max_concurrent(3),

    Parent = self(),
    Counter = atomics:new(1, [{signed, false}]),

    %% Track max concurrent
    MaxSeen = atomics:new(1, [{signed, false}]),

    NumProcs = 10,
    Pids = [spawn_link(fun() ->
        ok = py_semaphore:acquire(10000),
        atomics:add(Counter, 1, 1),
        Cur = atomics:get(Counter, 1),

        %% Update max seen
        OldMax = atomics:get(MaxSeen, 1),
        case Cur > OldMax of
            true -> atomics:put(MaxSeen, 1, Cur);
            false -> ok
        end,

        timer:sleep(50),
        atomics:sub(Counter, 1, 1),
        ok = py_semaphore:release(),
        Parent ! {done, self()}
    end) || _ <- lists:seq(1, NumProcs)],

    %% Wait for all
    [receive {done, Pid} -> ok after 30000 -> ct:fail({timeout, Pid}) end || Pid <- Pids],

    %% Check that max concurrent never exceeded limit
    MaxConcurrent = atomics:get(MaxSeen, 1),
    ct:pal("Max concurrent seen: ~p (limit was 3)~n", [MaxConcurrent]),
    true = MaxConcurrent =< 3,

    %% Restore
    ok = py_semaphore:set_max_concurrent(OrigMax),

    ok.

test_overload_protection(_Config) ->
    %% Test that py:call returns overload error when semaphore is exhausted
    OrigMax = py_semaphore:max_concurrent(),

    %% Set very low limit
    ok = py_semaphore:set_max_concurrent(1),

    %% Acquire the only slot
    ok = py_semaphore:acquire(1000),

    %% Now py:call should fail with overload
    Parent = self(),
    spawn_link(fun() ->
        Result = py:call(math, sqrt, [4], #{}, 100),
        Parent ! {call_result, Result}
    end),

    Result = receive
        {call_result, R} -> R
    after 2000 ->
        ct:fail(timeout_waiting_for_result)
    end,

    ct:pal("Overload result: ~p~n", [Result]),

    %% Should be overloaded error
    case Result of
        {error, {overloaded, _, _}} -> ok;
        {error, timeout} -> ok;  %% Also acceptable
        Other -> ct:fail({unexpected_result, Other})
    end,

    %% Cleanup
    ok = py_semaphore:release(),
    ok = py_semaphore:set_max_concurrent(OrigMax),

    ok.

test_shared_state(_Config) ->
    %% Test shared state from Erlang side
    ok = py:state_store(<<"test_key">>, #{value => 42}),
    {ok, #{value := 42}} = py:state_fetch(<<"test_key">>),

    %% Test from Python side - set from Python, read from Erlang
    ok = py:exec(<<"
from erlang import state_set
state_set('python_key', {'message': 'hello from python', 'numbers': [1, 2, 3]})
">>),
    {ok, Result} = py:state_fetch(<<"python_key">>),
    ct:pal("State from Python: ~p~n", [Result]),
    #{<<"message">> := <<"hello from python">>, <<"numbers">> := [1, 2, 3]} = Result,

    %% Test from Python side - read state set from Erlang
    ok = py:state_store(<<"erlang_key">>, #{count => 100}),
    ok = py:exec(<<"
from erlang import state_get
data = state_get('erlang_key')
assert data == {'count': 100}, f'Expected {{count: 100}}, got {data}'
">>),

    %% Test state_keys from Python
    ok = py:exec(<<"
from erlang import state_keys
keys = state_keys()
assert 'test_key' in keys, f'Expected test_key in {keys}'
assert 'python_key' in keys, f'Expected python_key in {keys}'
assert 'erlang_key' in keys, f'Expected erlang_key in {keys}'
">>),

    %% Test state_delete from Python
    ok = py:exec(<<"
from erlang import state_delete
state_delete('python_key')
">>),
    {error, not_found} = py:state_fetch(<<"python_key">>),

    %% Test state_clear
    ok = py:state_clear(),
    [] = py:state_keys(),

    %% Test atomic counters from Erlang
    1 = py:state_incr(<<"counter">>),
    2 = py:state_incr(<<"counter">>),
    12 = py:state_incr(<<"counter">>, 10),
    11 = py:state_decr(<<"counter">>),
    6 = py:state_decr(<<"counter">>, 5),

    %% Test atomic counters from Python
    ok = py:exec(<<"
from erlang import state_incr, state_decr

# Increment
val = state_incr('py_counter')
assert val == 1, f'Expected 1, got {val}'

val = state_incr('py_counter', 5)
assert val == 6, f'Expected 6, got {val}'

# Decrement
val = state_decr('py_counter')
assert val == 5, f'Expected 5, got {val}'

val = state_decr('py_counter', 3)
assert val == 2, f'Expected 2, got {val}'
">>),
    {ok, 2} = py:state_fetch(<<"py_counter">>),

    %% Cleanup
    ok = py:state_clear(),

    ok.

%% Test module reload across all workers
test_reload(_Config) ->
    %% Module reload can trigger imports that don't support subinterpreters
    case py_nif:subinterp_supported() of
        true ->
            {skip, "module reload may use modules not supported in subinterpreters"};
        false ->
            %% First, ensure json module is imported in at least one worker
            {ok, _} = py:call(json, dumps, [[1, 2, 3]]),

            %% Now reload it - should succeed across all workers
            ok = py:reload(json),

            %% Verify the module still works after reload
            {ok, <<"[1, 2, 3]">>} = py:call(json, dumps, [[1, 2, 3]]),

            %% Test reload of a module that might not be loaded (should not error)
            ok = py:reload(collections),

            %% Test reload with binary module name
            ok = py:reload(<<"os">>),

            %% Test reload with string module name
            ok = py:reload("sys"),

            ok
    end.

%% ============================================================================
%% Process-bound Python Environment Tests
%% ============================================================================

%% Test that different processes have isolated Python environments
test_process_env_isolation(_Config) ->
    Parent = self(),

    %% Spawn process A and set x = 'A'
    spawn(fun() ->
        ok = py:exec(<<"x = 'A'">>),
        {ok, <<"A">>} = py:eval(<<"x">>),
        Parent ! {proc_a, ok}
    end),

    %% Spawn process B and set x = 'B'
    spawn(fun() ->
        ok = py:exec(<<"x = 'B'">>),
        {ok, <<"B">>} = py:eval(<<"x">>),
        Parent ! {proc_b, ok}
    end),

    %% Wait for both processes to complete
    receive {proc_a, ok} -> ok after 5000 -> ct:fail("Process A timed out") end,
    receive {proc_b, ok} -> ok after 5000 -> ct:fail("Process B timed out") end,

    ct:pal("Process isolation test passed~n"),
    ok.

%% Test that functions defined via exec() are accessible via __main__ module
test_process_env_main_function(_Config) ->
    %% Define a function in the process-local environment
    ok = py:exec(<<"def greet(name): return f'Hello {name}'">>),

    %% Call it via __main__ module
    {ok, <<"Hello World">>} = py:call('__main__', greet, [<<"World">>]),

    %% Define another function
    ok = py:exec(<<"def add(a, b): return a + b">>),
    {ok, 7} = py:call('__main__', add, [3, 4]),

    ct:pal("__main__ function access test passed~n"),
    ok.

%% Test that state persists across multiple calls within the same process
test_process_env_state_persistence(_Config) ->
    %% Initialize counter
    ok = py:exec(<<"counter = 0">>),

    %% Increment counter multiple times
    ok = py:exec(<<"counter += 1">>),
    ok = py:exec(<<"counter += 1">>),
    ok = py:exec(<<"counter += 1">>),

    %% Verify counter value
    {ok, 3} = py:eval(<<"counter">>),

    %% Use a function to increment
    ok = py:exec(<<"def increment(): global counter; counter += 1; return counter">>),
    {ok, 4} = py:call('__main__', increment, []),
    {ok, 5} = py:call('__main__', increment, []),

    %% Verify final value
    {ok, 5} = py:eval(<<"counter">>),

    ct:pal("State persistence test passed~n"),
    ok.

%% Test that process-local environment is cleaned up when process exits
test_process_env_cleanup(_Config) ->
    %% Get initial memory stats
    {ok, InitialStats} = py:memory_stats(),
    ct:pal("Initial memory: ~p~n", [InitialStats]),

    %% Spawn a process that creates a large list
    spawn(fun() ->
        ok = py:exec(<<"big = list(range(1000000))">>),
        timer:sleep(50)  % Let it exist briefly
    end),

    %% Wait for the process to exit
    timer:sleep(100),

    %% Force GC
    erlang:garbage_collect(),
    py:gc(),

    %% Check that memory was freed (this is a soft check)
    {ok, FinalStats} = py:memory_stats(),
    ct:pal("Final memory: ~p~n", [FinalStats]),

    ct:pal("Cleanup test passed~n"),
    ok.

%%% ============================================================================
%%% SharedDict Tests
%%% ============================================================================

%% Test basic SharedDict operations
test_shared_dict_basic(_Config) ->
    %% Create a SharedDict
    {ok, SD} = py:shared_dict_new(),
    ct:pal("Created SharedDict: ~p~n", [SD]),

    %% Initially empty
    [] = py:shared_dict_keys(SD),

    %% Set and get values
    ok = py:shared_dict_set(SD, <<"key1">>, <<"value1">>),
    <<"value1">> = py:shared_dict_get(SD, <<"key1">>),

    %% Get with default
    undefined = py:shared_dict_get(SD, <<"nonexistent">>),
    default_val = py:shared_dict_get(SD, <<"nonexistent">>, default_val),

    %% Set another key
    ok = py:shared_dict_set(SD, <<"key2">>, 42),
    42 = py:shared_dict_get(SD, <<"key2">>),

    %% Verify keys
    Keys = py:shared_dict_keys(SD),
    2 = length(Keys),
    true = lists:member(<<"key1">>, Keys),
    true = lists:member(<<"key2">>, Keys),

    %% Delete a key
    ok = py:shared_dict_del(SD, <<"key1">>),
    undefined = py:shared_dict_get(SD, <<"key1">>),

    %% Keys after delete
    [<<"key2">>] = py:shared_dict_keys(SD),

    ct:pal("Basic SharedDict test passed~n"),
    ok.

%% Test SharedDict with complex types (pickled)
test_shared_dict_types(_Config) ->
    {ok, SD} = py:shared_dict_new(),

    %% Map
    Map = #{<<"foo">> => <<"bar">>, <<"num">> => 123},
    ok = py:shared_dict_set(SD, <<"map">>, Map),
    Retrieved = py:shared_dict_get(SD, <<"map">>),
    ct:pal("Map: ~p, Retrieved: ~p~n", [Map, Retrieved]),
    %% Note: Keys may be converted to Python strings and back
    true = maps:get(<<"foo">>, Retrieved) =:= <<"bar">> orelse
           maps:get(foo, Retrieved) =:= <<"bar">>,

    %% List
    List = [1, 2, 3, <<"four">>, 5.0],
    ok = py:shared_dict_set(SD, <<"list">>, List),
    List = py:shared_dict_get(SD, <<"list">>),

    %% Nested structure
    Nested = #{<<"data">> => [#{<<"id">> => 1}, #{<<"id">> => 2}]},
    ok = py:shared_dict_set(SD, <<"nested">>, Nested),
    RetrievedNested = py:shared_dict_get(SD, <<"nested">>),
    ct:pal("Nested: ~p~n", [RetrievedNested]),

    %% Update a value
    ok = py:shared_dict_set(SD, <<"map">>, #{updated => true}),
    Updated = py:shared_dict_get(SD, <<"map">>),
    ct:pal("Updated map: ~p~n", [Updated]),

    ct:pal("SharedDict types test passed~n"),
    ok.

%% Test SharedDict lifecycle (GC-based cleanup)
test_shared_dict_process_death(_Config) ->
    %% Note: Process monitoring is disabled in current implementation.
    %% SharedDict is cleaned up by garbage collection when no references remain.
    %% This test verifies basic resource lifecycle.

    Parent = self(),

    %% Create SharedDict in a child process
    Child = spawn(fun() ->
        {ok, SD} = py:shared_dict_new(),
        ok = py:shared_dict_set(SD, <<"test">>, <<"value">>),
        <<"value">> = py:shared_dict_get(SD, <<"test">>),
        Parent ! {sd_ref, SD},
        receive
            continue -> ok
        after 5000 ->
            ok
        end
    end),

    %% Get the SharedDict reference
    SDRef = receive
        {sd_ref, Ref} -> Ref
    after 5000 ->
        error(timeout_waiting_for_sd_ref)
    end,

    ct:pal("Got SharedDict ref from child: ~p~n", [SDRef]),

    %% SharedDict should work while child is alive
    <<"value">> = py:shared_dict_get(SDRef, <<"test">>),

    %% Kill the child process
    exit(Child, kill),
    timer:sleep(100),

    %% SharedDict still accessible since we hold a reference
    %% (process monitoring disabled - cleanup is GC-based)
    <<"value">> = py:shared_dict_get(SDRef, <<"test">>),

    ct:pal("SharedDict lifecycle test passed~n"),
    ok.

%% Test Python access to SharedDict
test_shared_dict_python_access(_Config) ->
    %% Create a SharedDict
    {ok, SD} = py:shared_dict_new(),

    %% Set a value from Erlang
    ok = py:shared_dict_set(SD, <<"config">>, #{port => 8080, host => <<"localhost">>}),

    %% First, store the handle in Python's global namespace using eval
    %% We use eval with locals to pass the handle, then store it globally
    {ok, _} = py:eval(<<"(globals().__setitem__('_sd_handle', handle), None)[-1]">>,
                       #{<<"handle">> => SD}),

    %% Now execute code that uses the stored handle
    Code = <<"
from erlang import SharedDict
sd = SharedDict(_sd_handle)

# Read value set from Erlang
config = sd['config']
result_port = config.get('port') or config.get(b'port')

# Set a value from Python
sd['python_key'] = {'set_from': 'python', 'value': 42}

# Test contains
has_config = 'config' in sd
has_missing = 'missing' in sd

# Test keys
key_count = len(sd.keys())
">>,

    ok = py:exec(Code),

    %% Get results from Python namespace
    {ok, ResultPort} = py:eval(<<"result_port">>),
    ct:pal("Result port: ~p~n", [ResultPort]),
    8080 = ResultPort,

    {ok, HasConfig} = py:eval(<<"has_config">>),
    true = HasConfig,

    {ok, HasMissing} = py:eval(<<"has_missing">>),
    false = HasMissing,

    {ok, KeyCount} = py:eval(<<"key_count">>),
    ct:pal("Key count: ~p~n", [KeyCount]),
    2 = KeyCount,

    %% Read value set from Python in Erlang
    PythonValue = py:shared_dict_get(SD, <<"python_key">>),
    ct:pal("Python value: ~p~n", [PythonValue]),
    42 = maps:get(<<"value">>, PythonValue),

    %% Test delete from Python
    ok = py:exec(<<"del sd['python_key']">>),
    undefined = py:shared_dict_get(SD, <<"python_key">>),

    ct:pal("SharedDict Python access test passed~n"),
    ok.

%% Test explicit SharedDict destroy
test_shared_dict_destroy(_Config) ->
    %% Create SharedDict and populate it
    {ok, SD} = py:shared_dict_new(),
    ok = py:shared_dict_set(SD, <<"key">>, <<"value">>),
    <<"value">> = py:shared_dict_get(SD, <<"key">>),
    ct:pal("SharedDict created and populated~n"),

    %% Destroy it
    ok = py:shared_dict_destroy(SD),
    ct:pal("SharedDict destroyed~n"),

    %% Further access returns badarg
    try
        py:shared_dict_get(SD, <<"key">>),
        ct:fail(expected_badarg)
    catch
        error:badarg -> ok
    end,
    ct:pal("Get after destroy correctly raises badarg~n"),

    %% set also returns badarg
    try
        py:shared_dict_set(SD, <<"key2">>, <<"value2">>),
        ct:fail(expected_badarg)
    catch
        error:badarg -> ok
    end,
    ct:pal("Set after destroy correctly raises badarg~n"),

    %% keys also returns badarg
    try
        py:shared_dict_keys(SD),
        ct:fail(expected_badarg)
    catch
        error:badarg -> ok
    end,
    ct:pal("Keys after destroy correctly raises badarg~n"),

    %% Destroy is idempotent
    ok = py:shared_dict_destroy(SD),
    ok = py:shared_dict_destroy(SD),
    ct:pal("Destroy is idempotent~n"),

    ct:pal("SharedDict destroy test passed~n"),
    ok.

%%% ============================================================================
%%% SharedDict Concurrent Tests
%%% ============================================================================

%% Test concurrent access to the same key from multiple processes
%% Verifies mutex protection - individual operations are atomic, data remains consistent
%% Note: SharedDict provides per-operation atomicity, not transactional atomicity
test_shared_dict_concurrent_same_key(_Config) ->
    {ok, SD} = py:shared_dict_new(),
    Parent = self(),
    NumProcs = 10,
    WritesPerProc = 100,

    %% Test 1: Concurrent writes to same key - verify no corruption
    %% Each process writes its process number repeatedly
    Pids1 = [spawn_link(fun() ->
        lists:foreach(fun(_) ->
            ok = py:shared_dict_set(SD, <<"shared_key">>, N)
        end, lists:seq(1, WritesPerProc)),
        Parent ! {done_write, self()}
    end) || N <- lists:seq(1, NumProcs)],

    [receive {done_write, Pid} -> ok after 30000 -> ct:fail({timeout, Pid}) end || Pid <- Pids1],

    %% Final value should be a valid process number (1-NumProcs)
    FinalWrite = py:shared_dict_get(SD, <<"shared_key">>),
    ct:pal("Concurrent writes - final value: ~p (valid if 1-~p)~n", [FinalWrite, NumProcs]),
    true = is_integer(FinalWrite),
    true = FinalWrite >= 1 andalso FinalWrite =< NumProcs,

    %% Test 2: Concurrent reads while writing - verify no crashes/corruption
    ok = py:shared_dict_set(SD, <<"rw_key">>, 0),

    %% Half writers, half readers
    WriterPids = [spawn_link(fun() ->
        lists:foreach(fun(I) ->
            ok = py:shared_dict_set(SD, <<"rw_key">>, I)
        end, lists:seq(1, WritesPerProc)),
        Parent ! {done_writer, self()}
    end) || _ <- lists:seq(1, NumProcs div 2)],

    ReaderPids = [spawn_link(fun() ->
        Reads = [py:shared_dict_get(SD, <<"rw_key">>) || _ <- lists:seq(1, WritesPerProc)],
        %% All reads should return integers (no corruption)
        true = lists:all(fun is_integer/1, Reads),
        Parent ! {done_reader, self()}
    end) || _ <- lists:seq(1, NumProcs div 2)],

    [receive {done_writer, Pid} -> ok after 30000 -> ct:fail({timeout, Pid}) end || Pid <- WriterPids],
    [receive {done_reader, Pid} -> ok after 30000 -> ct:fail({timeout, Pid}) end || Pid <- ReaderPids],

    %% Test 3: Rapid key updates - verify latest value is consistent type
    lists:foreach(fun(I) ->
        ok = py:shared_dict_set(SD, <<"type_key">>, I)
    end, lists:seq(1, 1000)),

    TypeVal = py:shared_dict_get(SD, <<"type_key">>),
    true = is_integer(TypeVal),

    ct:pal("SharedDict concurrent same-key test passed~n"),
    ok.

%% Test concurrent access to different keys from multiple processes
%% Verifies no data corruption across keys
test_shared_dict_concurrent_different_keys(_Config) ->
    {ok, SD} = py:shared_dict_new(),
    Parent = self(),
    NumProcs = 10,
    NumOps = 50,

    %% Spawn processes, each writing to its own key
    Pids = [spawn_link(fun() ->
        Key = list_to_binary("key_" ++ integer_to_list(N)),
        %% Write operations
        lists:foreach(fun(I) ->
            ok = py:shared_dict_set(SD, Key, I)
        end, lists:seq(1, NumOps)),
        %% Final value should be NumOps
        Final = py:shared_dict_get(SD, Key),
        Parent ! {done, self(), Key, Final}
    end) || N <- lists:seq(1, NumProcs)],

    %% Wait for all and collect results
    Results = [receive
        {done, Pid, Key, Final} -> {Key, Final}
    after 30000 ->
        ct:fail({timeout, Pid})
    end || Pid <- Pids],

    %% Verify all keys have correct final value
    lists:foreach(fun({Key, Final}) ->
        case Final of
            NumOps -> ok;
            Other ->
                ct:fail({wrong_value, Key, expected, NumOps, got, Other})
        end
    end, Results),

    %% Verify all keys are present
    Keys = py:shared_dict_keys(SD),
    NumProcs = length(Keys),

    ct:pal("SharedDict concurrent different-keys test passed~n"),
    ok.

%% Test mixed read/write operations on shared and unique keys
test_shared_dict_concurrent_mixed(_Config) ->
    {ok, SD} = py:shared_dict_new(),
    Parent = self(),
    NumProcs = 10,
    NumOps = 50,

    %% Initialize shared keys
    ok = py:shared_dict_set(SD, <<"shared_1">>, 0),
    ok = py:shared_dict_set(SD, <<"shared_2">>, 0),

    %% Spawn processes that do mixed operations
    Pids = [spawn_link(fun() ->
        UniqueKey = list_to_binary("unique_" ++ integer_to_list(N)),
        mixed_operations(SD, UniqueKey, NumOps),
        Parent ! {done, self(), UniqueKey}
    end) || N <- lists:seq(1, NumProcs)],

    %% Wait for all processes
    UniqueKeys = [receive
        {done, Pid, Key} -> Key
    after 30000 ->
        ct:fail({timeout, Pid})
    end || Pid <- Pids],

    %% Verify all unique keys exist with valid values
    lists:foreach(fun(Key) ->
        Value = py:shared_dict_get(SD, Key),
        true = is_integer(Value) andalso Value >= 0
    end, UniqueKeys),

    %% Verify shared keys are integers (may have any value due to races)
    Shared1 = py:shared_dict_get(SD, <<"shared_1">>),
    Shared2 = py:shared_dict_get(SD, <<"shared_2">>),
    true = is_integer(Shared1),
    true = is_integer(Shared2),

    ct:pal("Mixed operations - shared_1=~p, shared_2=~p~n", [Shared1, Shared2]),
    ct:pal("SharedDict concurrent mixed test passed~n"),
    ok.

%% Helper for mixed operations
mixed_operations(_SD, _UniqueKey, 0) -> ok;
mixed_operations(SD, UniqueKey, N) ->
    %% Mix of operations
    case N rem 4 of
        0 ->
            %% Write to unique key
            ok = py:shared_dict_set(SD, UniqueKey, N);
        1 ->
            %% Read from unique key
            _ = py:shared_dict_get(SD, UniqueKey, 0);
        2 ->
            %% Increment shared key
            Val = py:shared_dict_get(SD, <<"shared_1">>, 0),
            ok = py:shared_dict_set(SD, <<"shared_1">>, Val + 1);
        3 ->
            %% Read all keys
            _ = py:shared_dict_keys(SD)
    end,
    mixed_operations(SD, UniqueKey, N - 1).

%% Benchmark SharedDict operations
test_shared_dict_benchmark(_Config) ->
    {ok, SD} = py:shared_dict_new(),

    %% Warmup
    lists:foreach(fun(I) ->
        Key = list_to_binary("warmup_" ++ integer_to_list(I)),
        ok = py:shared_dict_set(SD, Key, I),
        I = py:shared_dict_get(SD, Key)
    end, lists:seq(1, 100)),

    %% Single process benchmark
    NumOps = 1000,

    %% Benchmark SET operations
    SetStart = erlang:monotonic_time(microsecond),
    lists:foreach(fun(I) ->
        Key = list_to_binary("bench_" ++ integer_to_list(I rem 100)),
        ok = py:shared_dict_set(SD, Key, I)
    end, lists:seq(1, NumOps)),
    SetEnd = erlang:monotonic_time(microsecond),
    SetDuration = SetEnd - SetStart,
    SetOpsPerSec = (NumOps * 1000000) div max(1, SetDuration),

    %% Benchmark GET operations
    GetStart = erlang:monotonic_time(microsecond),
    lists:foreach(fun(I) ->
        Key = list_to_binary("bench_" ++ integer_to_list(I rem 100)),
        _ = py:shared_dict_get(SD, Key)
    end, lists:seq(1, NumOps)),
    GetEnd = erlang:monotonic_time(microsecond),
    GetDuration = GetEnd - GetStart,
    GetOpsPerSec = (NumOps * 1000000) div max(1, GetDuration),

    ct:pal("~n=== SharedDict Single-Process Benchmark ===~n"),
    ct:pal("SET: ~p ops in ~p us (~p ops/sec)~n", [NumOps, SetDuration, SetOpsPerSec]),
    ct:pal("GET: ~p ops in ~p us (~p ops/sec)~n", [NumOps, GetDuration, GetOpsPerSec]),

    %% Multi-process concurrent benchmark
    Parent = self(),
    NumProcs = 4,
    OpsPerProc = 500,

    ConcStart = erlang:monotonic_time(microsecond),
    Pids = [spawn_link(fun() ->
        lists:foreach(fun(I) ->
            Key = list_to_binary("conc_" ++ integer_to_list(I rem 50)),
            ok = py:shared_dict_set(SD, Key, I),
            _ = py:shared_dict_get(SD, Key)
        end, lists:seq(1, OpsPerProc)),
        Parent ! {done, self()}
    end) || _ <- lists:seq(1, NumProcs)],

    [receive {done, Pid} -> ok after 30000 -> ct:fail({timeout, Pid}) end || Pid <- Pids],
    ConcEnd = erlang:monotonic_time(microsecond),
    ConcDuration = ConcEnd - ConcStart,

    TotalConcOps = NumProcs * OpsPerProc * 2,  % SET + GET per iteration
    ConcOpsPerSec = (TotalConcOps * 1000000) div max(1, ConcDuration),

    ct:pal("~n=== SharedDict Multi-Process Benchmark (~p procs) ===~n", [NumProcs]),
    ct:pal("TOTAL: ~p ops in ~p us (~p ops/sec)~n", [TotalConcOps, ConcDuration, ConcOpsPerSec]),

    %% Cleanup
    ok = py:shared_dict_destroy(SD),

    ct:pal("~nSharedDict benchmark completed~n"),
    ok.