Coverage for tests/test_ss_utils_async.py: 100%

1 

2import asyncio 

3import time 

4 

5import pytest 

6 

7from ss_utils_async import gather_w_semaphore 

8 

9 

10async def sample_coroutine(delay: float, value: int) -> int: 

11 """A sample coroutine that sleeps for a given delay and returns a value.""" 

12 await asyncio.sleep(delay) 

13 return value 

14 

15 

16@pytest.mark.asyncio 

17async def test_gather_w_semaphore_empty_list(): 

18 """Test that gather_w_semaphore handles an empty list of coroutines.""" 

19 coroutines = [] 

20 results = await gather_w_semaphore(coroutines) 

21 assert results == [] 

22 

23 

24@pytest.mark.asyncio 

25async def test_gather_w_semaphore_single_coroutine(): 

26 """Test that gather_w_semaphore works with a single coroutine.""" 

27 coroutines = [sample_coroutine(0.1, 1)] 

28 results = await gather_w_semaphore(coroutines) 

29 assert results == [1] 

30 

31 

32@pytest.mark.asyncio 

33async def test_gather_w_semaphore_multiple_coroutines(): 

34 """Test that gather_w_semaphore runs multiple coroutines and returns results in order.""" 

35 coroutines = [ 

36 sample_coroutine(0.1, 1), 

37 sample_coroutine(0.05, 2), 

38 sample_coroutine(0.15, 3), 

39 ] 

40 results = await gather_w_semaphore(coroutines) 

41 assert results == [1, 2, 3] 

42 

43 

44@pytest.mark.asyncio 

45async def test_gather_w_semaphore_concurrency_limit(): 

46 """Test that gather_w_semaphore respects the concurrency limit.""" 

47 max_coroutines = 2 

48 coroutines = [ 

49 sample_coroutine(0.2, 1), 

50 sample_coroutine(0.2, 2), 

51 sample_coroutine(0.2, 3), 

52 sample_coroutine(0.2, 4), 

53 ] 

54 start_time = time.time() 

55 await gather_w_semaphore(coroutines, max_coroutines=max_coroutines) 

56 end_time = time.time() 

57 duration = end_time - start_time 

58 # With 4 coroutines of 0.2s each and a limit of 2, it should take ~0.4s 

59 assert 0.4 <= duration < 0.6 

60 

61 

62@pytest.mark.asyncio 

63async def test_gather_w_semaphore_mixed_coroutines(): 

64 """Test that gather_w_semaphore handles a mix of fast and slow coroutines.""" 

65 coroutines = [ 

66 sample_coroutine(0.3, 1), 

67 sample_coroutine(0.1, 2), 

68 sample_coroutine(0.2, 3), 

69 ] 

70 results = await gather_w_semaphore(coroutines) 

71 assert results == [1, 2, 3]