test_api.py 17.1 KB
Newer Older
Mark Smith's avatar
Mark Smith committed
1
import unittest
2
import random
Mark Smith's avatar
Mark Smith committed
3
import shutil
4
import time
Josh Roesslein's avatar
Josh Roesslein committed
5
import os
6
from ast import literal_eval
7

Joshua Roesslein's avatar
Joshua Roesslein committed
8
9
from nose import SkipTest

10
11
from tweepy import Friendship, MemoryCache, FileCache, API
from tweepy.parsers import Parser
Mark Smith's avatar
Mark Smith committed
12
from .config import TweepyTestCase, username, use_replay, tape
Josh Roesslein's avatar
Josh Roesslein committed
13

14
test_tweet_id = '266367358078169089'
15
tweet_text = 'testing 1000'
16

17
18
"""Unit tests"""

Mark Smith's avatar
Mark Smith committed
19

20
21
22
23
24
25
26
27
28
29
30
31
class TweepyErrorTests(unittest.TestCase):

    def testpickle(self):
        """Verify exceptions can be pickled and unpickled."""
        import pickle
        from tweepy.error import TweepError

        e = TweepError('no reason', {'status': 200})
        e2 = pickle.loads(pickle.dumps(e))

        self.assertEqual(e.reason, e2.reason)
        self.assertEqual(e.response, e2.response)
Josh Roesslein's avatar
Josh Roesslein committed
32

Mark Smith's avatar
Mark Smith committed
33

34
class TweepyAPITests(TweepyTestCase):
Josh Roesslein's avatar
Josh Roesslein committed
35

36
37
38
39
40
41
42
43
44
45
46
47
    @tape.use_cassette('testfailure.json')
    def testapierror(self):
        from tweepy.error import TweepError

        with self.assertRaises(TweepError) as cm:
            self.api.direct_messages()

        reason, = literal_eval(cm.exception.reason)
        self.assertEqual(reason['message'], 'Bad Authentication data.')
        self.assertEqual(reason['code'], 215)
        self.assertEqual(cm.exception.api_code, 215)

Miriam Sexton's avatar
Miriam Sexton committed
48
    # TODO: Actually have some sort of better assertion
Mark Smith's avatar
Mark Smith committed
49
    @tape.use_cassette('testgetoembed.json')
Miriam Sexton's avatar
Miriam Sexton committed
50
    def testgetoembed(self):
Miriam Sexton's avatar
Miriam Sexton committed
51
        data = self.api.get_oembed(test_tweet_id)
Miriam Sexton's avatar
Miriam Sexton committed
52
53
        self.assertEqual(data['author_name'], "Twitter")

Mark Smith's avatar
Mark Smith committed
54
    @tape.use_cassette('testparserargumenthastobeaparserinstance.json')
55
56
57
    def testparserargumenthastobeaparserinstance(self):
        """ Testing the issue https://github.com/tweepy/tweepy/issues/421"""
        self.assertRaises(TypeError, API, self.auth, parser=Parser)
Miriam Sexton's avatar
Miriam Sexton committed
58

Mark Smith's avatar
Mark Smith committed
59
    @tape.use_cassette('testhometimeline.json')
60
61
62
    def testhometimeline(self):
        self.api.home_timeline()

Mark Smith's avatar
Mark Smith committed
63
    @tape.use_cassette('testusertimeline.json')
Josh Roesslein's avatar
Josh Roesslein committed
64
    def testusertimeline(self):
65
66
        self.api.user_timeline()
        self.api.user_timeline('twitter')
Josh Roesslein's avatar
Josh Roesslein committed
67

Mark Smith's avatar
Mark Smith committed
68
    @tape.use_cassette('testmentionstimeline.json')
Joshua Roesslein's avatar
Joshua Roesslein committed
69
70
    def testmentionstimeline(self):
        self.api.mentions_timeline()
71

