<s> """<STR_LIT>""" <EOL> from binascii import b2a_hex <EOL> import mock <EOL> from Queue import Queue <EOL> import time <EOL> from nose . tools import ok_ , eq_ , raises <EOL> from apns import Frame , Payload , PayloadTooLargeError <EOL> from apns_proxy_server . worker import SendWorkerThread <EOL> dummy_setting = { <EOL> '<STR_LIT>' : Queue ( ) , <EOL> '<STR_LIT:name>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : True , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } <EOL> dummy_token = '<STR_LIT>' <EOL> def test_create_instance ( ) : <EOL> ok_ ( SendWorkerThread ( ** dummy_setting ) ) <EOL> def test_create_payload ( ) : <EOL> worker = SendWorkerThread ( ** dummy_setting ) <EOL> payload = worker . create_payload ( <EOL> alert = '<STR_LIT>' , <EOL> content_available = True <EOL> ) <EOL> ok_ ( isinstance ( payload , Payload ) ) <EOL> def test_create_payload_with_dict_alert ( ) : <EOL> worker = SendWorkerThread ( ** dummy_setting ) <EOL> payload = worker . create_payload ( <EOL> alert = { <EOL> '<STR_LIT:body>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } , <EOL> content_available = True <EOL> ) <EOL> ok_ ( isinstance ( payload , Payload ) ) <EOL> @ raises ( PayloadTooLargeError ) <EOL> def test_create_payload_with_too_large_content ( ) : <EOL> worker = SendWorkerThread ( ** dummy_setting ) <EOL> worker . create_payload ( <EOL> alert = '<STR_LIT>' , <EOL> sound = '<STR_LIT:default>' , <EOL> badge = <NUM_LIT> , <EOL> custom = { <EOL> '<STR_LIT:foo>' : '<STR_LIT>' , <EOL> '<STR_LIT:bar>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } , <EOL> content_available = True <EOL> ) <EOL> has_called_check_method = False <EOL> def test_add_frame_item ( ) : <EOL> global has_called_check_method <EOL> has_called_check_method = False <EOL> def check_add_item_args ( token , payload , identifier , expiry , priority ) : <EOL> default_expiry = int ( time . time ( ) ) + ( <NUM_LIT> * <NUM_LIT> ) <EOL> eq_ ( token , dummy_token ) <EOL> eq_ ( payload . alert , '<STR_LIT>' ) <EOL> eq_ ( payload . badge , None ) <EOL> eq_ ( payload . sound , None ) <EOL> eq_ ( payload . custom , { } ) <EOL> eq_ ( payload . content_available , False ) <EOL> eq_ ( identifier , <NUM_LIT> ) <EOL> eq_ ( priority , <NUM_LIT:10> , '<STR_LIT>' ) <EOL> eq_ ( expiry , default_expiry , '<STR_LIT>' ) <EOL> global has_called_check_method <EOL> has_called_check_method = True <EOL> worker = SendWorkerThread ( ** dummy_setting ) <EOL> worker . count = <NUM_LIT> <EOL> frame = Frame ( ) <EOL> frame . add_item = check_add_item_args <EOL> worker . add_frame_item ( frame , '<STR_LIT>' , dummy_token , { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> ok_ ( has_called_check_method ) <EOL> def test_add_frame_item_with_full_args ( ) : <EOL> global has_called_check_method <EOL> has_called_check_method = False <EOL> def check_add_item_args ( token , payload , identifier , expiry , priority ) : <EOL> eq_ ( token , dummy_token ) <EOL> eq_ ( payload . alert , '<STR_LIT>' ) <EOL> eq_ ( payload . badge , <NUM_LIT:50> ) <EOL> eq_ ( payload . sound , '<STR_LIT>' ) <EOL> eq_ ( payload . custom , { '<STR_LIT:foo>' : '<STR_LIT:bar>' } ) <EOL> eq_ ( payload . content_available , True ) <EOL> eq_ ( identifier , <NUM_LIT:1> ) <EOL> eq_ ( priority , <NUM_LIT:5> ) <EOL> eq_ ( expiry , <NUM_LIT:0> ) <EOL> global has_called_check_method <EOL> has_called_check_method = True <EOL> worker = SendWorkerThread ( ** dummy_setting ) <EOL> worker . count = <NUM_LIT:1> <EOL> frame = Frame ( ) <EOL> frame . add_item = check_add_item_args <EOL> worker . add_frame_item ( frame , '<STR_LIT>' , dummy_token , { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : <NUM_LIT:50> , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : { '<STR_LIT:foo>' : '<STR_LIT:bar>' } , <EOL> '<STR_LIT>' : True , <EOL> } , expiry = <NUM_LIT:0> , priority = <NUM_LIT:5> ) <EOL> ok_ ( has_called_check_method ) <EOL> def test_frame_content ( ) : <EOL> worker = SendWorkerThread ( ** dummy_setting ) <EOL> frame = Frame ( ) <EOL> ok_ ( len ( frame . frame_data ) == <NUM_LIT:0> ) <EOL> worker . add_frame_item ( frame , '<STR_LIT>' , dummy_token , { '<STR_LIT>' : '<STR_LIT:hello>' } , expiry = None , priority = <NUM_LIT:10> ) <EOL> hex_frame = b2a_hex ( str ( frame . frame_data ) ) <EOL> eq_ ( '<STR_LIT>' , hex_frame [ <NUM_LIT:0> : <NUM_LIT:2> ] ) <EOL> eq_ ( '<STR_LIT>' , hex_frame [ <NUM_LIT:2> : <NUM_LIT:10> ] ) <EOL> eq_ ( '<STR_LIT>' , hex_frame [ <NUM_LIT:10> : <NUM_LIT:12> ] ) <EOL> eq_ ( '<STR_LIT>' , hex_frame [ <NUM_LIT:12> : <NUM_LIT:16> ] ) <EOL> eq_ ( dummy_token , hex_frame [ <NUM_LIT:16> : <NUM_LIT> ] ) <EOL> def test_increment_count ( ) : <EOL> q = Queue ( ) <EOL> worker = SendWorkerThread ( ** { <EOL> '<STR_LIT>' : q , <EOL> '<STR_LIT:name>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : True , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } ) <EOL> worker . count = <NUM_LIT> <EOL> worker . send = mock . Mock ( ) <EOL> worker . check_error = mock . Mock ( ) <EOL> q . put ( { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : dummy_token , <EOL> '<STR_LIT>' : { '<STR_LIT>' : '<STR_LIT>' } <EOL> } ) <EOL> q . put ( { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : dummy_token , <EOL> '<STR_LIT>' : { '<STR_LIT>' : '<STR_LIT>' } <EOL> } ) <EOL> worker . main ( ) <EOL> eq_ ( <NUM_LIT> , worker . count , '<STR_LIT>' ) <EOL> def test_send ( ) : <EOL> """<STR_LIT>""" <EOL> q = Queue ( ) <EOL> worker = SendWorkerThread ( ** { <EOL> '<STR_LIT>' : q , <EOL> '<STR_LIT:name>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : True , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } ) <EOL> worker . send = mock . Mock ( ) <EOL> worker . check_error = mock . Mock ( ) <EOL> q . put ( { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : dummy_token , <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT:default>' <EOL> } <EOL> } ) <EOL> worker . main ( ) <EOL> eq_ ( True , worker . send . called ) <EOL> eq_ ( True , worker . check_error . called ) <EOL> def test_send_with_test ( ) : <EOL> """<STR_LIT>""" <EOL> q = Queue ( ) <EOL> worker = SendWorkerThread ( ** { <EOL> '<STR_LIT>' : q , <EOL> '<STR_LIT:name>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : True , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } ) <EOL> worker . send = mock . Mock ( ) <EOL> worker . check_error = mock . Mock ( ) <EOL> q . put ( { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : dummy_token , <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT:default>' <EOL> } , <EOL> '<STR_LIT:test>' : True <EOL> } ) <EOL> worker . main ( ) <EOL> eq_ ( False , worker . send . called ) <EOL> eq_ ( False , worker . check_error . called ) </s>
