1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
|
"""
Tests for `tqdm.contrib.concurrent`.
"""
from pytest import warns
from tqdm.contrib.concurrent import process_map, thread_map
from .tests_tqdm import StringIO, TqdmWarning, closing, importorskip, mark, skip
def incr(x):
"""Dummy function"""
return x + 1
def test_thread_map():
"""Test contrib.concurrent.thread_map"""
with closing(StringIO()) as our_file:
a = range(9)
b = [i + 1 for i in a]
try:
assert thread_map(lambda x: x + 1, a, file=our_file) == b
except ImportError as err:
skip(str(err))
assert thread_map(incr, a, file=our_file) == b
def test_process_map():
"""Test contrib.concurrent.process_map"""
with closing(StringIO()) as our_file:
a = range(9)
b = [i + 1 for i in a]
try:
assert process_map(incr, a, file=our_file) == b
except ImportError as err:
skip(str(err))
@mark.parametrize("iterables,should_warn", [([], False), (['x'], False), ([()], False),
(['x', ()], False), (['x' * 1001], True),
(['x' * 100, ('x',) * 1001], True)])
def test_chunksize_warning(iterables, should_warn):
"""Test contrib.concurrent.process_map chunksize warnings"""
patch = importorskip('unittest.mock').patch
with patch('tqdm.contrib.concurrent._executor_map'):
if should_warn:
warns(TqdmWarning, process_map, incr, *iterables)
else:
process_map(incr, *iterables)
|