Mark Smith's avatar
Mark Smith committed
72
    @tape.use_cassette('testretweetsofme.json')
73
74
75
76
    def testretweetsofme(self):
        self.api.retweets_of_me()

    def testretweet(self):
Joshua Roesslein's avatar
Joshua Roesslein committed
77
78
        # TODO(josh): Need a way to get random tweets to retweet.
        raise SkipTest()
79

Mark Smith's avatar
Mark Smith committed
80
    @tape.use_cassette('testretweets.json')
81
    def testretweets(self):
82
        self.api.retweets(test_tweet_id)
83

Mark Smith's avatar
Mark Smith committed
84
    @tape.use_cassette('testretweeters.json')
Joshua Roesslein's avatar
Joshua Roesslein committed
85
86
87
    def testretweeters(self):
        self.api.retweeters(test_tweet_id)

Mark Smith's avatar
Mark Smith committed
88
    @tape.use_cassette('testgetstatus.json')
Josh Roesslein's avatar
Josh Roesslein committed
89
    def testgetstatus(self):
90
        self.api.get_status(id=test_tweet_id)
Josh Roesslein's avatar
Josh Roesslein committed
91

Mark Smith's avatar
Mark Smith committed
92
    @tape.use_cassette('testupdateanddestroystatus.json')
Josh Roesslein's avatar
Josh Roesslein committed
93
94
    def testupdateanddestroystatus(self):
        # test update
95
        text = tweet_text if use_replay else 'testing %i' % random.randint(0, 1000)
Josh Roesslein's avatar
Josh Roesslein committed
96
97
98
99
100
101
102
        update = self.api.update_status(status=text)
        self.assertEqual(update.text, text)

        # test destroy
        deleted = self.api.destroy_status(id=update.id)
        self.assertEqual(deleted.id, update.id)

103
104
105
106
107
108
109
110
111
112
113
    @tape.use_cassette('testupdateanddestroystatus.json')
    def testupdateanddestroystatuswithoutkwarg(self):
        # test update, passing text as a positional argument (#554)
        text = tweet_text if use_replay else 'testing %i' % random.randint(0, 1000)
        update = self.api.update_status(text)
        self.assertEqual(update.text, text)

        # test destroy
        deleted = self.api.destroy_status(id=update.id)
        self.assertEqual(deleted.id, update.id)

Mark Smith's avatar
Mark Smith committed
114
    @tape.use_cassette('testupdatestatuswithmedia.yaml', serializer='yaml')
115
116
    def testupdatestatuswithmedia(self):
        update = self.api.update_with_media('examples/banner.png', status=tweet_text)
Aaron Hill's avatar
Aaron Hill committed
117
        self.assertIn(tweet_text + ' http://t.co', update.text)
118

Mark Smith's avatar
Mark Smith committed
119
    @tape.use_cassette('testgetuser.json')
Josh Roesslein's avatar
Josh Roesslein committed
120
    def testgetuser(self):
121
122
123
124
        u = self.api.get_user('twitter')
        self.assertEqual(u.screen_name, 'twitter')

        u = self.api.get_user(783214)
Josh Roesslein's avatar
Josh Roesslein committed
125
126
        self.assertEqual(u.screen_name, 'twitter')

Mark Smith's avatar
Mark Smith committed
127
    @tape.use_cassette('testlookupusers.json')
128
129
130
131
132
133
    def testlookupusers(self):
        def check(users):
            self.assertEqual(len(users), 2)
        check(self.api.lookup_users(user_ids=[6844292, 6253282]))
        check(self.api.lookup_users(screen_names=['twitterapi', 'twitter']))

Mark Smith's avatar
Mark Smith committed
134
    @tape.use_cassette('testsearchusers.json')
135
136
137
    def testsearchusers(self):
        self.api.search_users('twitter')

Mark Smith's avatar
Mark Smith committed
138
    @tape.use_cassette('testsuggestedcategories.json')
