Skip to content

Commit fb73aaf

Browse files
committed
test: cover osw.utils.util.parallelize
- result ordering, kwargs forwarding, empty and non-list iterables - concurrency guarded by a barrier instead of wall-clock timing - both failure modes: propagate first exception, or collect in place - the already-running event loop path used from notebooks - closes #25
1 parent 0d32bc2 commit fb73aaf

1 file changed

Lines changed: 114 additions & 0 deletions

File tree

‎tests/utils/parallelize_test.py‎

Lines changed: 114 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,114 @@
1+
"""Unit tests for osw.utils.util.parallelize.
2+
3+
Covers issue #25. The behaviour that callers depend on is result ordering,
4+
kwargs forwarding, and the two failure modes: abort on the first exception, or
5+
collect exceptions in place so the batch completes with results still aligned
6+
to the input.
7+
"""
8+
9+
import asyncio
10+
import threading
11+
12+
import pytest
13+
14+
from osw.utils.util import parallelize
15+
16+
# Generous enough that a loaded CI machine still passes, short enough that a
17+
# serialized execution trips it instead of hanging the suite.
18+
BARRIER_TIMEOUT = 30
19+
20+
21+
def _double(item):
22+
return item * 2
23+
24+
25+
def _add(item, offset=0, factor=1):
26+
return item * factor + offset
27+
28+
29+
def _fail_on_even(item):
30+
if item % 2 == 0:
31+
raise ValueError(f"boom on {item}")
32+
return item
33+
34+
35+
def test_returns_results_in_input_order():
36+
"""Results map back to the input positionally, not by completion time."""
37+
delays = [0.05, 0.0, 0.03, 0.0]
38+
39+
def sleep_then_return(item):
40+
import time
41+
42+
time.sleep(delays[item])
43+
return item
44+
45+
assert parallelize(sleep_then_return, [0, 1, 2, 3]) == [0, 1, 2, 3]
46+
47+
48+
def test_maps_the_function_over_the_iterable():
49+
assert parallelize(_double, [1, 2, 3]) == [2, 4, 6]
50+
51+
52+
def test_forwards_kwargs_to_the_function():
53+
assert parallelize(_add, [1, 2, 3], offset=10, factor=2) == [12, 14, 16]
54+
55+
56+
def test_empty_iterable_returns_empty_list():
57+
assert parallelize(_double, []) == []
58+
59+
60+
def test_accepts_any_iterable_not_only_a_list():
61+
assert parallelize(_double, range(3)) == [0, 2, 4]
62+
63+
64+
def test_tasks_run_concurrently():
65+
"""A barrier that only clears if all items are in flight at once."""
66+
barrier = threading.Barrier(4, timeout=BARRIER_TIMEOUT)
67+
68+
def wait_for_the_others(item):
69+
barrier.wait()
70+
return item
71+
72+
assert parallelize(wait_for_the_others, [1, 2, 3, 4]) == [1, 2, 3, 4]
73+
74+
75+
def test_first_exception_propagates_by_default():
76+
with pytest.raises(ValueError, match="boom on 2"):
77+
parallelize(_fail_on_even, [1, 2, 3])
78+
79+
80+
def test_return_exceptions_keeps_results_aligned():
81+
"""Regression guard: a failing item must not drop its neighbours."""
82+
results = parallelize(_fail_on_even, [1, 2, 3, 4, 5], return_exceptions=True)
83+
84+
assert len(results) == 5
85+
assert results[0] == 1
86+
assert isinstance(results[1], ValueError)
87+
assert results[2] == 3
88+
assert isinstance(results[3], ValueError)
89+
assert results[4] == 5
90+
91+
92+
def test_return_exceptions_completes_every_task():
93+
"""Every item is attempted even though the first one raises."""
94+
attempted = []
95+
lock = threading.Lock()
96+
97+
def record_then_fail(item):
98+
with lock:
99+
attempted.append(item)
100+
raise RuntimeError(f"boom on {item}")
101+
102+
results = parallelize(record_then_fail, [1, 2, 3], return_exceptions=True)
103+
104+
assert sorted(attempted) == [1, 2, 3]
105+
assert all(isinstance(r, RuntimeError) for r in results)
106+
107+
108+
def test_works_inside_a_running_event_loop():
109+
"""The Jupyter case: parallelize is called while a loop is already running."""
110+
111+
async def caller():
112+
return parallelize(_double, [1, 2, 3])
113+
114+
assert asyncio.run(caller()) == [2, 4, 6]

0 commit comments

Comments
 (0)