<s> import subprocess <EOL> import struct <EOL> import os <EOL> import json <EOL> import uuid <EOL> import StringIO <EOL> import django <EOL> from django . test import TestCase <EOL> try : <EOL> from django . test . utils import override_settings <EOL> except ImportError : <EOL> from override_settings import override_settings <EOL> from django . conf import settings <EOL> from django . core . urlresolvers import reverse <EOL> from django . contrib . auth . models import User <EOL> from django . http import HttpResponseNotAllowed <EOL> from django . core import management <EOL> try : <EOL> from django . utils . timezone import now as dt_now <EOL> except ImportError : <EOL> import datetime <EOL> dt_now = datetime . datetime . now <EOL> from . models import APNService , Device , Notification , NotificationPayloadSizeExceeded <EOL> from . http import JSONResponse <EOL> from . utils import generate_cert_and_pkey <EOL> from . forms import APNServiceForm <EOL> from . settings import get_setting <EOL> TOKEN = '<STR_LIT>' <EOL> TEST_PEM = os . path . abspath ( os . path . join ( os . path . dirname ( __file__ ) , '<STR_LIT>' ) ) <EOL> SSL_SERVER_COMMAND = ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , TEST_PEM ) <EOL> class UseMockSSLServerMixin ( object ) : <EOL> @ classmethod <EOL> def setUpClass ( cls ) : <EOL> super ( UseMockSSLServerMixin , cls ) . setUpClass ( ) <EOL> cls . test_server_proc = subprocess . Popen ( SSL_SERVER_COMMAND , stdout = subprocess . PIPE ) <EOL> @ classmethod <EOL> def tearDownClass ( cls ) : <EOL> cls . test_server_proc . kill ( ) <EOL> super ( UseMockSSLServerMixin , cls ) . tearDownClass ( ) <EOL> class APNServiceTest ( UseMockSSLServerMixin , TestCase ) : <EOL> def setUp ( self ) : <EOL> cert , key = generate_cert_and_pkey ( ) <EOL> self . service = APNService . objects . create ( name = '<STR_LIT>' , hostname = '<STR_LIT:127.0.0.1>' , <EOL> certificate = cert , private_key = key ) <EOL> self . device = Device . objects . create ( token = TOKEN , service = self . service ) <EOL> self . notification = Notification . objects . create ( message = '<STR_LIT>' , service = self . service ) <EOL> def test_invalid_payload_size ( self ) : <EOL> n = Notification ( message = '<STR_LIT:.>' * <NUM_LIT> ) <EOL> self . assertRaises ( NotificationPayloadSizeExceeded , self . service . pack_message , n . payload , self . device ) <EOL> def test_payload_packed_correctly ( self ) : <EOL> fmt = self . service . fmt <EOL> payload = self . notification . payload <EOL> msg = self . service . pack_message ( payload , self . device ) <EOL> unpacked = struct . unpack ( fmt % len ( payload ) , msg ) <EOL> self . assertEqual ( unpacked [ - <NUM_LIT:1> ] , payload ) <EOL> def test_pack_message_with_invalid_device ( self ) : <EOL> self . assertRaises ( TypeError , self . service . pack_message , None ) <EOL> def test_can_connect_and_push_notification ( self ) : <EOL> self . assertIsNone ( self . notification . last_sent_at ) <EOL> self . assertIsNone ( self . device . last_notified_at ) <EOL> self . service . push_notification_to_devices ( self . notification , [ self . device ] ) <EOL> self . assertIsNotNone ( self . notification . last_sent_at ) <EOL> self . device = Device . objects . get ( pk = self . device . pk ) <EOL> self . assertIsNotNone ( self . device . last_notified_at ) <EOL> def test_create_with_passphrase ( self ) : <EOL> cert , key = generate_cert_and_pkey ( as_string = True , passphrase = '<STR_LIT>' ) <EOL> form = APNServiceForm ( { '<STR_LIT:name>' : '<STR_LIT:test>' , '<STR_LIT>' : '<STR_LIT:localhost>' , '<STR_LIT>' : cert , '<STR_LIT>' : key , '<STR_LIT>' : '<STR_LIT>' } ) <EOL> self . assertTrue ( form . is_valid ( ) ) <EOL> def test_create_with_invalid_passphrase ( self ) : <EOL> cert , key = generate_cert_and_pkey ( as_string = True , passphrase = '<STR_LIT>' ) <EOL> form = APNServiceForm ( { '<STR_LIT:name>' : '<STR_LIT:test>' , '<STR_LIT>' : '<STR_LIT:localhost>' , '<STR_LIT>' : cert , '<STR_LIT>' : key , '<STR_LIT>' : '<STR_LIT>' } ) <EOL> self . assertFalse ( form . is_valid ( ) ) <EOL> self . assertTrue ( '<STR_LIT>' in form . errors ) <EOL> def test_pushing_notification_in_chunks ( self ) : <EOL> devices = [ ] <EOL> for i in xrange ( <NUM_LIT:10> ) : <EOL> token = uuid . uuid1 ( ) . get_hex ( ) * <NUM_LIT:2> <EOL> device = Device . objects . create ( token = token , service = self . service ) <EOL> devices . append ( device ) <EOL> started_at = dt_now ( ) <EOL> self . service . push_notification_to_devices ( self . notification , devices , chunk_size = <NUM_LIT:2> ) <EOL> device_count = len ( devices ) <EOL> self . assertEquals ( device_count , <EOL> Device . objects . filter ( last_notified_at__gte = started_at ) . count ( ) ) <EOL> @ override_settings ( IOS_NOTIFICATIONS_AUTHENTICATION = '<STR_LIT>' ) <EOL> class APITest ( UseMockSSLServerMixin , TestCase ) : <EOL> urls = '<STR_LIT>' <EOL> def setUp ( self ) : <EOL> cert , key = generate_cert_and_pkey ( ) <EOL> self . service = APNService . objects . create ( name = '<STR_LIT>' , hostname = '<STR_LIT:127.0.0.1>' , <EOL> certificate = cert , private_key = key ) <EOL> self . device_token = TOKEN <EOL> self . user = User . objects . create ( username = '<STR_LIT>' , email = '<STR_LIT>' ) <EOL> self . device = Device . objects . create ( service = self . service , token = '<STR_LIT>' ) <EOL> def test_register_device_invalid_params ( self ) : <EOL> """<STR_LIT>""" <EOL> resp = self . client . post ( reverse ( '<STR_LIT>' ) ) <EOL> self . assertEqual ( resp . status_code , <NUM_LIT> ) <EOL> self . assertTrue ( isinstance ( resp , JSONResponse ) ) <EOL> content = json . loads ( resp . content ) <EOL> keys = content . keys ( ) <EOL> self . assertTrue ( '<STR_LIT>' in keys and '<STR_LIT>' in keys ) <EOL> def test_register_device ( self ) : <EOL> """<STR_LIT>""" <EOL> resp = self . client . post ( reverse ( '<STR_LIT>' ) , <EOL> { '<STR_LIT>' : self . device_token , <EOL> '<STR_LIT>' : self . service . id } ) <EOL> self . assertEqual ( resp . status_code , <NUM_LIT> ) <EOL> self . assertTrue ( isinstance ( resp , JSONResponse ) ) <EOL> content = resp . content <EOL> device_json = json . loads ( content ) <EOL> self . assertEqual ( device_json . get ( '<STR_LIT>' ) , '<STR_LIT>' ) <EOL> def test_disallowed_method ( self ) : <EOL> resp = self . client . delete ( reverse ( '<STR_LIT>' ) ) <EOL> self . assertEqual ( resp . status_code , <NUM_LIT> ) <EOL> self . assertTrue ( isinstance ( resp , HttpResponseNotAllowed ) ) <EOL> def test_update_device ( self ) : <EOL> kwargs = { '<STR_LIT>' : self . device . token , '<STR_LIT>' : self . device . service . id } <EOL> url = reverse ( '<STR_LIT>' , kwargs = kwargs ) <EOL> resp = self . client . put ( url , '<STR_LIT>' % self . user . id , <EOL> content_type = '<STR_LIT>' ) <EOL> self . assertEqual ( resp . status_code , <NUM_LIT:200> ) <EOL> self . assertTrue ( isinstance ( resp , JSONResponse ) ) <EOL> device_json = json . loads ( resp . content ) <EOL> self . assertEqual ( device_json . get ( '<STR_LIT>' ) , self . device . id ) <EOL> self . assertTrue ( self . user in self . device . users . all ( ) ) <EOL> def test_get_device_details ( self ) : <EOL> kwargs = { '<STR_LIT>' : self . device . token , '<STR_LIT>' : self . device . service . id } <EOL> url = reverse ( '<STR_LIT>' , kwargs = kwargs ) <EOL> resp = self . client . get ( url ) <EOL> self . assertEqual ( resp . status_code , <NUM_LIT:200> ) <EOL> content = resp . content <EOL> device_json = json . loads ( content ) <EOL> self . assertEqual ( device_json . get ( '<STR_LIT>' ) , '<STR_LIT>' ) <EOL> class AuthenticationDecoratorTestAuthBasic ( UseMockSSLServerMixin , TestCase ) : <EOL> urls = '<STR_LIT>' <EOL> def setUp ( self ) : <EOL> cert , key = generate_cert_and_pkey ( ) <EOL> self . service = APNService . objects . create ( name = '<STR_LIT>' , hostname = '<STR_LIT:127.0.0.1>' , <EOL> certificate = cert , private_key = key ) <EOL> self . device_token = TOKEN <EOL> self . user_password = '<STR_LIT>' <EOL> self . user = User . objects . create ( username = '<STR_LIT>' , email = '<STR_LIT>' ) <EOL> self . user . set_password ( self . user_password ) <EOL> self . user . is_staff = True <EOL> self . user . save ( ) <EOL> self . device = Device . objects . create ( service = self . service , token = '<STR_LIT>' ) <EOL> @ override_settings ( IOS_NOTIFICATIONS_AUTHENTICATION = '<STR_LIT>' ) <EOL> def test_basic_authorization_request ( self ) : <EOL> kwargs = { '<STR_LIT>' : self . device . token , '<STR_LIT>' : self . device . service . id } <EOL> url = reverse ( '<STR_LIT>' , kwargs = kwargs ) <EOL> user_pass = '<STR_LIT>' % ( self . user . username , self . user_password ) <EOL> auth_header = '<STR_LIT>' % user_pass . encode ( '<STR_LIT>' ) <EOL> resp = self . client . get ( url , { } , HTTP_AUTHORIZATION = auth_header ) <EOL> self . assertEquals ( resp . status_code , <NUM_LIT:200> ) <EOL> @ override_settings ( IOS_NOTIFICATIONS_AUTHENTICATION = '<STR_LIT>' ) <EOL> def test_basic_authorization_request_invalid_credentials ( self ) : <EOL> user_pass = '<STR_LIT>' % ( self . user . username , '<STR_LIT>' ) <EOL> auth_header = '<STR_LIT>' % user_pass . encode ( '<STR_LIT>' ) <EOL> url = reverse ( '<STR_LIT>' ) <EOL> resp = self . client . get ( url , HTTP_AUTHORIZATION = auth_header ) <EOL> self . assertEquals ( resp . status_code , <NUM_LIT> ) <EOL> self . assertTrue ( '<STR_LIT>' in resp . content ) <EOL> @ override_settings ( IOS_NOTIFICATIONS_AUTHENTICATION = '<STR_LIT>' ) <EOL> def test_basic_authorization_missing_header ( self ) : <EOL> url = reverse ( '<STR_LIT>' ) <EOL> resp = self . client . get ( url ) <EOL> self . assertEquals ( resp . status_code , <NUM_LIT> ) <EOL> self . assertTrue ( '<STR_LIT>' in resp . content ) <EOL> @ override_settings ( IOS_NOTIFICATIONS_AUTHENTICATION = '<STR_LIT>' ) <EOL> def test_invalid_authentication_type ( self ) : <EOL> from ios_notifications . decorators import InvalidAuthenticationType <EOL> url = reverse ( '<STR_LIT>' ) <EOL> self . assertRaises ( InvalidAuthenticationType , self . client . get , url ) <EOL> @ override_settings ( IOS_NOTIFICATIONS_AUTHENTICATION = None ) <EOL> def test_no_authentication_type ( self ) : <EOL> from ios_notifications . decorators import InvalidAuthenticationType <EOL> url = reverse ( '<STR_LIT>' ) <EOL> self . assertRaises ( InvalidAuthenticationType , self . client . get , url ) <EOL> @ override_settings ( IOS_NOTIFICATIONS_AUTHENTICATION = '<STR_LIT>' ) <EOL> def test_basic_authorization_is_staff ( self ) : <EOL> kwargs = { '<STR_LIT>' : self . device . token , '<STR_LIT>' : self . device . service . id } <EOL> url = reverse ( '<STR_LIT>' , kwargs = kwargs ) <EOL> user_pass = '<STR_LIT>' % ( self . user . username , self . user_password ) <EOL> auth_header = '<STR_LIT>' % user_pass . encode ( '<STR_LIT>' ) <EOL> self . user . is_staff = True <EOL> resp = self . client . get ( url , HTTP_AUTHORIZATION = auth_header ) <EOL> self . assertEquals ( resp . status_code , <NUM_LIT:200> ) <EOL> @ override_settings ( IOS_NOTIFICATIONS_AUTHENTICATION = '<STR_LIT>' ) <EOL> def test_basic_authorization_is_staff_with_non_staff_user ( self ) : <EOL> kwargs = { '<STR_LIT>' : self . device . token , '<STR_LIT>' : self . device . service . id } <EOL> url = reverse ( '<STR_LIT>' , kwargs = kwargs ) <EOL> user_pass = '<STR_LIT>' % ( self . user . username , self . user_password ) <EOL> auth_header = '<STR_LIT>' % user_pass . encode ( '<STR_LIT>' ) <EOL> self . user . is_staff = False <EOL> self . user . save ( ) <EOL> resp = self . client . get ( url , HTTP_AUTHORIZATION = auth_header ) <EOL> self . assertEquals ( resp . status_code , <NUM_LIT> ) <EOL> self . assertTrue ( '<STR_LIT>' in resp . content ) <EOL> class NotificationTest ( UseMockSSLServerMixin , TestCase ) : <EOL> def setUp ( self ) : <EOL> cert , key = generate_cert_and_pkey ( ) <EOL> self . service = APNService . objects . create ( name = '<STR_LIT>' , hostname = '<STR_LIT:127.0.0.1>' , <EOL> private_key = key , certificate = cert ) <EOL> self . service . PORT = <NUM_LIT> <EOL> self . custom_payload = json . dumps ( { "<STR_LIT:.>" * <NUM_LIT:10> : "<STR_LIT:.>" * <NUM_LIT:50> } ) <EOL> self . notification = Notification . objects . create ( service = self . service , message = '<STR_LIT>' , custom_payload = self . custom_payload ) <EOL> def test_valid_length ( self ) : <EOL> self . notification . message = '<STR_LIT>' <EOL> self . assertTrue ( self . notification . is_valid_length ( ) ) <EOL> def test_invalid_length ( self ) : <EOL> self . notification . message = '<STR_LIT:.>' * <NUM_LIT> <EOL> self . assertFalse ( self . notification . is_valid_length ( ) ) <EOL> def test_invalid_length_with_custom_payload ( self ) : <EOL> self . notification . message = '<STR_LIT:.>' * <NUM_LIT:100> <EOL> self . notification . custom_payload = '<STR_LIT>' % ( "<STR_LIT:.>" * <NUM_LIT:20> , "<STR_LIT:.>" * <NUM_LIT> ) <EOL> self . assertFalse ( self . notification . is_valid_length ( ) ) <EOL> def test_extra_property_with_custom_payload ( self ) : <EOL> custom_payload = { "<STR_LIT:.>" * <NUM_LIT:10> : "<STR_LIT:.>" * <NUM_LIT:50> , "<STR_LIT>" : { "<STR_LIT:+>" * <NUM_LIT:10> : "<STR_LIT:+>" * <NUM_LIT:50> } } <EOL> self . notification . extra = custom_payload <EOL> self . assertEqual ( self . notification . custom_payload , json . dumps ( custom_payload ) ) <EOL> self . assertEqual ( self . notification . extra , custom_payload ) <EOL> self . assertTrue ( self . notification . is_valid_length ( ) ) <EOL> def test_loc_data_payload ( self ) : <EOL> self . notification . set_loc_data ( '<STR_LIT>' , [ <NUM_LIT:1> , '<STR_LIT>' , <NUM_LIT> , '<STR_LIT>' ] ) <EOL> self . notification . message = '<STR_LIT>' <EOL> loc_data = { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : [ '<STR_LIT:1>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] } <EOL> self . assertEqual ( self . notification . loc_data , loc_data ) <EOL> self . assertTrue ( self . notification . is_valid_length ( ) ) <EOL> p = self . notification . payload <EOL> self . assertEqual ( json . loads ( p ) [ '<STR_LIT>' ] [ '<STR_LIT>' ] , loc_data ) <EOL> def test_extra_property_not_dict ( self ) : <EOL> with self . assertRaises ( TypeError ) : <EOL> self . notification . extra = <NUM_LIT> <EOL> def test_extra_property_none ( self ) : <EOL> self . notification . extra = None <EOL> self . assertEqual ( self . notification . extra , None ) <EOL> self . assertEqual ( self . notification . custom_payload , '<STR_LIT>' ) <EOL> self . assertTrue ( self . notification . is_valid_length ( ) ) <EOL> def test_push_to_all_devices_persist_existing ( self ) : <EOL> self . assertIsNone ( self . notification . last_sent_at ) <EOL> self . notification . persist = False <EOL> self . notification . push_to_all_devices ( ) <EOL> self . assertIsNotNone ( self . notification . last_sent_at ) <EOL> def test_push_to_all_devices_persist_new ( self ) : <EOL> notification = Notification ( service = self . service , message = '<STR_LIT>' ) <EOL> notification . persist = True <EOL> notification . push_to_all_devices ( ) <EOL> self . assertIsNotNone ( notification . last_sent_at ) <EOL> self . assertIsNotNone ( notification . pk ) <EOL> def test_push_to_all_devices_no_persist ( self ) : <EOL> notification = Notification ( service = self . service , message = '<STR_LIT>' ) <EOL> notification . persist = False <EOL> notification . push_to_all_devices ( ) <EOL> self . assertIsNone ( notification . last_sent_at ) <EOL> self . assertIsNone ( notification . pk ) <EOL> class ManagementCommandPushNotificationTest ( UseMockSSLServerMixin , TestCase ) : <EOL> def setUp ( self ) : <EOL> self . started_at = dt_now ( ) <EOL> cert , key = generate_cert_and_pkey ( ) <EOL> self . service = APNService . objects . create ( name = '<STR_LIT>' , hostname = '<STR_LIT:127.0.0.1>' , <EOL> private_key = key , certificate = cert ) <EOL> self . service . PORT = <NUM_LIT> <EOL> self . device = Device . objects . create ( token = TOKEN , service = self . service ) <EOL> def test_call_push_ios_notification_command_explicit_persist ( self ) : <EOL> msg = '<STR_LIT>' <EOL> management . call_command ( '<STR_LIT>' , ** { '<STR_LIT:message>' : msg , '<STR_LIT>' : self . service . id , '<STR_LIT>' : <NUM_LIT:0> , '<STR_LIT>' : True } ) <EOL> self . assertTrue ( Notification . objects . filter ( message = msg , last_sent_at__gt = self . started_at ) . exists ( ) ) <EOL> self . assertTrue ( self . device in Device . objects . filter ( last_notified_at__gt = self . started_at ) ) <EOL> def test_call_push_ios_notification_command_explicit_no_persist ( self ) : <EOL> msg = '<STR_LIT>' <EOL> management . call_command ( '<STR_LIT>' , ** { '<STR_LIT:message>' : msg , '<STR_LIT>' : self . service . id , '<STR_LIT>' : <NUM_LIT:0> , '<STR_LIT>' : False } ) <EOL> self . assertFalse ( Notification . objects . filter ( message = msg , last_sent_at__gt = self . started_at ) . exists ( ) ) <EOL> self . assertTrue ( self . device in Device . objects . filter ( last_notified_at__gt = self . started_at ) ) <EOL> @ override_settings ( IOS_NOTIFICATIONS_PERSIST_NOTIFICATIONS = True ) <EOL> def test_call_push_ios_notification_command_default_persist ( self ) : <EOL> msg = '<STR_LIT>' <EOL> management . call_command ( '<STR_LIT>' , ** { '<STR_LIT:message>' : msg , '<STR_LIT>' : self . service . id , '<STR_LIT>' : <NUM_LIT:0> } ) <EOL> self . assertTrue ( Notification . objects . filter ( message = msg , last_sent_at__gt = self . started_at ) . exists ( ) ) <EOL> self . assertTrue ( self . device in Device . objects . filter ( last_notified_at__gt = self . started_at ) ) <EOL> @ override_settings ( IOS_NOTIFICATIONS_PERSIST_NOTIFICATIONS = False ) <EOL> def test_call_push_ios_notification_command_default_no_persist ( self ) : <EOL> msg = '<STR_LIT>' <EOL> management . call_command ( '<STR_LIT>' , ** { '<STR_LIT:message>' : msg , '<STR_LIT>' : self . service . id , '<STR_LIT>' : <NUM_LIT:0> } ) <EOL> self . assertFalse ( Notification . objects . filter ( message = msg , last_sent_at__gt = self . started_at ) . exists ( ) ) <EOL> self . assertTrue ( self . device in Device . objects . filter ( last_notified_at__gt = self . started_at ) ) <EOL> @ override_settings ( ) <EOL> def test_call_push_ios_notification_command_default_persist_not_specified ( self ) : <EOL> try : <EOL> del settings . IOS_NOTIFICATIONS_PERSIST_NOTIFICATIONS <EOL> except AttributeError : <EOL> pass <EOL> msg = '<STR_LIT>' <EOL> management . call_command ( '<STR_LIT>' , ** { '<STR_LIT:message>' : msg , '<STR_LIT>' : self . service . id , '<STR_LIT>' : <NUM_LIT:0> } ) <EOL> self . assertTrue ( Notification . objects . filter ( message = msg , last_sent_at__gt = self . started_at ) . exists ( ) ) <EOL> self . assertTrue ( self . device in Device . objects . filter ( last_notified_at__gt = self . started_at ) ) <EOL> def test_either_message_or_extra_option_required ( self ) : <EOL> exception = SystemExit if django . VERSION < ( <NUM_LIT:1> , <NUM_LIT:5> ) else management . base . CommandError <EOL> with self . assertRaises ( exception ) : <EOL> management . call_command ( '<STR_LIT>' , service = self . service . pk , <EOL> verbosity = <NUM_LIT:0> , stderr = StringIO . StringIO ( ) ) <EOL> class ManagementCommandCallFeedbackService ( TestCase ) : <EOL> pass <EOL> class DefaultSettings ( TestCase ) : <EOL> def test_persist_notifications_setting ( self ) : <EOL> self . assertEqual ( True , get_setting ( '<STR_LIT>' ) ) <EOL> def test_authentication_setting ( self ) : <EOL> self . assertEqual ( None , get_setting ( '<STR_LIT>' ) ) <EOL> def test_auth_user_model ( self ) : <EOL> self . assertEqual ( '<STR_LIT>' , get_setting ( '<STR_LIT>' ) ) <EOL> def test_invalid_setting ( self ) : <EOL> setting_name = '<STR_LIT>' <EOL> with self . assertRaises ( KeyError ) : <EOL> get_setting ( setting_name ) </s>