139
140
141
    def testsuggestedcategories(self):
        self.api.suggested_categories()

Mark Smith's avatar
Mark Smith committed
142
    @tape.use_cassette('testsuggestedusers.json')
143
144
145
146
147
    def testsuggestedusers(self):
        categories = self.api.suggested_categories()
        if len(categories) != 0:
            self.api.suggested_users(categories[0].slug)

Mark Smith's avatar
Mark Smith committed
148
    @tape.use_cassette('testsuggesteduserstweets.json')
149
150
151
152
    def testsuggesteduserstweets(self):
        categories = self.api.suggested_categories()
        if len(categories) != 0:
            self.api.suggested_users_tweets(categories[0].slug)
153

Mark Smith's avatar
Mark Smith committed
154
    @tape.use_cassette('testme.json')
Josh Roesslein's avatar
Josh Roesslein committed
155
156
157
158
    def testme(self):
        me = self.api.me()
        self.assertEqual(me.screen_name, username)

Mark Smith's avatar
Mark Smith committed
159
    @tape.use_cassette('testdirectmessages.json')
Josh Roesslein's avatar
Josh Roesslein committed
160
    def testdirectmessages(self):
161
162
        self.api.direct_messages()

Mark Smith's avatar
Mark Smith committed
163
    @tape.use_cassette('testsentdirectmessages.json')
164
165
    def testsentdirectmessages(self):
        self.api.sent_direct_messages()
Josh Roesslein's avatar
Josh Roesslein committed
166

Mark Smith's avatar
Mark Smith committed
167
    @tape.use_cassette('testsendanddestroydirectmessage.json')
Josh Roesslein's avatar
Josh Roesslein committed
168
169
    def testsendanddestroydirectmessage(self):
        # send
Joshua Roesslein's avatar
Joshua Roesslein committed
170
        sent_dm = self.api.send_direct_message(username, text='test message')
Josh Roesslein's avatar
Josh Roesslein committed
171
172
173
174
175
176
177
178
179
180
181
        self.assertEqual(sent_dm.text, 'test message')
        self.assertEqual(sent_dm.sender.screen_name, username)
        self.assertEqual(sent_dm.recipient.screen_name, username)

        # destroy
        destroyed_dm = self.api.destroy_direct_message(sent_dm.id)
        self.assertEqual(destroyed_dm.text, sent_dm.text)
        self.assertEqual(destroyed_dm.id, sent_dm.id)
        self.assertEqual(destroyed_dm.sender.screen_name, username)
        self.assertEqual(destroyed_dm.recipient.screen_name, username)

Mark Smith's avatar
Mark Smith committed
182
    @tape.use_cassette('testcreatedestroyfriendship.json')
183
    def testcreatedestroyfriendship(self):
Josh Roesslein's avatar
Josh Roesslein committed
184
185
186
        enemy = self.api.destroy_friendship('twitter')
        self.assertEqual(enemy.screen_name, 'twitter')

187
188
189
190
        # Wait 5 seconds to allow Twitter time
        # to process the friendship destroy request.
        sleep(5)

191
192
193
        friend = self.api.create_friendship('twitter')
        self.assertEqual(friend.screen_name, 'twitter')

Mark Smith's avatar
Mark Smith committed
194
    @tape.use_cassette('testshowfriendship.json')
Josh Roesslein's avatar
Josh Roesslein committed
195
    def testshowfriendship(self):
Joshua Roesslein's avatar
Joshua Roesslein committed
196
        source, target = self.api.show_friendship(target_screen_name='twitter')
Josh Roesslein's avatar
Josh Roesslein committed
197
198
199
        self.assert_(isinstance(source, Friendship))
        self.assert_(isinstance(target, Friendship))

Mark Smith's avatar
Mark Smith committed
200
    @tape.use_cassette('testfriendsids.json')
201
202
203
    def testfriendsids(self):
        self.api.friends_ids(username)

