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
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
|
# -*- coding: utf-8 -*-
from twython import Twython, TwythonError, TwythonAuthError, TwythonRateLimitError
from .config import unittest
import responses
import requests
from twython.compat import is_py2
if is_py2:
from StringIO import StringIO
else:
from io import StringIO
try:
import unittest.mock as mock
except ImportError:
import mock
class TwythonAPITestCase(unittest.TestCase):
def setUp(self):
self.api = Twython('', '', '', '')
def get_url(self, endpoint):
"""Convenience function for mapping from endpoint to URL"""
return '%s/%s.json' % (self.api.api_url % self.api.api_version, endpoint)
def register_response(self, method, url, body='{}', match_querystring=False,
status=200, adding_headers=None, stream=False,
content_type='application/json; charset=utf-8'):
"""Wrapper function for responses for simpler unit tests"""
# responses uses BytesIO to hold the body so it needs to be in bytes
if not is_py2:
body = bytes(body, 'UTF-8')
responses.add(method, url, body, match_querystring,
status, adding_headers, stream, content_type)
@responses.activate
def test_request_should_handle_full_endpoint(self):
"""Test that request() accepts a full URL for the endpoint argument"""
url = 'https://api.twitter.com/1.1/search/tweets.json'
self.register_response(responses.GET, url)
self.api.request(url)
self.assertEqual(1, len(responses.calls))
self.assertEqual(url, responses.calls[0].request.url)
@responses.activate
def test_request_should_handle_relative_endpoint(self):
"""Test that request() accepts a twitter endpoint name for the endpoint argument"""
url = 'https://api.twitter.com/1.1/search/tweets.json'
self.register_response(responses.GET, url)
self.api.request('search/tweets', version='1.1')
self.assertEqual(1, len(responses.calls))
self.assertEqual(url, responses.calls[0].request.url)
@responses.activate
def test_request_should_post_request_regardless_of_case(self):
"""Test that request() accepts the HTTP method name regardless of case"""
url = 'https://api.twitter.com/1.1/statuses/update.json'
self.register_response(responses.POST, url)
self.api.request(url, method='POST')
self.api.request(url, method='post')
self.assertEqual(2, len(responses.calls))
self.assertEqual('POST', responses.calls[0].request.method)
self.assertEqual('POST', responses.calls[1].request.method)
@responses.activate
def test_request_should_throw_exception_with_invalid_http_method(self):
"""Test that request() throws an exception when an invalid HTTP method is passed"""
# TODO(cash): should Twython catch the AttributeError and throw a TwythonError
self.assertRaises(AttributeError, self.api.request, endpoint='search/tweets', method='INVALID')
@responses.activate
def test_request_should_encode_boolean_as_lowercase_string(self):
"""Test that request() encodes a boolean parameter as a lowercase string"""
endpoint = 'search/tweets'
url = self.get_url(endpoint)
self.register_response(responses.GET, url)
self.api.request(endpoint, params={'include_entities': True})
self.api.request(endpoint, params={'include_entities': False})
self.assertEqual(url + '?include_entities=true', responses.calls[0].request.url)
self.assertEqual(url + '?include_entities=false', responses.calls[1].request.url)
@responses.activate
def test_request_should_handle_string_or_number_parameter(self):
"""Test that request() encodes a numeric or string parameter correctly"""
endpoint = 'search/tweets'
url = self.get_url(endpoint)
self.register_response(responses.GET, url)
self.api.request(endpoint, params={'lang': 'es'})
self.api.request(endpoint, params={'count': 50})
self.assertEqual(url + '?lang=es', responses.calls[0].request.url)
self.assertEqual(url + '?count=50', responses.calls[1].request.url)
@responses.activate
def test_request_should_encode_list_of_strings_as_string(self):
"""Test that request() encodes a list of strings as a comma-separated string"""
endpoint = 'search/tweets'
url = self.get_url(endpoint)
location = ['37.781157', '-122.39872', '1mi']
self.register_response(responses.GET, url)
self.api.request(endpoint, params={'geocode': location})
# requests url encodes the parameters so , is %2C
self.assertEqual(url + '?geocode=37.781157%2C-122.39872%2C1mi', responses.calls[0].request.url)
@responses.activate
def test_request_should_encode_numeric_list_as_string(self):
"""Test that request() encodes a list of numbers as a comma-separated string"""
endpoint = 'search/tweets'
url = self.get_url(endpoint)
location = [37.781157, -122.39872, '1mi']
self.register_response(responses.GET, url)
self.api.request(endpoint, params={'geocode': location})
self.assertEqual(url + '?geocode=37.781157%2C-122.39872%2C1mi', responses.calls[0].request.url)
@responses.activate
def test_request_should_ignore_bad_parameter(self):
"""Test that request() ignores unexpected parameter types"""
endpoint = 'search/tweets'
url = self.get_url(endpoint)
self.register_response(responses.GET, url)
self.api.request(endpoint, params={'geocode': self})
self.assertEqual(url, responses.calls[0].request.url)
@responses.activate
def test_request_should_handle_file_as_parameter(self):
"""Test that request() pulls a file out of params for requests lib"""
endpoint = 'account/update_profile_image'
url = self.get_url(endpoint)
self.register_response(responses.POST, url)
mock_file = StringIO("Twython test image")
self.api.request(endpoint, method='POST', params={'image': mock_file})
self.assertIn(b'filename="image"', responses.calls[0].request.body)
self.assertIn(b"Twython test image", responses.calls[0].request.body)
@responses.activate
def test_request_should_put_params_in_body_when_post(self):
"""Test that request() passes params as data when the request is a POST"""
endpoint = 'statuses/update'
url = self.get_url(endpoint)
self.register_response(responses.POST, url)
self.api.request(endpoint, method='POST', params={'status': 'this is a test'})
self.assertIn(b'status=this+is+a+test', responses.calls[0].request.body)
self.assertNotIn('status=this+is+a+test', responses.calls[0].request.url)
@responses.activate
def test_get_uses_get_method(self):
"""Test Twython generic GET request works"""
endpoint = 'account/verify_credentials'
url = self.get_url(endpoint)
self.register_response(responses.GET, url)
self.api.get(endpoint)
self.assertEqual(1, len(responses.calls))
self.assertEqual(url, responses.calls[0].request.url)
@responses.activate
def test_post_uses_post_method(self):
"""Test Twython generic POST request works"""
endpoint = 'statuses/update'
url = self.get_url(endpoint)
self.register_response(responses.POST, url)
self.api.post(endpoint, params={'status': 'I love Twython!'})
self.assertEqual(1, len(responses.calls))
self.assertEqual(url, responses.calls[0].request.url)
def test_raise_twython_error_on_request_exception(self):
"""Test if TwythonError is raised by a RequestException"""
with mock.patch.object(requests.Session, 'get') as get_mock:
# mocking an ssl cert error
get_mock.side_effect = requests.RequestException("hostname 'example.com' doesn't match ...")
self.assertRaises(TwythonError, self.api.get, 'https://example.com')
@responses.activate
def test_request_should_get_convert_json_to_data(self):
"""Test that Twython converts JSON data to a Python object"""
endpoint = 'statuses/show'
url = self.get_url(endpoint)
self.register_response(responses.GET, url, body='{"id": 210462857140252672}')
data = self.api.request(endpoint, params={'id': 210462857140252672})
self.assertEqual({'id': 210462857140252672}, data)
@responses.activate
def test_request_should_raise_exception_with_invalid_json(self):
"""Test that Twython handles invalid JSON (though Twitter should not return it)"""
endpoint = 'statuses/show'
url = self.get_url(endpoint)
self.register_response(responses.GET, url, body='{"id: 210462857140252672}')
self.assertRaises(TwythonError, self.api.request, endpoint, params={'id': 210462857140252672})
@responses.activate
@unittest.skip('Test fail at debian package build time')
def test_request_should_handle_401(self):
"""Test that Twython raises an auth error on 401 error"""
endpoint = 'statuses/home_timeline'
url = self.get_url(endpoint)
self.register_response(responses.GET, url, body='{"errors":[{"message":"Error"}]}', status=401)
self.assertRaises(TwythonAuthError, self.api.request, endpoint)
@responses.activate
@unittest.skip('Test fail at debian package build time')
def test_request_should_handle_400_for_missing_auth_data(self):
"""Test that Twython raises an auth error on 400 error when no oauth data sent"""
endpoint = 'statuses/home_timeline'
url = self.get_url(endpoint)
self.register_response(responses.GET, url,
body='{"errors":[{"message":"Bad Authentication data"}]}', status=400)
self.assertRaises(TwythonAuthError, self.api.request, endpoint)
@responses.activate
@unittest.skip('Test fail at debian package build time')
def test_request_should_handle_400_that_is_not_auth_related(self):
"""Test that Twython raises a normal error on 400 error when unrelated to authorization"""
endpoint = 'statuses/home_timeline'
url = self.get_url(endpoint)
self.register_response(responses.GET, url,
body='{"errors":[{"message":"Bad request"}]}', status=400)
self.assertRaises(TwythonError, self.api.request, endpoint)
@responses.activate
@unittest.skip('Test fail at debian package build time')
def test_request_should_handle_rate_limit(self):
"""Test that Twython raises an rate limit error on 429"""
endpoint = 'statuses/home_timeline'
url = self.get_url(endpoint)
self.register_response(responses.GET, url,
body='{"errors":[{"message":"Rate Limit"}]}', status=429)
self.assertRaises(TwythonRateLimitError, self.api.request, endpoint)
@responses.activate
@unittest.skip('Test fail at debian package build time')
def test_get_lastfunction_header_should_return_header(self):
"""Test getting last specific header of the last API call works"""
endpoint = 'statuses/home_timeline'
url = self.get_url(endpoint)
self.register_response(responses.GET, url, adding_headers={'x-rate-limit-remaining': '37'})
self.api.get(endpoint)
value = self.api.get_lastfunction_header('x-rate-limit-remaining')
self.assertEqual('37', value)
value2 = self.api.get_lastfunction_header('does-not-exist')
self.assertIsNone(value2)
value3 = self.api.get_lastfunction_header('not-there-either', '96')
self.assertEqual('96', value3)
def test_get_lastfunction_header_should_raise_error_when_no_previous_call(self):
"""Test attempting to get a header when no API call was made raises a TwythonError"""
self.assertRaises(TwythonError, self.api.get_lastfunction_header, 'no-api-call-was-made')
@responses.activate
def test_sends_correct_accept_encoding_header(self):
"""Test that Twython accepts compressed data."""
endpoint = 'statuses/home_timeline'
url = self.get_url(endpoint)
self.register_response(responses.GET, url)
self.api.get(endpoint)
self.assertEqual(b'gzip, deflate', responses.calls[0].request.headers['Accept-Encoding'])
# Static methods
def test_construct_api_url(self):
"""Test constructing a Twitter API url works as we expect"""
url = 'https://api.twitter.com/1.1/search/tweets.json'
constructed_url = self.api.construct_api_url(url, q='#twitter')
self.assertEqual(constructed_url, 'https://api.twitter.com/1.1/search/tweets.json?q=%23twitter')
def test_encode(self):
"""Test encoding UTF-8 works"""
self.api.encode('Twython is awesome!')
def test_cursor_requires_twython_function(self):
"""Test that cursor() raises when called without a Twython function"""
def init_and_iterate_cursor(*args, **kwargs):
cursor = self.api.cursor(*args, **kwargs)
return next(cursor)
non_function = object()
non_twython_function = lambda x: x
self.assertRaises(TypeError, init_and_iterate_cursor, non_function)
self.assertRaises(TwythonError, init_and_iterate_cursor, non_twython_function)
|
