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
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
|
"""Module containing the tests for requests_toolbelt.threaded.pool."""
try:
import queue # Python 3
except ImportError:
import Queue as queue
import unittest
try:
from unittest import mock
except ImportError:
import mock
import pytest
from requests_toolbelt.threaded import pool
from requests_toolbelt.threaded import thread
class TestPool(unittest.TestCase):
"""Collection of tests for requests_toolbelt.threaded.pool.Pool."""
def test_requires_positive_number_of_processes(self):
"""Show that the number of processes has to be > 0."""
with pytest.raises(ValueError):
pool.Pool(None, num_processes=0)
with pytest.raises(ValueError):
pool.Pool(None, num_processes=-1)
def test_number_of_processes_can_be_arbitrary(self):
"""Show that the number of processes can be set."""
job_queue = queue.Queue()
p = pool.Pool(job_queue, num_processes=100)
assert p._processes == 100
assert len(p._pool) == 100
job_queue = queue.Queue()
p = pool.Pool(job_queue, num_processes=1)
assert p._processes == 1
assert len(p._pool) == 1
def test_initializer_is_called(self):
"""Ensure that the initializer function is called."""
job_queue = queue.Queue()
initializer = mock.MagicMock()
pool.Pool(job_queue, num_processes=1, initializer=initializer)
assert initializer.called is True
initializer.assert_called_once_with(mock.ANY)
def test_auth_generator_is_called(self):
"""Ensure that the auth_generator function is called."""
job_queue = queue.Queue()
auth_generator = mock.MagicMock()
pool.Pool(job_queue, num_processes=1, auth_generator=auth_generator)
assert auth_generator.called is True
auth_generator.assert_called_once_with(mock.ANY)
def test_session_is_called(self):
"""Ensure that the session function is called."""
job_queue = queue.Queue()
session = mock.MagicMock()
pool.Pool(job_queue, num_processes=1, session=session)
assert session.called is True
session.assert_called_once_with()
def test_from_exceptions_populates_a_queue(self):
"""Ensure a Queue is properly populated from exceptions."""
urls = ["https://httpbin.org/get?n={}".format(n) for n in range(5)]
Exc = pool.ThreadException
excs = (Exc({'method': 'GET', 'url': url}, None) for url in urls)
job_queue = mock.MagicMock()
with mock.patch.object(queue, 'Queue', return_value=job_queue):
with mock.patch.object(thread, 'SessionThread'):
pool.Pool.from_exceptions(excs)
assert job_queue.put.call_count == 5
assert job_queue.put.mock_calls == [
mock.call({'method': 'GET', 'url': url})
for url in urls
]
def test_from_urls_constructs_get_requests(self):
"""Ensure a Queue is properly populated from an iterable of urls."""
urls = ["https://httpbin.org/get?n={}".format(n) for n in range(5)]
job_queue = mock.MagicMock()
with mock.patch.object(queue, 'Queue', return_value=job_queue):
with mock.patch.object(thread, 'SessionThread'):
pool.Pool.from_urls(urls)
assert job_queue.put.call_count == 5
assert job_queue.put.mock_calls == [
mock.call({'method': 'GET', 'url': url})
for url in urls
]
def test_from_urls_constructs_get_requests_with_kwargs(self):
"""Ensure a Queue is properly populated from an iterable of urls."""
def merge(*args):
final = {}
for d in args:
final.update(d)
return final
urls = ["https://httpbin.org/get?n={}".format(n) for n in range(5)]
kwargs = {'stream': True, 'headers': {'Accept': 'application/json'}}
job_queue = mock.MagicMock()
with mock.patch.object(queue, 'Queue', return_value=job_queue):
with mock.patch.object(thread, 'SessionThread'):
pool.Pool.from_urls(urls, kwargs)
assert job_queue.put.call_count == 5
assert job_queue.put.mock_calls == [
mock.call(merge({'method': 'GET', 'url': url}, kwargs))
for url in urls
]
def test_join_all(self):
"""Ensure that all threads are joined properly."""
session_threads = []
def _side_effect(*args, **kwargs):
thread = mock.MagicMock()
session_threads.append(thread)
return thread
with mock.patch.object(thread, 'SessionThread',
side_effect=_side_effect):
pool.Pool(None).join_all()
for st in session_threads:
st.join.assert_called_once_with()
def test_get_response_returns_thread_response(self):
"""Ensure that a ThreadResponse is made when there's data."""
queues = []
def _side_effect():
q = mock.MagicMock()
q.get_nowait.return_value = ({}, None)
queues.append(q)
return q
with mock.patch.object(queue, 'Queue', side_effect=_side_effect):
with mock.patch.object(thread, 'SessionThread'):
p = pool.Pool(None)
assert len(queues) == 2
assert isinstance(p.get_response(), pool.ThreadResponse)
assert len([q for q in queues if q.get_nowait.called]) == 1
def test_get_exception_returns_thread_exception(self):
"""Ensure that a ThreadException is made when there's data."""
queues = []
def _side_effect():
q = mock.MagicMock()
q.get_nowait.return_value = ({}, None)
queues.append(q)
return q
with mock.patch.object(queue, 'Queue', side_effect=_side_effect):
with mock.patch.object(thread, 'SessionThread'):
p = pool.Pool(None)
assert len(queues) == 2
assert isinstance(p.get_exception(), pool.ThreadException)
assert len([q for q in queues if q.get_nowait.called]) == 1
def test_get_response_returns_none_when_queue_is_empty(self):
"""Ensure that None is returned when the response Queue is empty."""
queues = []
def _side_effect():
q = mock.MagicMock()
q.get_nowait.side_effect = queue.Empty()
queues.append(q)
return q
with mock.patch.object(queue, 'Queue', side_effect=_side_effect):
with mock.patch.object(thread, 'SessionThread'):
p = pool.Pool(None)
assert len(queues) == 2
assert p.get_response() is None
assert len([q for q in queues if q.get_nowait.called]) == 1
def test_get_exception_returns_none_when_queue_is_empty(self):
"""Ensure that None is returned when the exception Queue is empty."""
queues = []
def _side_effect():
q = mock.MagicMock()
q.get_nowait.side_effect = queue.Empty()
queues.append(q)
return q
with mock.patch.object(queue, 'Queue', side_effect=_side_effect):
with mock.patch.object(thread, 'SessionThread'):
p = pool.Pool(None)
assert len(queues) == 2
assert p.get_exception() is None
assert len([q for q in queues if q.get_nowait.called]) == 1
def test_lists_are_correctly_returned(self):
"""Ensure that exceptions and responses return correct lists."""
def _make_queue():
q = queue.Queue()
q.put(({}, None))
return q
with mock.patch.object(thread, 'SessionThread'):
p = pool.Pool(None)
# Set up real queues.
p._response_queue = _make_queue()
p._exc_queue = _make_queue()
excs = list(p.exceptions())
assert len(excs) == 1
for exc in excs:
assert isinstance(exc, pool.ThreadException)
resps = list(p.responses())
assert len(resps) == 1
for resp in resps:
assert isinstance(resp, pool.ThreadResponse)
|