Mark Smith's avatar
Mark Smith committed
204
    @tape.use_cassette('testfollowersids.json')
205
206
207
    def testfollowersids(self):
        self.api.followers_ids(username)

Mark Smith's avatar
Mark Smith committed
208
    @tape.use_cassette('testfriends.json')
209
210
211
    def testfriends(self):
        self.api.friends(username)

Mark Smith's avatar
Mark Smith committed
212
    @tape.use_cassette('testfollowers.json')
213
214
215
    def testfollowers(self):
        self.api.followers(username)

Mark Smith's avatar
Mark Smith committed
216
    @tape.use_cassette('testverifycredentials.json')
217
    def testverifycredentials(self):
218
        self.assertNotEqual(self.api.verify_credentials(), False)
219

220
221
222
223
224
225
226
227
        # make sure that `me.status.entities` is not an empty dict
        me = self.api.verify_credentials(include_entities=True)
        self.assertTrue(me.status.entities)

        # `status` shouldn't be included
        me = self.api.verify_credentials(skip_status=True)
        self.assertFalse(hasattr(me, 'status'))

Mark Smith's avatar
Mark Smith committed
228
    @tape.use_cassette('testratelimitstatus.json')
229
230
231
    def testratelimitstatus(self):
        self.api.rate_limit_status()

232
    """ TODO(josh): Remove once this deprecated API is gone.
233
234
235
    def testsetdeliverydevice(self):
        self.api.set_delivery_device('im')
        self.api.set_delivery_device('none')
236
    """
237

Mark Smith's avatar
Mark Smith committed
238
    @tape.use_cassette('testupdateprofilecolors.json')
239
240
    def testupdateprofilecolors(self):
        original = self.api.me()
Josh Roesslein's avatar
Josh Roesslein committed
241
        updated = self.api.update_profile_colors('000', '000', '000', '000', '000')
242
243
244
245
246
247
248
249
250
251

        # restore colors
        self.api.update_profile_colors(
            original.profile_background_color,
            original.profile_text_color,
            original.profile_link_color,
            original.profile_sidebar_fill_color,
            original.profile_sidebar_border_color
        )

Joshua Roesslein's avatar
Joshua Roesslein committed
252
253
254
255
256
        self.assertEqual(updated.profile_background_color, '000000')
        self.assertEqual(updated.profile_text_color, '000000')
        self.assertEqual(updated.profile_link_color, '000000')
        self.assertEqual(updated.profile_sidebar_fill_color, '000000')
        self.assertEqual(updated.profile_sidebar_border_color, '000000')
257

258
    """
259
260
261
262
263
    def testupateprofileimage(self):
        self.api.update_profile_image('examples/profile.png')

    def testupdateprofilebg(self):
        self.api.update_profile_background_image('examples/bg.png')
264
    """
265

Mark Smith's avatar
Mark Smith committed
266
    @tape.use_cassette('testupdateprofilebannerimage.yaml', serializer='yaml')
267
268
269
    def testupdateprofilebannerimage(self):
        self.api.update_profile_banner('examples/banner.png')

Mark Smith's avatar
Mark Smith committed
270
    @tape.use_cassette('testupdateprofile.json')
271
272
273
274
275
276
277
278
279
    def testupdateprofile(self):
        original = self.api.me()
        profile = {
            'name': 'Tweepy test 123',
            'location': 'pytopia',
            'description': 'just testing things out'
        }
        updated = self.api.update_profile(**profile)
        self.api.update_profile(
280
            name = original.name, url = original.url,
281
282
283
284
            location = original.location, description = original.description
        )

        for k,v in profile.items():
Josh Roesslein's avatar
Josh Roesslein committed
285
            if k == 'email': continue
286
287
            self.assertEqual(getattr(updated, k), v)

Mark Smith's avatar
Mark Smith committed
288
    @tape.use_cassette('testfavorites.json')
