forked from bear/python-twitter
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathtest_api.py
More file actions
326 lines (273 loc) · 13.7 KB
/
test_api.py
File metadata and controls
326 lines (273 loc) · 13.7 KB
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
318
319
320
321
322
323
324
325
326
# encoding: utf-8
import os
import time
import urllib
import unittest
import twitter
CONSUMER_KEY = os.getenv('CONSUMER_KEY', None)
CONSUMER_SECRET = os.getenv('CONSUMER_SECRET', None)
ACCESS_TOKEN_KEY = os.getenv('ACCESS_TOKEN_KEY', None)
ACCESS_TOKEN_SECRET = os.getenv('ACCESS_TOKEN_SECRET', None)
@unittest.skipIf(not CONSUMER_KEY and not CONSUMER_SECRET, "No tokens provided")
class ApiTest(unittest.TestCase):
def setUp(self):
self._urllib = MockUrllib()
time.sleep(15)
api = twitter.Api(consumer_key=CONSUMER_SECRET,
consumer_secret=CONSUMER_SECRET,
access_token_key=ACCESS_TOKEN_KEY,
access_token_secret=ACCESS_TOKEN_SECRET,
cache=None)
api.SetUrllib(self._urllib)
self._api = api
print("Testing the API class. This test is time controlled")
def testTwitterError(self):
'''Test that twitter responses containing an error message are wrapped.'''
self._AddHandler('https://api.twitter.com/1.1/statuses/user_timeline.json',
curry(self._OpenTestData, 'public_timeline_error.json'))
# Manually try/catch so we can check the exception's value
try:
self._api.GetUserTimeline()
except twitter.TwitterError as error:
# If the error message matches, the test passes
self.assertEqual('test error', error.message)
else:
self.fail('TwitterError expected')
def testGetUserTimeline(self):
'''Test the twitter.Api GetUserTimeline method'''
time.sleep(8)
print('Testing GetUserTimeline')
self._AddHandler('https://api.twitter.com/1.1/statuses/user_timeline.json?count=1&screen_name=kesuke',
curry(self._OpenTestData, 'user_timeline-kesuke.json'))
statuses = self._api.GetUserTimeline(screen_name='kesuke', count=1)
# This is rather arbitrary, but spot checking is better than nothing
self.assertEqual(89512102, statuses[0].id)
self.assertEqual(718443, statuses[0].user.id)
# def testGetFriendsTimeline(self):
# '''Test the twitter.Api GetFriendsTimeline method'''
# self._AddHandler('https://api.twitter.com/1.1/statuses/friends_timeline/kesuke.json',
# curry(self._OpenTestData, 'friends_timeline-kesuke.json'))
# statuses = self._api.GetFriendsTimeline('kesuke')
# # This is rather arbitrary, but spot checking is better than nothing
# self.assertEqual(20, len(statuses))
# self.assertEqual(718443, statuses[0].user.id)
def testGetStatus(self):
'''Test the twitter.Api GetStatus method'''
time.sleep(8)
print('Testing GetStatus')
self._AddHandler('https://api.twitter.com/1.1/statuses/show.json?include_my_retweet=1&id=89512102',
curry(self._OpenTestData, 'show-89512102.json'))
status = self._api.GetStatus(89512102)
self.assertEqual(89512102, status.id)
self.assertEqual(718443, status.user.id)
def testDestroyStatus(self):
'''Test the twitter.Api DestroyStatus method'''
time.sleep(8)
print('Testing DestroyStatus')
self._AddHandler('https://api.twitter.com/1.1/statuses/destroy/103208352.json',
curry(self._OpenTestData, 'status-destroy.json'))
status = self._api.DestroyStatus(103208352)
self.assertEqual(103208352, status.id)
def testPostUpdate(self):
'''Test the twitter.Api PostUpdate method'''
time.sleep(8)
print('Testing PostUpdate')
self._AddHandler('https://api.twitter.com/1.1/statuses/update.json',
curry(self._OpenTestData, 'update.json'))
status = self._api.PostUpdate(u'Моё судно на воздушной подушке полно угрей'.encode('utf8'))
# This is rather arbitrary, but spot checking is better than nothing
self.assertEqual(u'Моё судно на воздушной подушке полно угрей', status.text)
def testPostRetweet(self):
'''Test the twitter.Api PostRetweet method'''
time.sleep(8)
print('Testing PostRetweet')
self._AddHandler('https://api.twitter.com/1.1/statuses/retweet/89512102.json',
curry(self._OpenTestData, 'retweet.json'))
status = self._api.PostRetweet(89512102)
self.assertEqual(89512102, status.id)
def testPostUpdateLatLon(self):
'''Test the twitter.Api PostUpdate method, when used in conjunction with latitude and longitude'''
time.sleep(8)
print('Testing PostUpdateLatLon')
self._AddHandler('https://api.twitter.com/1.1/statuses/update.json',
curry(self._OpenTestData, 'update_latlong.json'))
# test another update with geo parameters, again test somewhat arbitrary
status = self._api.PostUpdate(u'Моё судно на воздушной подушке полно угрей'.encode('utf8'), latitude=54.2,
longitude=-2)
self.assertEqual(u'Моё судно на воздушной подушке полно угрей', status.text)
self.assertEqual(u'Point', status.GetGeo()['type'])
self.assertEqual(26.2, status.GetGeo()['coordinates'][0])
self.assertEqual(127.5, status.GetGeo()['coordinates'][1])
def testGetReplies(self):
'''Test the twitter.Api GetReplies method'''
time.sleep(8)
print('Testing GetReplies')
self._AddHandler('https://api.twitter.com/1.1/statuses/user_timeline.json',
curry(self._OpenTestData, 'replies.json'))
statuses = self._api.GetReplies()
self.assertEqual(36657062, statuses[0].id)
def testGetRetweetsOfMe(self):
'''Test the twitter.API GetRetweetsOfMe method'''
time.sleep(8)
print('Testing GetRetweetsOfMe')
self._AddHandler('https://api.twitter.com/1.1/statuses/retweets_of_me.json',
curry(self._OpenTestData, 'retweets_of_me.json'))
retweets = self._api.GetRetweetsOfMe()
self.assertEqual(253650670274637824, retweets[0].id)
def testGetFriends(self):
'''Test the twitter.Api GetFriends method'''
time.sleep(8)
print('Testing GetFriends')
self._AddHandler('https://api.twitter.com/1.1/friends/list.json?cursor=123',
curry(self._OpenTestData, 'friends.json'))
users = self._api.GetFriends(cursor=123)
buzz = [u.status for u in users if u.screen_name == 'buzz']
self.assertEqual(89543882, buzz[0].id)
def testGetFollowers(self):
'''Test the twitter.Api GetFollowers method'''
time.sleep(8)
print('Testing GetFollowers')
self._AddHandler('https://api.twitter.com/1.1/followers/list.json?cursor=-1',
curry(self._OpenTestData, 'followers.json'))
users = self._api.GetFollowers()
# This is rather arbitrary, but spot checking is better than nothing
alexkingorg = [u.status for u in users if u.screen_name == 'alexkingorg']
self.assertEqual(89554432, alexkingorg[0].id)
# def testGetFeatured(self):
# '''Test the twitter.Api GetFeatured method'''
# self._AddHandler('https://api.twitter.com/1.1/statuses/featured.json',
# curry(self._OpenTestData, 'featured.json'))
# users = self._api.GetFeatured()
# # This is rather arbitrary, but spot checking is better than nothing
# stevenwright = [u.status for u in users if u.screen_name == 'stevenwright']
# self.assertEqual(86991742, stevenwright[0].id)
def testGetDirectMessages(self):
'''Test the twitter.Api GetDirectMessages method'''
time.sleep(8)
print('Testing GetDirectMessages')
self._AddHandler('https://api.twitter.com/1.1/direct_messages.json',
curry(self._OpenTestData, 'direct_messages.json'))
statuses = self._api.GetDirectMessages()
self.assertEqual(u'A légpárnás hajóm tele van angolnákkal.', statuses[0].text)
def testPostDirectMessage(self):
'''Test the twitter.Api PostDirectMessage method'''
time.sleep(8)
print('Testing PostDirectMessage')
self._AddHandler('https://api.twitter.com/1.1/direct_messages/new.json',
curry(self._OpenTestData, 'direct_messages-new.json'))
status = self._api.PostDirectMessage('test', u'Моё судно на воздушной подушке полно угрей'.encode('utf8'))
# This is rather arbitrary, but spot checking is better than nothing
self.assertEqual(u'Моё судно на воздушной подушке полно угрей', status.text)
def testDestroyDirectMessage(self):
'''Test the twitter.Api DestroyDirectMessage method'''
time.sleep(8)
print('Testing DestroyDirectMessage')
self._AddHandler('https://api.twitter.com/1.1/direct_messages/destroy.json',
curry(self._OpenTestData, 'direct_message-destroy.json'))
status = self._api.DestroyDirectMessage(3496342)
# This is rather arbitrary, but spot checking is better than nothing
self.assertEqual(673483, status.sender_id)
def testCreateFriendship(self):
'''Test the twitter.Api CreateFriendship method'''
time.sleep(8)
print('Testing CreateFriendship')
self._AddHandler('https://api.twitter.com/1.1/friendships/create.json',
curry(self._OpenTestData, 'friendship-create.json'))
user = self._api.CreateFriendship('dewitt')
# This is rather arbitrary, but spot checking is better than nothing
self.assertEqual(673483, user.id)
def testDestroyFriendship(self):
'''Test the twitter.Api DestroyFriendship method'''
time.sleep(8)
print('Testing Destroy Friendship')
self._AddHandler('https://api.twitter.com/1.1/friendships/destroy.json',
curry(self._OpenTestData, 'friendship-destroy.json'))
user = self._api.DestroyFriendship('dewitt')
# This is rather arbitrary, but spot checking is better than nothing
self.assertEqual(673483, user.id)
def testGetUser(self):
'''Test the twitter.Api GetUser method'''
time.sleep(8)
print('Testing GetUser')
self._AddHandler('https://api.twitter.com/1.1/users/show.json?user_id=dewitt',
curry(self._OpenTestData, 'show-dewitt.json'))
user = self._api.GetUser('dewitt')
self.assertEqual('dewitt', user.screen_name)
self.assertEqual(89586072, user.status.id)
def _AddHandler(self, url, callback):
self._urllib.AddHandler(url, callback)
def _GetTestDataPath(self, filename):
directory = os.path.dirname(os.path.abspath(__file__))
test_data_dir = os.path.join(directory, 'testdata')
return os.path.join(test_data_dir, filename)
def _OpenTestData(self, filename):
f = open(self._GetTestDataPath(filename))
# make sure that the returned object contains an .info() method:
# headers are set to {}
return urllib.addinfo(f, {})
class MockUrllib(object):
'''A mock replacement for urllib that hardcodes specific responses.'''
def __init__(self):
self._handlers = {}
self.HTTPBasicAuthHandler = MockHTTPBasicAuthHandler
def AddHandler(self, url, callback):
self._handlers[url] = callback
def build_opener(self, *handlers):
return MockOpener(self._handlers)
def HTTPHandler(self, *args, **kwargs):
return None
def HTTPSHandler(self, *args, **kwargs):
return None
def OpenerDirector(self):
return self.build_opener()
def ProxyHandler(self, *args, **kwargs):
return None
class MockOpener(object):
'''A mock opener for urllib'''
def __init__(self, handlers):
self._handlers = handlers
self._opened = False
def open(self, url, data=None):
if self._opened:
raise Exception('MockOpener already opened.')
# Remove parameters from URL - they're only added by oauth and we
# don't want to test oauth
if '?' in url:
# We split using & and filter on the beginning of each key
# This is crude but we have to keep the ordering for now
(url, qs) = url.split('?')
tokens = [token for token in qs.split('&')
if not token.startswith('oauth')]
if len(tokens) > 0:
url = "%s?%s" % (url, '&'.join(tokens))
if url in self._handlers:
self._opened = True
return self._handlers[url]()
else:
print(url)
print(self._handlers)
raise Exception('Unexpected URL %s (Checked: %s)' % (url, self._handlers))
def add_handler(self, *args, **kwargs):
pass
def close(self):
if not self._opened:
raise Exception('MockOpener closed before it was opened.')
self._opened = False
class curry(object):
# http://aspn.activestate.com/ASPN/Cookbook/Python/Recipe/52549
def __init__(self, fun, *args, **kwargs):
self.fun = fun
self.pending = args[:]
self.kwargs = kwargs.copy()
def __call__(self, *args, **kwargs):
if kwargs and self.kwargs:
kw = self.kwargs.copy()
kw.update(kwargs)
else:
kw = kwargs or self.kwargs
return self.fun(*(self.pending + args), **kw)
class MockHTTPBasicAuthHandler(object):
'''A mock replacement for HTTPBasicAuthHandler'''
def add_password(self, realm, uri, user, passwd):
# TODO(dewitt): Add verification that the proper args are passed
pass