Coverage for tests/test_parallel.py: 85%

1import time 

2from typing import List 

3 

4import appl 

5from appl import CallFuture, call, ppl, records 

6from appl.types import ExecutorType 

7 

8 

9def test_call_future(): 

10 def run(t): 

11 time.sleep(t) 

12 return 1 

13 

14 t0 = time.time() 

15 n = 3 

16 calls: List[CallFuture] = [] 

17 for i in range(n): 

18 calls.append(call(run, t=0.1)) 

19 used = time.time() - t0 

20 assert used < 0.05 

21 

22 c = sum(call() for call in calls) 

23 used = time.time() - t0 

24 assert c == n 

25 assert used < 0.2 

26 

27 c = sum(call.val for call in calls) 

28 used = time.time() - t0 

29 assert c == n 

30 assert used < 0.2 

31 

32 

33def sleep(t): 

34 time.sleep(t) 

35 return 1 

36 

37 

38def test_call_future_process(): 

39 c = call(sleep, t=0.1, executor_type=ExecutorType.NEW_PROCESS) 

40 assert c.val == 1 

41 

42 

43def test_format_future(): 

44 def run(t): 

45 time.sleep(t) 

46 return 1 

47 

48 n = 3 

49 

50 @ppl 

51 def func(): 

52 t0 = time.time() 

53 for i in range(n): 

54 f"{call(run, t=0.1)}" 

55 print(time.time() - t0) 

56 return records() 

57 

58 t0 = time.time() 

59 assert str(func()) == "1\n1\n1" 

60 used = time.time() - t0 

61 assert used < 0.2