289
290
291
    def testfavorites(self):
        self.api.favorites()

Mark Smith's avatar
Mark Smith committed
292
    @tape.use_cassette('testcreatedestroyfavorite.json')
293
    def testcreatedestroyfavorite(self):
Josh Roesslein's avatar
Josh Roesslein committed
294
295
        self.api.create_favorite(4901062372)
        self.api.destroy_favorite(4901062372)
296

Mark Smith's avatar
Mark Smith committed
297
    @tape.use_cassette('testcreatedestroyblock.json')
298
299
300
    def testcreatedestroyblock(self):
        self.api.create_block('twitter')
        self.api.destroy_block('twitter')
Josh Roesslein's avatar
Josh Roesslein committed
301
        self.api.create_friendship('twitter') # restore
302

Mark Smith's avatar
Mark Smith committed
303
    @tape.use_cassette('testblocks.json')
304
305
306
    def testblocks(self):
        self.api.blocks()

Mark Smith's avatar
Mark Smith committed
307
    @tape.use_cassette('testblocksids.json')
308
309
310
    def testblocksids(self):
        self.api.blocks_ids()

Joshua Roesslein's avatar
Joshua Roesslein committed
311
312
313
314
315
316
317
318
319
320
    # TODO: Rewrite test to be less brittle. It fails way too often.
    # def testcreateupdatedestroylist(self):
    #     params = {
    #         'owner_screen_name': username,
    #         'slug': 'tweeps'
    #     }
    #     l = self.api.create_list(name=params['slug'], **params)
    #     l = self.api.update_list(list_id=l.id, description='updated!')
    #     self.assertEqual(l.description, 'updated!')
    #     self.api.destroy_list(list_id=l.id)
Josh Roesslein's avatar
Josh Roesslein committed
321

Mark Smith's avatar
Mark Smith committed
322
    @tape.use_cassette('testlistsall.json')
Joshua Roesslein's avatar
Joshua Roesslein committed
323
324
    def testlistsall(self):
        self.api.lists_all()
Josh Roesslein's avatar
Josh Roesslein committed
325

Mark Smith's avatar
Mark Smith committed
326
    @tape.use_cassette('testlistsmemberships.json')
Josh Roesslein's avatar
Josh Roesslein committed
327
328
329
    def testlistsmemberships(self):
        self.api.lists_memberships()

Mark Smith's avatar
Mark Smith committed
330
    @tape.use_cassette('testlistssubscriptions.json')
Josh Roesslein's avatar
Josh Roesslein committed
331
332
333
    def testlistssubscriptions(self):
        self.api.lists_subscriptions()

Mark Smith's avatar
Mark Smith committed
334
    @tape.use_cassette('testlisttimeline.json')
Josh Roesslein's avatar
Josh Roesslein committed
335
    def testlisttimeline(self):
Joshua's avatar
Joshua committed
336
        self.api.list_timeline('applepie', 'stars')
Josh Roesslein's avatar
Josh Roesslein committed
337

Mark Smith's avatar
Mark Smith committed
338
    @tape.use_cassette('testgetlist.json')
Josh Roesslein's avatar
Josh Roesslein committed
339
    def testgetlist(self):
Joshua Roesslein's avatar
Joshua Roesslein committed
340
        self.api.get_list(owner_screen_name='applepie', slug='stars')
Josh Roesslein's avatar
Josh Roesslein committed
341

Mark Smith's avatar
Mark Smith committed
342
    @tape.use_cassette('testaddremovelistmember.json')
Josh Roesslein's avatar
Josh Roesslein committed
343
    def testaddremovelistmember(self):
344
345
346
347
348
349
350
351
352
353
        params = {
            'slug': 'test',
            'owner_screen_name': username,
            'screen_name': 'twitter'
        }

        def assert_list(l):
            self.assertEqual(l.name, params['slug'])

        assert_list(self.api.add_list_member(**params))
