<s> import datetime <EOL> import mock <EOL> from django . contrib . auth import get_user_model <EOL> from django . test import TestCase <EOL> from django . test . utils import override_settings <EOL> from django . utils import timezone <EOL> from gcm . gcm import GCMNotRegisteredException <EOL> from pushy . models import ( <EOL> PushNotification , <EOL> Device , <EOL> get_filtered_devices_queryset <EOL> ) <EOL> from pushy . tasks import ( <EOL> check_pending_push_notifications , <EOL> send_push_notification_group , <EOL> send_single_push_notification , <EOL> create_push_notification_groups , <EOL> clean_sent_notifications <EOL> ) <EOL> class TasksTestCase ( TestCase ) : <EOL> def setUp ( self ) : <EOL> self . payload = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } <EOL> def test_get_filtered_devices_queryset_on_type ( self ) : <EOL> notification = PushNotification . objects . create ( <EOL> title = '<STR_LIT:test>' , <EOL> payload = self . payload , <EOL> active = PushNotification . PUSH_ACTIVE , <EOL> sent = PushNotification . PUSH_NOT_SENT , <EOL> filter_type = Device . DEVICE_TYPE_IOS <EOL> ) <EOL> for i in range ( <NUM_LIT:0> , <NUM_LIT:3> ) : <EOL> Device . objects . create ( <EOL> key = '<STR_LIT>' . format ( i ) , <EOL> type = Device . DEVICE_TYPE_IOS <EOL> ) <EOL> for i in range ( <NUM_LIT:0> , <NUM_LIT:10> ) : <EOL> Device . objects . create ( <EOL> key = '<STR_LIT>' . format ( i ) , <EOL> type = Device . DEVICE_TYPE_ANDROID <EOL> ) <EOL> devices = get_filtered_devices_queryset ( notification ) <EOL> self . assertEqual ( devices . count ( ) , <NUM_LIT:3> ) <EOL> def test_get_filtered_devices_queryset_on_user ( self ) : <EOL> for i in range ( <NUM_LIT:0> , <NUM_LIT:3> ) : <EOL> get_user_model ( ) . objects . create_user ( <EOL> username = '<STR_LIT>' % i , <EOL> email = '<STR_LIT>' % i , <EOL> password = i <EOL> ) <EOL> user1 = get_user_model ( ) . objects . get ( pk = <NUM_LIT:1> ) <EOL> user2 = get_user_model ( ) . objects . get ( pk = <NUM_LIT:2> ) <EOL> for i in range ( <NUM_LIT:0> , <NUM_LIT:5> ) : <EOL> Device . objects . create ( <EOL> key = '<STR_LIT>' . format ( i ) , <EOL> type = Device . DEVICE_TYPE_ANDROID , <EOL> user = user2 <EOL> ) <EOL> notification = PushNotification . objects . create ( <EOL> title = '<STR_LIT:test>' , <EOL> payload = self . payload , <EOL> active = PushNotification . PUSH_ACTIVE , <EOL> sent = PushNotification . PUSH_NOT_SENT , <EOL> filter_user = user2 . id <EOL> ) <EOL> devices = get_filtered_devices_queryset ( notification ) <EOL> self . assertEqual ( devices . count ( ) , <NUM_LIT:5> ) <EOL> notification = PushNotification . objects . create ( <EOL> title = '<STR_LIT:test>' , <EOL> payload = self . payload , <EOL> active = PushNotification . PUSH_ACTIVE , <EOL> sent = PushNotification . PUSH_NOT_SENT , <EOL> filter_user = user1 . id <EOL> ) <EOL> devices = get_filtered_devices_queryset ( notification ) <EOL> self . assertEqual ( devices . count ( ) , <NUM_LIT:0> ) <EOL> def test_pending_notifications ( self ) : <EOL> notification = PushNotification . objects . create ( <EOL> title = '<STR_LIT:test>' , <EOL> payload = self . payload , <EOL> active = PushNotification . PUSH_ACTIVE , <EOL> sent = PushNotification . PUSH_NOT_SENT <EOL> ) <EOL> mocked_task = mock . Mock ( ) <EOL> with mock . patch ( '<STR_LIT>' , new = mocked_task ) : <EOL> check_pending_push_notifications ( ) <EOL> notification = PushNotification . objects . get ( pk = notification . id ) <EOL> self . assertEqual ( notification . sent , PushNotification . PUSH_IN_PROGRESS ) <EOL> create_push_notification_groups ( notification . id ) <EOL> notification = PushNotification . objects . get ( pk = notification . id ) <EOL> self . assertEqual ( notification . sent , PushNotification . PUSH_SENT ) <EOL> def test_send_notification_groups ( self ) : <EOL> notification = PushNotification . objects . create ( <EOL> title = '<STR_LIT:test>' , <EOL> payload = self . payload , <EOL> active = PushNotification . PUSH_ACTIVE , <EOL> sent = PushNotification . PUSH_NOT_SENT <EOL> ) <EOL> self . assertFalse ( send_push_notification_group ( <NUM_LIT> , <NUM_LIT:0> , <NUM_LIT:1> ) ) <EOL> device = Device . objects . create ( key = '<STR_LIT>' , type = Device . DEVICE_TYPE_ANDROID ) <EOL> gcm = mock . Mock ( ) <EOL> gcm . return_value = <NUM_LIT> <EOL> with mock . patch ( '<STR_LIT>' , new = gcm ) : <EOL> send_push_notification_group ( notification . id , <NUM_LIT:0> , <NUM_LIT:1> ) <EOL> device = Device . objects . get ( pk = device . id ) <EOL> self . assertEqual ( device . key , '<STR_LIT>' ) <EOL> gcm = mock . Mock ( ) <EOL> gcm . side_effect = GCMNotRegisteredException <EOL> with mock . patch ( '<STR_LIT>' , new = gcm ) : <EOL> send_push_notification_group ( notification . id , <NUM_LIT:0> , <NUM_LIT:1> ) <EOL> self . assertRaises ( Device . DoesNotExist , Device . objects . get , pk = device . id ) <EOL> device = Device . objects . create ( key = '<STR_LIT>' , type = Device . DEVICE_TYPE_ANDROID ) <EOL> gcm = mock . Mock ( ) <EOL> gcm . return_value = False <EOL> with mock . patch ( '<STR_LIT>' , new = gcm ) : <EOL> send_push_notification_group ( notification . id , <NUM_LIT:0> , <NUM_LIT:1> ) <EOL> device = Device . objects . get ( pk = device . id ) <EOL> self . assertEqual ( device . key , '<STR_LIT>' ) <EOL> def test_create_push_notification_groups_non_existent_notification ( self ) : <EOL> result = create_push_notification_groups ( <NUM_LIT:1000> ) <EOL> self . assertFalse ( result ) <EOL> def test_non_existent_device ( self ) : <EOL> result = send_single_push_notification ( <NUM_LIT:1000> , { '<STR_LIT>' : '<STR_LIT:test>' } ) <EOL> self . assertFalse ( result ) <EOL> def test_delete_old_notifications_undefine_max_age ( self ) : <EOL> self . assertRaises ( ValueError , clean_sent_notifications ) <EOL> @ override_settings ( PUSHY_NOTIFICATION_MAX_AGE = datetime . timedelta ( days = <NUM_LIT> ) ) <EOL> def test_delete_old_notifications ( self ) : <EOL> for i in xrange ( <NUM_LIT:10> ) : <EOL> date_started = timezone . now ( ) - datetime . timedelta ( days = <NUM_LIT> ) <EOL> date_finished = date_started <EOL> notification = PushNotification ( ) <EOL> notification . title = '<STR_LIT>' . format ( i ) <EOL> notification . body = '<STR_LIT:{}>' <EOL> notification . sent = PushNotification . PUSH_SENT <EOL> notification . date_started = date_started <EOL> notification . date_finished = date_finished <EOL> notification . save ( ) <EOL> clean_sent_notifications ( ) <EOL> self . assertEquals ( PushNotification . objects . count ( ) , <NUM_LIT:0> ) <EOL> @ override_settings ( PUSHY_NOTIFICATION_MAX_AGE = datetime . timedelta ( days = <NUM_LIT> ) ) <EOL> def test_delete_old_notifications_with_remaining_onces ( self ) : <EOL> for i in xrange ( <NUM_LIT:10> ) : <EOL> date_started = timezone . now ( ) - datetime . timedelta ( days = <NUM_LIT> ) <EOL> date_finished = date_started <EOL> notification = PushNotification ( ) <EOL> notification . title = '<STR_LIT>' . format ( i ) <EOL> notification . body = '<STR_LIT:{}>' <EOL> notification . sent = PushNotification . PUSH_SENT <EOL> notification . date_started = date_started <EOL> notification . date_finished = date_finished <EOL> notification . save ( ) <EOL> for i in xrange ( <NUM_LIT:10> ) : <EOL> date_started = timezone . now ( ) - datetime . timedelta ( days = <NUM_LIT> ) <EOL> date_finished = date_started <EOL> notification = PushNotification ( ) <EOL> notification . title = '<STR_LIT>' . format ( i ) <EOL> notification . body = '<STR_LIT:{}>' <EOL> notification . sent = PushNotification . PUSH_SENT <EOL> notification . date_started = date_started <EOL> notification . date_finished = date_finished <EOL> notification . save ( ) <EOL> clean_sent_notifications ( ) <EOL> self . assertEquals ( PushNotification . objects . count ( ) , <NUM_LIT:10> ) </s>
