sg101: accounts/tests/test_tasks.py annotate

Add unit test for accounts task.

rev	line source
bgneal@1209	1 """
bgneal@1209	2 Tests for the account application tasks.
bgneal@1209	3
bgneal@1209	4 """
bgneal@1209	5 from django.test import TestCase
bgneal@1209	6 from mock import call, patch, Mock
bgneal@1209	7
bgneal@1209	8 import accounts.stats
bgneal@1209	9 from accounts.tasks import user_stats_task
bgneal@1209	10
bgneal@1209	11
bgneal@1209	12 class UserStatsTaskTestCase(TestCase):
bgneal@1209	13 fixtures = ['accounts.json']
bgneal@1209	14
bgneal@1209	15 @patch('accounts.stats.get_redis_connection')
bgneal@1209	16 def test_user_stats_task(self, connection_mock):
bgneal@1209	17 redis_mock = Mock()
bgneal@1209	18 redis_mock.incr.return_value = 1
bgneal@1209	19 connection_mock.return_value = redis_mock
bgneal@1209	20
bgneal@1209	21 task = user_stats_task.s(1).apply()
bgneal@1209	22
bgneal@1209	23 self.assertEquals(redis_mock.mock_calls, [
bgneal@1209	24 call.incr('accounts:user_count'),
bgneal@1209	25 call.set('accounts:user_count', 1),
bgneal@1209	26 call.pipeline(),
bgneal@1209	27 call.pipeline().lpush('accounts:new_users', u'Gremmie'),
bgneal@1209	28 call.pipeline().ltrim('accounts:new_users', 0, 9),
bgneal@1209	29 call.pipeline().execute(),
bgneal@1209	30 ])

Mercurial > public > sg101