354
        sleep(3)
355
        assert_list(self.api.remove_list_member(**params))
Josh Roesslein's avatar
Josh Roesslein committed
356

Mark Smith's avatar
Mark Smith committed
357
    @tape.use_cassette('testaddremovelistmembers.json')
358
359
360
361
    def testaddremovelistmembers(self):
        params = {
            'slug': 'test',
            'owner_screen_name': username,
kk6's avatar
kk6 committed
362
            'screen_name': ['twitterapi', 'twittermobile']
363
364
365
366
367
368
369
370
        }

        def assert_list(l):
            self.assertEqual(l.name, params['slug'])

        assert_list(self.api.add_list_members(**params))
        assert_list(self.api.remove_list_members(**params))

Mark Smith's avatar
Mark Smith committed
371
    @tape.use_cassette('testlistmembers.json')
Josh Roesslein's avatar
Josh Roesslein committed
372
    def testlistmembers(self):
Joshua's avatar
Joshua committed
373
        self.api.list_members('applepie', 'stars')
Josh Roesslein's avatar
Josh Roesslein committed
374

Mark Smith's avatar
Mark Smith committed
375
    @tape.use_cassette('testshowlistmember.json')
Joshua Roesslein's avatar
Joshua Roesslein committed
376
377
    def testshowlistmember(self):
        self.assertTrue(self.api.show_list_member(owner_screen_name='applepie', slug='stars', screen_name='NathanFillion'))
Josh Roesslein's avatar
Josh Roesslein committed
378

Mark Smith's avatar
Mark Smith committed
379
    @tape.use_cassette('testsubscribeunsubscribelist.json')
Josh Roesslein's avatar
Josh Roesslein committed
380
    def testsubscribeunsubscribelist(self):
Joshua Roesslein's avatar
Joshua Roesslein committed
381
382
383
384
385
386
        params = {
            'owner_screen_name': 'applepie',
            'slug': 'stars'
        }
        self.api.subscribe_list(**params)
        self.api.unsubscribe_list(**params)
Josh Roesslein's avatar
Josh Roesslein committed
387

Mark Smith's avatar
Mark Smith committed
388
    @tape.use_cassette('testlistsubscribers.json')
Josh Roesslein's avatar
Josh Roesslein committed
389
    def testlistsubscribers(self):
Joshua's avatar
Joshua committed
390
        self.api.list_subscribers('applepie', 'stars')
Josh Roesslein's avatar
Josh Roesslein committed
391

Mark Smith's avatar
Mark Smith committed
392
    @tape.use_cassette('testshowlistsubscriber.json')
Joshua Roesslein's avatar
Joshua Roesslein committed
393
    def testshowlistsubscriber(self):
Joshua Roesslein's avatar
Joshua Roesslein committed
394
        self.assertTrue(self.api.show_list_subscriber('tweepytest', 'test', 'applepie'))
Josh Roesslein's avatar
Josh Roesslein committed
395

Mark Smith's avatar
Mark Smith committed
396
    @tape.use_cassette('testsavedsearches.json')
397
398
399
    def testsavedsearches(self):
        s = self.api.create_saved_search('test')
        self.api.saved_searches()
Josh Roesslein's avatar
Josh Roesslein committed
400
        self.assertEqual(self.api.get_saved_search(s.id).query, 'test')
401
402
        self.api.destroy_saved_search(s.id)

Mark Smith's avatar
Mark Smith committed
403
    @tape.use_cassette('testsearch.json')
404
    def testsearch(self):
Josh Roesslein's avatar
Josh Roesslein committed
405
        self.api.search('tweepy')
406

Mark Smith's avatar
Mark Smith committed
407
    @tape.use_cassette('testgeoapis.json')
408
    def testgeoapis(self):
409
410
411
412
        def place_name_in_list(place_name, place_list):
            """Return True if a given place_name is in place_list."""
            return any([x.full_name.lower() == place_name.lower() for x in place_list])

