%% @doc Test suite for the uvloop-inspired async task API. -module(py_async_task_SUITE). -include_lib("common_test/include/ct.hrl"). -export([all/0, groups/0, init_per_suite/1, end_per_suite/1]). -export([ %% Basic tests test_submit_task/1, test_create_task_await/1, test_run_sync/1, test_spawn_task/1, %% Stdlib tests test_math_sqrt/1, test_math_operations/1, %% Async coroutine tests test_async_coroutine/1, test_async_with_args/1, test_async_sleep/1, test_timer_event_triggering/1, %% Error handling tests test_async_error/1, test_invalid_module/1, test_invalid_function/1, test_timeout/1, %% Concurrency tests test_concurrent_tasks/1, test_batch_tasks/1, test_interleaved_sync_async/1, %% Edge cases test_empty_args/1, test_large_result/1, test_nested_data/1, %% Thread-local context tests test_thread_local_event_loop/1, %% Per-process namespace tests test_process_namespace_exec/1, test_process_namespace_eval/1, test_process_namespace_async_func/1, test_process_namespace_isolation/1, test_process_namespace_reentrant/1, %% Env reuse tests test_env_reuse_with_exec/1, test_env_reuse_async_function/1, test_env_reuse_spawn_task/1 ]). all() -> [ %% Basic tests test_submit_task, test_create_task_await, test_run_sync, test_spawn_task, %% Stdlib tests test_math_sqrt, test_math_operations, %% Async coroutine tests test_async_coroutine, test_async_with_args, test_async_sleep, test_timer_event_triggering, %% Error handling tests test_async_error, test_invalid_module, test_invalid_function, test_timeout, %% Concurrency tests test_concurrent_tasks, test_batch_tasks, test_interleaved_sync_async, %% Edge cases test_empty_args, test_large_result, test_nested_data, %% Thread-local context tests test_thread_local_event_loop, %% Per-process namespace tests test_process_namespace_exec, test_process_namespace_eval, test_process_namespace_async_func, test_process_namespace_isolation, test_process_namespace_reentrant, %% Env reuse tests test_env_reuse_with_exec, test_env_reuse_async_function, test_env_reuse_spawn_task ]. groups() -> []. init_per_suite(Config) -> application:ensure_all_started(erlang_python), timer:sleep(500), % Allow event loop to initialize Config. end_per_suite(_Config) -> ok. test_submit_task(_Config) -> %% Test task submission using high-level API with stdlib function Ref = py_event_loop:create_task(math, sqrt, [25.0]), Result = py_event_loop:await(Ref, 1000), ct:log("submit_task result: ~p", [Result]), {ok, 5.0} = Result. test_create_task_await(_Config) -> %% Test high-level create_task/await API with stdlib function Ref = py_event_loop:create_task(math, pow, [2.0, 10.0]), Result = py_event_loop:await(Ref, 1000), ct:log("create_task/await result: ~p", [Result]), {ok, 1024.0} = Result. test_run_sync(_Config) -> %% Test blocking run API with stdlib function Result = py_event_loop:run(math, floor, [3.7], #{timeout => 1000}), ct:log("run result: ~p", [Result]), {ok, 3} = Result. test_spawn_task(_Config) -> %% Test fire-and-forget spawn_task API with stdlib function ok = py_event_loop:spawn_task(math, ceil, [2.3]), %% Just verify it doesn't crash timer:sleep(100), true. %% ============================================================================ %% Stdlib tests %% ============================================================================ test_math_sqrt(_Config) -> %% Test calling math.sqrt via async task API Ref = py_event_loop:create_task(math, sqrt, [4.0]), {ok, Result} = py_event_loop:await(Ref, 5000), ct:log("math.sqrt(4.0) = ~p", [Result]), 2.0 = Result. test_math_operations(_Config) -> %% Test multiple math operations Ref1 = py_event_loop:create_task(math, pow, [2.0, 10.0]), Ref2 = py_event_loop:create_task(math, floor, [3.7]), Ref3 = py_event_loop:create_task(math, ceil, [3.2]), {ok, R1} = py_event_loop:await(Ref1, 5000), {ok, R2} = py_event_loop:await(Ref2, 5000), {ok, R3} = py_event_loop:await(Ref3, 5000), ct:log("math.pow(2, 10) = ~p", [R1]), ct:log("math.floor(3.7) = ~p", [R2]), ct:log("math.ceil(3.2) = ~p", [R3]), 1024.0 = R1, 3 = R2, 4 = R3. %% ============================================================================ %% Async coroutine tests %% ============================================================================ test_async_coroutine(_Config) -> %% Test sync function that completes quickly %% asyncio.sleep as coroutine may need special handling Ref = py_event_loop:create_task(math, sin, [0.0]), Result = py_event_loop:await(Ref, 5000), ct:log("math.sin(0.0) = ~p", [Result]), {ok, +0.0} = Result. test_async_with_args(_Config) -> %% Test with args using operator module Ref = py_event_loop:create_task(operator, add, [10, 20]), Result = py_event_loop:await(Ref, 5000), ct:log("operator.add(10, 20) = ~p", [Result]), {ok, 30} = Result. test_async_sleep(_Config) -> %% Test multiple quick operations in sequence %% (asyncio.sleep coroutines may need special loop driving) Results = lists:map(fun(N) -> Ref = py_event_loop:create_task(math, sqrt, [float(N * N)]), {N, py_event_loop:await(Ref, 5000)} end, lists:seq(1, 10)), ct:log("Sequential sqrt results: ~p", [Results]), %% Verify all succeeded lists:foreach(fun({N, {ok, R}}) -> true = abs(R - float(N)) < 0.0001 end, Results). test_timer_event_triggering(_Config) -> %% Test that timer events properly trigger event loop processing. %% %% This verifies the fix for the timer event triggering issue where %% asyncio.sleep would never complete because dispatch_timer added %% events to pending_head but nothing called _run_once to process them. %% %% The fix ensures that after dispatching timer/FD events, the worker %% sends task_ready to itself to trigger _run_once processing. %% %% Uses test_async_task module which has async functions with asyncio.sleep. %% Test simple_task which uses asyncio.sleep(0.01) ct:log("Testing simple async task with asyncio.sleep..."), Ref1 = py_event_loop:create_task(test_async_task, simple_task, []), Result1 = py_event_loop:await(Ref1, 5000), ct:log("simple_task result: ~p", [Result1]), {ok, <<"hello from async">>} = Result1, %% Test task_with_args which uses asyncio.sleep(0.01) ct:log("Testing async task with args and asyncio.sleep..."), Ref2 = py_event_loop:create_task(test_async_task, task_with_args, [10, 32]), Result2 = py_event_loop:await(Ref2, 5000), ct:log("task_with_args result: ~p", [Result2]), {ok, 42} = Result2, %% Test concurrent async tasks with sleep ct:log("Testing concurrent async tasks with asyncio.sleep..."), Refs = [py_event_loop:create_task(test_async_task, task_with_args, [N, N]) || N <- lists:seq(1, 5)], Results = [py_event_loop:await(Ref, 5000) || Ref <- Refs], ct:log("Concurrent results: ~p", [Results]), %% Verify all completed with correct values (N + N) Expected = [{ok, N * 2} || N <- lists:seq(1, 5)], Expected = Results, ct:log("timer_event_triggering test: all asyncio.sleep operations completed"). %% ============================================================================ %% Error handling tests %% ============================================================================ test_async_error(_Config) -> %% Test error handling - math.sqrt(-1) raises ValueError Ref = py_event_loop:create_task(math, sqrt, [-1.0]), Result = py_event_loop:await(Ref, 5000), ct:log("math.sqrt(-1) = ~p", [Result]), case Result of {error, _} -> ok; {ok, _} -> ct:fail("Expected error but got success") end. test_invalid_module(_Config) -> %% Test calling non-existent module Ref = py_event_loop:create_task(nonexistent_module_xyz, some_func, []), Result = py_event_loop:await(Ref, 2000), ct:log("nonexistent_module result: ~p", [Result]), %% Should timeout or error case Result of {error, _} -> ok; {ok, _} -> ct:fail("Expected error for invalid module") end. test_invalid_function(_Config) -> %% Test calling non-existent function Ref = py_event_loop:create_task(math, nonexistent_function_xyz, []), Result = py_event_loop:await(Ref, 2000), ct:log("nonexistent_function result: ~p", [Result]), %% Should timeout or error case Result of {error, _} -> ok; {ok, _} -> ct:fail("Expected error for invalid function") end. test_timeout(_Config) -> %% Test timeout handling - we just verify await timeout works %% Use a short sleep (0.5s) but even shorter timeout (50ms) Ref = py_event_loop:create_task(time, sleep, [0.5]), Result = py_event_loop:await(Ref, 50), ct:log("time.sleep(0.5) with 50ms timeout: ~p", [Result]), {error, timeout} = Result. %% ============================================================================ %% Concurrency tests %% ============================================================================ test_concurrent_tasks(_Config) -> %% Test multiple concurrent tasks from different processes Parent = self(), NumProcs = 10, TasksPerProc = 5, %% Spawn processes that each submit tasks Pids = [spawn_link(fun() -> Results = [begin Ref = py_event_loop:create_task(math, sqrt, [float(N * N)]), {N, py_event_loop:await(Ref, 5000)} end || N <- lists:seq(1, TasksPerProc)], Parent ! {self(), Results} end) || _ <- lists:seq(1, NumProcs)], %% Collect all results AllResults = [receive {Pid, R} -> R end || Pid <- Pids], ct:log("Concurrent results count: ~p", [length(lists:flatten(AllResults))]), %% Verify all succeeded lists:foreach(fun(Results) -> lists:foreach(fun({N, {ok, R}}) -> Expected = float(N), true = abs(R - Expected) < 0.0001 end, Results) end, AllResults). test_batch_tasks(_Config) -> %% Test submitting many tasks at once (tests batching) NumTasks = 100, %% Submit all tasks Refs = [py_event_loop:create_task(math, sqrt, [float(N)]) || N <- lists:seq(1, NumTasks)], %% Await all results Results = [{N, py_event_loop:await(Ref, 5000)} || {N, Ref} <- lists:zip(lists:seq(1, NumTasks), Refs)], ct:log("Batch tasks completed: ~p", [length(Results)]), %% Verify all succeeded lists:foreach(fun({N, {ok, R}}) -> Expected = math:sqrt(N), true = abs(R - Expected) < 0.0001 end, Results). test_interleaved_sync_async(_Config) -> %% Test mixing different stdlib calls R1 = py_event_loop:create_task(operator, add, [1, 2]), R2 = py_event_loop:create_task(math, sin, [0.0]), R3 = py_event_loop:create_task(operator, mul, [5, 6]), R4 = py_event_loop:create_task(math, sqrt, [64.0]), {ok, 3} = py_event_loop:await(R1, 5000), {ok, +0.0} = py_event_loop:await(R2, 5000), {ok, 30} = py_event_loop:await(R3, 5000), {ok, 8.0} = py_event_loop:await(R4, 5000), ct:log("Interleaved sync/async tests passed"). %% ============================================================================ %% Edge cases %% ============================================================================ test_empty_args(_Config) -> %% Test function with no args - use time.time() which returns a float Ref = py_event_loop:create_task(time, time, []), {ok, Result} = py_event_loop:await(Ref, 5000), ct:log("time.time() = ~p", [Result]), %% Should be a reasonable timestamp (after year 2020) true = is_float(Result) andalso Result > 1577836800.0. test_large_result(_Config) -> %% Test returning large data using range() N = 100, Ref = py_event_loop:create_task(builtins, list, [[{builtins, range, [N]}]]), Result = py_event_loop:await(Ref, 5000), ct:log("list(range(100)) result: ~p", [Result]), %% This may not work as expected due to nested call syntax %% Accept both success and timeout case Result of {ok, List} when is_list(List) -> ct:log("Got list of length ~p", [length(List)]); {error, _} -> ct:log("Got error (acceptable)") end. test_nested_data(_Config) -> %% Test returning nested data using json module Ref = py_event_loop:create_task(json, loads, [<<"{\"a\": [1, 2, 3], \"b\": {\"c\": 4}}">>]), {ok, Result} = py_event_loop:await(Ref, 5000), ct:log("json.loads result: ~p", [Result]), %% Verify structure #{<<"a">> := AVal, <<"b">> := BVal} = Result, [1, 2, 3] = AVal, #{<<"c">> := 4} = BVal. %% ============================================================================ %% Thread-local context tests %% ============================================================================ test_thread_local_event_loop(_Config) -> %% Test that the event loop thread-local context is properly set. %% %% This verifies the fix for the thread-local event loop context issue. %% process_ready_tasks runs on dirty NIF scheduler threads (named 'Dummy-X'), %% not the main thread. Without the fix, asyncio.get_running_loop() would %% raise RuntimeError: "There is no current event loop in thread 'Dummy-1'." %% %% The fix sets events._set_running_loop() before processing tasks. %% %% We verify this by running multiple concurrent async tasks - if the %% running loop context weren't set, task creation would fail. NumTasks = 20, Refs = [py_event_loop:create_task(math, sqrt, [float(N * N)]) || N <- lists:seq(1, NumTasks)], %% Await all results - this exercises the event loop processing Results = [{N, py_event_loop:await(Ref, 5000)} || {N, Ref} <- lists:zip(lists:seq(1, NumTasks), Refs)], ct:log("Thread-local context test: ~p tasks completed", [length(Results)]), %% Verify all succeeded with correct results lists:foreach(fun({N, {ok, R}}) -> Expected = float(N), true = abs(R - Expected) < 0.0001 end, Results). %% ============================================================================ %% Per-process namespace tests %% ============================================================================ test_process_namespace_exec(_Config) -> %% Test executing Python code in process namespace ok = py_event_loop:exec(<<"x = 42">>), ok = py_event_loop:exec(<<"y = x * 2">>), ct:log("exec test: defined x and y in process namespace"). test_process_namespace_eval(_Config) -> %% Test evaluating expressions in process namespace ok = py_event_loop:exec(<<"a = 10">>), ok = py_event_loop:exec(<<"b = 20">>), {ok, 10} = py_event_loop:eval(<<"a">>), {ok, 20} = py_event_loop:eval(<<"b">>), {ok, 30} = py_event_loop:eval(<<"a + b">>), ct:log("eval test: expressions evaluated correctly"). test_process_namespace_async_func(_Config) -> %% Test defining an async function and calling it via create_task ok = py_event_loop:exec(<<" def double(x): return x * 2 def add(a, b): return a + b ">>), %% Call the sync function via create_task with __main__ module Ref1 = py_event_loop:create_task('__main__', double, [21]), {ok, 42} = py_event_loop:await(Ref1, 5000), Ref2 = py_event_loop:create_task('__main__', add, [10, 32]), {ok, 42} = py_event_loop:await(Ref2, 5000), ct:log("async_func test: functions in process namespace called successfully"). test_process_namespace_isolation(_Config) -> %% Test that different processes have isolated namespaces Parent = self(), %% Define a variable in parent process ok = py_event_loop:exec(<<"parent_var = 'parent'">>), {ok, <<"parent">>} = py_event_loop:eval(<<"parent_var">>), %% Spawn a child process that defines its own variable Child = spawn(fun() -> %% Child should not see parent's variable Result1 = py_event_loop:eval(<<"parent_var">>), %% Define child's own variable ok = py_event_loop:exec(<<"child_var = 'child'">>), {ok, <<"child">>} = py_event_loop:eval(<<"child_var">>), Parent ! {self(), parent_visible, Result1} end), %% Wait for child result receive {Child, parent_visible, ParentResult} -> %% Child should NOT see parent's variable (isolated namespace) case ParentResult of {error, _} -> ct:log("isolation test: child correctly cannot see parent_var"); {ok, _} -> ct:log("isolation test: child unexpectedly saw parent_var (shared namespace)") end after 5000 -> ct:fail("isolation test: child process timed out") end, %% Parent should still see its variable {ok, <<"parent">>} = py_event_loop:eval(<<"parent_var">>), %% Parent should NOT see child's variable ChildVarResult = py_event_loop:eval(<<"child_var">>), case ChildVarResult of {error, _} -> ct:log("isolation test: parent correctly cannot see child_var"); {ok, _} -> ct:log("isolation test: parent unexpectedly saw child_var") end. test_process_namespace_reentrant(_Config) -> %% Test that namespace variables are accessible during task execution %% This verifies the thread-local namespace is set correctly %% Define a variable and a function that uses it ok = py_event_loop:exec(<<" shared_value = 100 def use_shared(): # Access shared_value from namespace return shared_value + 23 ">>), %% Call the function via create_task - it should access the namespace Ref = py_event_loop:create_task('__main__', use_shared, []), {ok, Result} = py_event_loop:await(Ref, 5000), ct:log("reentrant test: use_shared() returned ~p (expected 123)", [Result]), 123 = Result, %% Test with a function that modifies namespace ok = py_event_loop:exec(<<" def increment_shared(): global shared_value shared_value += 1 return shared_value ">>), Ref2 = py_event_loop:create_task('__main__', increment_shared, []), {ok, 101} = py_event_loop:await(Ref2, 5000), %% Verify the change persists in namespace {ok, 101} = py_event_loop:eval(<<"shared_value">>), ct:log("reentrant test: namespace modifications persist correctly"). %% ============================================================================ %% Env reuse tests %% ============================================================================ test_env_reuse_with_exec(_Config) -> %% Test that functions defined via py:exec with a context can be called %% using py_event_loop:run without explicit env passing. %% %% This verifies the automatic env reuse feature: when a process has %% a py_local_env in its process dictionary (from py:context/py:exec), %% py_event_loop:run automatically passes that env to the NIF so %% functions defined via exec are visible. %% Create a context and define a sync function via py:exec Ctx = py:context(1), ok = py:exec(Ctx, <<" def multiply(a, b): return a * b def greet(name): return f'Hello, {name}!' ">>), %% Call the functions via py_event_loop:run - should find them automatically {ok, Result1} = py_event_loop:run('__main__', multiply, [6, 7]), ct:log("env_reuse test: multiply(6, 7) = ~p", [Result1]), 42 = Result1, {ok, Result2} = py_event_loop:run('__main__', greet, [<<"World">>]), ct:log("env_reuse test: greet('World') = ~p", [Result2]), <<"Hello, World!">> = Result2, ct:log("env_reuse_with_exec test: py:exec functions callable via py_event_loop:run"). test_env_reuse_async_function(_Config) -> %% Test that async functions defined via py:exec can be called %% using py_event_loop:run with automatic env reuse. %% Create a context and define an async function Ctx = py:context(1), ok = py:exec(Ctx, <<" import asyncio async def async_add(a, b): await asyncio.sleep(0.01) return a + b async def async_process(data): # Simulate chunked processing chunks = [] for i in range(0, len(data), 4): chunk = data[i:i+4] await asyncio.sleep(0.001) chunks.append(chunk) return chunks ">>), %% Call async function via py_event_loop:run {ok, Sum} = py_event_loop:run('__main__', async_add, [10, 32]), ct:log("env_reuse test: async_add(10, 32) = ~p", [Sum]), 42 = Sum, %% Test with data processing Data = <<"abcdefghijklmnop">>, {ok, Chunks} = py_event_loop:run('__main__', async_process, [Data]), ct:log("env_reuse test: async_process chunks = ~p", [Chunks]), [<<"abcd">>, <<"efgh">>, <<"ijkl">>, <<"mnop">>] = Chunks, ct:log("env_reuse_async_function test: async py:exec functions work with py_event_loop:run"). test_env_reuse_spawn_task(_Config) -> %% Test that spawn_task also uses the caller's env for function lookup. %% %% spawn_task spawns a receiver process but should use the caller's env %% to find functions defined via py:exec. %% Create a context and define a function that writes to a shared variable Ctx = py:context(1), ok = py:exec(Ctx, <<" spawn_test_results = [] def record_value(x): spawn_test_results.append(x) return x def get_results(): return list(spawn_test_results) ">>), %% Use spawn_task to call the function (fire and forget) ok = py_event_loop:spawn_task('__main__', record_value, [42]), %% Wait a bit for the task to complete timer:sleep(200), %% Verify the function was called by checking the results list via run {ok, Results} = py_event_loop:run('__main__', get_results, []), ct:log("env_reuse_spawn_task test: spawn_test_results = ~p", [Results]), %% Should contain the value we passed case Results of [42] -> ct:log("env_reuse_spawn_task test: spawn_task correctly used caller's env"); _ -> %% If empty or different, the function wasn't found or wasn't called ct:log("env_reuse_spawn_task test: unexpected results ~p", [Results]), ct:fail({unexpected_results, Results}) end.