%% @doc Tests for erlang.sleep() and asyncio integration.

%%

%% Tests the erlang.sleep() function and erlang module asyncio integration.

-module(py_erlang_sleep_SUITE).

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

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

-export([

test_erlang_sleep_available/1,

test_erlang_sleep_basic/1,

test_erlang_sleep_zero/1,

test_erlang_sleep_accuracy/1,

test_erlang_run_module/1,

test_erlang_asyncio_gather/1,

test_erlang_asyncio_wait_for/1,

test_erlang_asyncio_create_task/1

]).

all() ->

[

test_erlang_sleep_available,

test_erlang_sleep_basic,

test_erlang_sleep_zero,

test_erlang_sleep_accuracy,

test_erlang_run_module,

test_erlang_asyncio_gather,

test_erlang_asyncio_wait_for,

test_erlang_asyncio_create_task

].

init_per_suite(Config) ->

{ok, _} = application:ensure_all_started(erlang_python),

{ok, _} = py:start_contexts(),

timer:sleep(500),

Config.

end_per_suite(_Config) ->

ok.

%% Test that erlang.sleep is available

test_erlang_sleep_available(_Config) ->

ok = py:exec(<<"

import erlang

result = hasattr(erlang, 'sleep')

assert result, 'erlang.sleep not found'

">>),

ct:pal("erlang.sleep is available"),

ok.

%% Test basic sleep functionality (sync context via callback)

test_erlang_sleep_basic(_Config) ->

ok = py:exec(<<"

import erlang

# Test basic sleep in sync context - should not raise

erlang.sleep(0.01) # 10ms

">>),

ct:pal("Basic sleep completed"),

ok.

%% Test zero/negative delay returns immediately

test_erlang_sleep_zero(_Config) ->

ok = py:exec(<<"

import erlang

import time

start = time.time()

erlang.sleep(0)

elapsed = (time.time() - start) * 1000

# Should return immediately (< 10ms accounting for Python overhead)

assert elapsed < 10, f'Zero sleep was slow: {elapsed}ms'

">>),

ct:pal("Zero sleep returned fast"),

ok.

%% Test sleep accuracy

test_erlang_sleep_accuracy(_Config) ->

ok = py:exec(<<"

import erlang

import time

delays = [0.01, 0.05, 0.1] # seconds

for delay in delays:

start = time.time()

erlang.sleep(delay)

elapsed = time.time() - start

# Allow wide tolerance for CI runners (can be slow/unpredictable)

assert delay * 0.5 <= elapsed <= delay * 10.0, \\

f'{delay}s sleep took {elapsed:.3f}s'

">>),

ct:pal("Sleep accuracy within tolerance"),

ok.

%% Test erlang.run() with asyncio

test_erlang_run_module(_Config) ->

ok = py:exec(<<"

import erlang

import asyncio

# Test erlang module has expected functions for event loop integration

funcs = ['run', 'new_event_loop', 'EventLoopPolicy', 'sleep']

for f in funcs:

assert hasattr(erlang, f), f'erlang missing {f}'

# Test run() with asyncio.sleep

async def test_sleep():

await asyncio.sleep(0.01) # 10ms

return 'done'

result = erlang.run(test_sleep())

assert result == 'done', f'Expected done, got {result}'

">>),

ct:pal("erlang.run() with asyncio works"),

ok.

%% Test asyncio.gather with erlang.run()

test_erlang_asyncio_gather(_Config) ->

ok = py:exec(<<"

import erlang

import asyncio

async def task(n):

await asyncio.sleep(0.01)

return n * 2

async def main():

results = await asyncio.gather(task(1), task(2), task(3))

assert results == [2, 4, 6], f'Expected [2, 4, 6], got {results}'

erlang.run(main())

">>),

ct:pal("asyncio.gather with erlang.run() works"),

ok.

%% Test asyncio.wait_for with timeout

test_erlang_asyncio_wait_for(_Config) ->

ok = py:exec(<<"

import erlang

import asyncio

async def fast_task():

await asyncio.sleep(0.01)

return 'fast'

async def main():

# Should complete before timeout

result = await asyncio.wait_for(fast_task(), timeout=1.0)

assert result == 'fast', f'Expected fast, got {result}'

erlang.run(main())

">>),

ct:pal("asyncio.wait_for with erlang.run() works"),

ok.

%% Test asyncio.create_task with erlang.run()

test_erlang_asyncio_create_task(_Config) ->

ok = py:exec(<<"

import erlang

import asyncio

async def background():

await asyncio.sleep(0.01)

return 'background_done'

async def main():

task = asyncio.create_task(background())

# Do some other work

await asyncio.sleep(0.005)

# Wait for task

result = await task

assert result == 'background_done', f'Expected background_done, got {result}'

erlang.run(main())

">>),

ct:pal("asyncio.create_task with erlang.run() works"),

ok.