413
414
        twitter_hq = self.api.geo_similar_places(lat='37.7821120598956',
                                                 long='-122.400612831116',
Mark Smith's avatar
Mark Smith committed
415
                                                 name='South of Market Child Care')
416
        # Assumes that twitter_hq is first Place returned...
417
        self.assertEqual(twitter_hq[0].id, '1d019624e6b4dcff')
418
        # Test various API functions using Austin, TX, USA
Mark Smith's avatar
Mark Smith committed
419
        self.assertEqual(self.api.geo_id(id='1ffd3558f2e98349').full_name, 'Dogpatch, San Francisco')
420
421
        self.assertTrue(place_name_in_list('Austin, TX',
            self.api.reverse_geocode(lat=30.267370168467806, long= -97.74261474609375))) # Austin, TX, USA
422

Mark Smith's avatar
Mark Smith committed
423
    @tape.use_cassette('testsupportedlanguages.json')
424
425
426
427
428
429
430
431
432
    def testsupportedlanguages(self):
        languages = self.api.supported_languages()
        expected_dict = {
            "name": "English",
            "code": "en",
            "status": "production"
        }
        self.assertTrue(expected_dict in languages)

Mark Smith's avatar
Mark Smith committed
433
    @tape.use_cassette('testcachedresult.json')
Joshua Roesslein's avatar
Joshua Roesslein committed
434
435
436
437
438
439
440
    def testcachedresult(self):
        self.api.cache = MemoryCache()
        self.api.home_timeline()
        self.assertFalse(self.api.cached_result)
        self.api.home_timeline()
        self.assertTrue(self.api.cached_result)

441

Mark Smith's avatar
Mark Smith committed
442
class TweepyCacheTests(unittest.TestCase):
443
    timeout = 0.5
Josh Roesslein's avatar
Josh Roesslein committed
444
445
446
447
448
449
450
451
452
    memcache_servers = ['127.0.0.1:11211']  # must be running for test to pass

    def _run_tests(self, do_cleanup=True):
        # test store and get
        self.cache.store('testkey', 'testvalue')
        self.assertEqual(self.cache.get('testkey'), 'testvalue',
            'Stored value does not match retrieved value')

        # test timeout
453
        sleep(self.timeout, True)
Josh Roesslein's avatar
Josh Roesslein committed
454
455
456
457
458
459
        self.assertEqual(self.cache.get('testkey'), None,
            'Cache entry should have expired')

        # test cleanup
        if do_cleanup:
            self.cache.store('testkey', 'testvalue')
460
            sleep(self.timeout, True)
Josh Roesslein's avatar
Josh Roesslein committed
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
            self.cache.cleanup()
            self.assertEqual(self.cache.count(), 0, 'Cache cleanup failed')

        # test count
        for i in range(0, 20):
            self.cache.store('testkey%i' % i, 'testvalue')
        self.assertEqual(self.cache.count(), 20, 'Count is wrong')

        # test flush
        self.cache.flush()
        self.assertEqual(self.cache.count(), 0, 'Cache failed to flush')

    def testmemorycache(self):
        self.cache = MemoryCache(timeout=self.timeout)
        self._run_tests()

    def testfilecache(self):
        os.mkdir('cache_test_dir')
Mark Smith's avatar
Mark Smith committed
479
480
481
482
483
484
485
486
        try:
            self.cache = FileCache('cache_test_dir', self.timeout)
            self._run_tests()
            self.cache.flush()
        finally:
            if os.path.exists('cache_test_dir'):
                shutil.rmtree('cache_test_dir')

487
488
489
490
491
old_sleep = time.sleep

def sleep(t, override=False):
    if not use_replay or override:
        old_sleep(t)
492

493
if __name__ == '__main__':
Josh Roesslein's avatar
Josh Roesslein committed
494
    unittest.main()