<s> import collections <EOL> import random <EOL> import unittest <EOL> from gearman . client import GearmanClient <EOL> from gearman . client_handler import GearmanClientCommandHandler <EOL> from gearman . constants import PRIORITY_NONE , PRIORITY_HIGH , PRIORITY_LOW , JOB_UNKNOWN , JOB_PENDING , JOB_CREATED , JOB_FAILED , JOB_COMPLETE <EOL> from gearman . errors import ExceededConnectionAttempts , ServerUnavailable , InvalidClientState <EOL> from gearman . protocol import submit_cmd_for_background_priority , GEARMAN_COMMAND_STATUS_RES , GEARMAN_COMMAND_GET_STATUS , GEARMAN_COMMAND_JOB_CREATED , GEARMAN_COMMAND_WORK_STATUS , GEARMAN_COMMAND_WORK_FAIL , GEARMAN_COMMAND_WORK_COMPLETE , GEARMAN_COMMAND_WORK_DATA , GEARMAN_COMMAND_WORK_WARNING <EOL> from tests . _core_testing import _GearmanAbstractTest , MockGearmanConnectionManager , MockGearmanConnection <EOL> class MockGearmanClient ( GearmanClient , MockGearmanConnectionManager ) : <EOL> pass <EOL> class ClientTest ( _GearmanAbstractTest ) : <EOL> """<STR_LIT>""" <EOL> connection_manager_class = MockGearmanClient <EOL> command_handler_class = GearmanClientCommandHandler <EOL> def setUp ( self ) : <EOL> super ( ClientTest , self ) . setUp ( ) <EOL> self . original_handle_connection_activity = self . connection_manager . handle_connection_activity <EOL> def tearDown ( self ) : <EOL> super ( ClientTest , self ) . tearDown ( ) <EOL> self . connection_manager . handle_connection_activity = self . original_handle_connection_activity <EOL> def generate_job_request ( self , submitted = True , accepted = True ) : <EOL> current_request = super ( ClientTest , self ) . generate_job_request ( ) <EOL> if submitted or accepted : <EOL> self . connection_manager . establish_request_connection ( current_request ) <EOL> self . command_handler . send_job_request ( current_request ) <EOL> if submitted and accepted : <EOL> self . command_handler . recv_command ( GEARMAN_COMMAND_JOB_CREATED , job_handle = current_request . job . handle ) <EOL> self . assert_ ( current_request . job . handle in self . command_handler . handle_to_request_map ) <EOL> return current_request <EOL> def test_establish_request_connection_complex ( self ) : <EOL> failed_connection = MockGearmanConnection ( ) <EOL> failed_connection . _fail_on_bind = True <EOL> failed_then_retried_connection = MockGearmanConnection ( ) <EOL> failed_then_retried_connection . _fail_on_bind = True <EOL> good_connection = MockGearmanConnection ( ) <EOL> good_connection . connect ( ) <EOL> self . connection_manager . connection_list = [ failed_connection , failed_then_retried_connection , good_connection ] <EOL> current_request = self . generate_job_request ( submitted = False , accepted = False ) <EOL> self . failIf ( current_request in self . connection_manager . request_to_rotating_connection_queue ) <EOL> chosen_connection = self . connection_manager . establish_request_connection ( current_request ) <EOL> self . assertEqual ( chosen_connection , good_connection ) <EOL> self . assertFalse ( failed_connection . connected ) <EOL> self . assertFalse ( failed_then_retried_connection . connected ) <EOL> self . assertTrue ( good_connection . connected ) <EOL> chosen_connection = self . connection_manager . establish_request_connection ( current_request ) <EOL> self . assertEqual ( chosen_connection , good_connection ) <EOL> good_connection . _reset_connection ( ) <EOL> good_connection . _fail_on_bind = True <EOL> failed_then_retried_connection . _fail_on_bind = False <EOL> failed_then_retried_connection . connect ( ) <EOL> chosen_connection = self . connection_manager . establish_request_connection ( current_request ) <EOL> self . assertEqual ( chosen_connection , failed_then_retried_connection ) <EOL> self . assertFalse ( failed_connection . connected ) <EOL> self . assertTrue ( failed_then_retried_connection . connected ) <EOL> self . assertFalse ( good_connection . connected ) <EOL> def test_establish_request_connection_dead ( self ) : <EOL> self . connection_manager . connection_list = [ ] <EOL> self . connection_manager . command_handlers = { } <EOL> current_request = self . generate_job_request ( submitted = False , accepted = False ) <EOL> self . assertRaises ( ServerUnavailable , self . connection_manager . establish_request_connection , current_request ) <EOL> failed_connection = MockGearmanConnection ( ) <EOL> failed_connection . _fail_on_bind = True <EOL> self . connection_manager . connection_list . append ( failed_connection ) <EOL> self . assertRaises ( ServerUnavailable , self . connection_manager . establish_request_connection , current_request ) <EOL> def test_auto_retry_behavior ( self ) : <EOL> current_request = self . generate_job_request ( submitted = False , accepted = False ) <EOL> def fail_then_create_jobs ( rx_conns , wr_conns , ex_conns ) : <EOL> if self . connection_manager . current_failures < self . connection_manager . expected_failures : <EOL> self . connection_manager . current_failures += <NUM_LIT:1> <EOL> self . assertTrue ( self . connection . connected ) <EOL> self . connection_manager . handle_error ( self . connection ) <EOL> self . assertFalse ( self . connection . connected ) <EOL> self . connection_manager . establish_connection ( self . connection ) <EOL> else : <EOL> self . assertEquals ( current_request . state , JOB_PENDING ) <EOL> self . command_handler . recv_command ( GEARMAN_COMMAND_JOB_CREATED , job_handle = current_request . job . handle ) <EOL> return rx_conns , wr_conns , ex_conns <EOL> self . connection_manager . handle_connection_activity = fail_then_create_jobs <EOL> self . connection_manager . expected_failures = <NUM_LIT:5> <EOL> self . connection_manager . current_failures = current_request . connection_attempts = <NUM_LIT:0> <EOL> current_request . max_connection_attempts = self . connection_manager . expected_failures + <NUM_LIT:1> <EOL> current_request . state = JOB_UNKNOWN <EOL> accepted_jobs = self . connection_manager . wait_until_jobs_accepted ( [ current_request ] ) <EOL> self . assertEquals ( current_request . state , JOB_CREATED ) <EOL> self . assertEquals ( current_request . connection_attempts , current_request . max_connection_attempts ) <EOL> self . connection_manager . current_failures = current_request . connection_attempts = <NUM_LIT:0> <EOL> current_request . max_connection_attempts = self . connection_manager . expected_failures <EOL> current_request . state = JOB_UNKNOWN <EOL> self . assertRaises ( ExceededConnectionAttempts , self . connection_manager . wait_until_jobs_accepted , [ current_request ] ) <EOL> self . assertEquals ( current_request . state , JOB_UNKNOWN ) <EOL> self . assertEquals ( current_request . connection_attempts , current_request . max_connection_attempts ) <EOL> def test_multiple_fg_job_submission ( self ) : <EOL> submitted_job_count = <NUM_LIT:5> <EOL> expected_job_list = [ self . generate_job ( ) for _ in xrange ( submitted_job_count ) ] <EOL> def mark_jobs_created ( rx_conns , wr_conns , ex_conns ) : <EOL> for current_job in expected_job_list : <EOL> self . command_handler . recv_command ( GEARMAN_COMMAND_JOB_CREATED , job_handle = current_job . handle ) <EOL> return rx_conns , wr_conns , ex_conns <EOL> self . connection_manager . handle_connection_activity = mark_jobs_created <EOL> job_dictionaries = [ current_job . to_dict ( ) for current_job in expected_job_list ] <EOL> job_requests = self . connection_manager . submit_multiple_jobs ( job_dictionaries , wait_until_complete = False ) <EOL> for current_request , expected_job in zip ( job_requests , expected_job_list ) : <EOL> current_job = current_request . job <EOL> self . assert_jobs_equal ( current_job , expected_job ) <EOL> self . assertEqual ( current_request . priority , PRIORITY_NONE ) <EOL> self . assertEqual ( current_request . background , False ) <EOL> self . assertEqual ( current_request . state , JOB_CREATED ) <EOL> self . assertFalse ( current_request . complete ) <EOL> def test_single_bg_job_submission ( self ) : <EOL> expected_job = self . generate_job ( ) <EOL> def mark_job_created ( rx_conns , wr_conns , ex_conns ) : <EOL> self . command_handler . recv_command ( GEARMAN_COMMAND_JOB_CREATED , job_handle = expected_job . handle ) <EOL> return rx_conns , wr_conns , ex_conns <EOL> self . connection_manager . handle_connection_activity = mark_job_created <EOL> job_request = self . connection_manager . submit_job ( expected_job . task , expected_job . data , unique = expected_job . unique , background = True , priority = PRIORITY_LOW , wait_until_complete = False ) <EOL> current_job = job_request . job <EOL> self . assert_jobs_equal ( current_job , expected_job ) <EOL> self . assertEqual ( job_request . priority , PRIORITY_LOW ) <EOL> self . assertEqual ( job_request . background , True ) <EOL> self . assertEqual ( job_request . state , JOB_CREATED ) <EOL> self . assertTrue ( job_request . complete ) <EOL> def test_single_fg_job_submission_timeout ( self ) : <EOL> expected_job = self . generate_job ( ) <EOL> def job_failed_submission ( rx_conns , wr_conns , ex_conns ) : <EOL> return rx_conns , wr_conns , ex_conns <EOL> self . connection_manager . handle_connection_activity = job_failed_submission <EOL> job_request = self . connection_manager . submit_job ( expected_job . task , expected_job . data , unique = expected_job . unique , priority = PRIORITY_HIGH , poll_timeout = <NUM_LIT> ) <EOL> self . assertEqual ( job_request . priority , PRIORITY_HIGH ) <EOL> self . assertEqual ( job_request . background , False ) <EOL> self . assertEqual ( job_request . state , JOB_PENDING ) <EOL> self . assertFalse ( job_request . complete ) <EOL> self . assertTrue ( job_request . timed_out ) <EOL> def test_wait_for_multiple_jobs_to_complete_or_timeout ( self ) : <EOL> completed_request = self . generate_job_request ( ) <EOL> failed_request = self . generate_job_request ( ) <EOL> timeout_request = self . generate_job_request ( ) <EOL> self . update_requests = True <EOL> def multiple_job_updates ( rx_conns , wr_conns , ex_conns ) : <EOL> if self . update_requests : <EOL> self . command_handler . recv_command ( GEARMAN_COMMAND_WORK_COMPLETE , job_handle = completed_request . job . handle , data = '<STR_LIT>' ) <EOL> self . command_handler . recv_command ( GEARMAN_COMMAND_WORK_FAIL , job_handle = failed_request . job . handle ) <EOL> self . update_requests = False <EOL> return rx_conns , wr_conns , ex_conns <EOL> self . connection_manager . handle_connection_activity = multiple_job_updates <EOL> finished_requests = self . connection_manager . wait_until_jobs_completed ( [ completed_request , failed_request , timeout_request ] , poll_timeout = <NUM_LIT> ) <EOL> del self . update_requests <EOL> finished_completed_request , finished_failed_request , finished_timeout_request = finished_requests <EOL> self . assert_jobs_equal ( finished_completed_request . job , completed_request . job ) <EOL> self . assertEqual ( finished_completed_request . state , JOB_COMPLETE ) <EOL> self . assertEqual ( finished_completed_request . result , '<STR_LIT>' ) <EOL> self . assertFalse ( finished_completed_request . timed_out ) <EOL> self . assert_jobs_equal ( finished_failed_request . job , failed_request . job ) <EOL> self . assertEqual ( finished_failed_request . state , JOB_FAILED ) <EOL> self . assertEqual ( finished_failed_request . result , None ) <EOL> self . assertFalse ( finished_failed_request . timed_out ) <EOL> self . assertEqual ( finished_timeout_request . state , JOB_CREATED ) <EOL> self . assertEqual ( finished_timeout_request . result , None ) <EOL> self . assertTrue ( finished_timeout_request . timed_out ) <EOL> self . assert_ ( finished_timeout_request . job . handle in self . command_handler . handle_to_request_map ) <EOL> def test_get_job_status ( self ) : <EOL> single_request = self . generate_job_request ( ) <EOL> def retrieve_status ( rx_conns , wr_conns , ex_conns ) : <EOL> self . command_handler . recv_command ( GEARMAN_COMMAND_STATUS_RES , job_handle = single_request . job . handle , known = '<STR_LIT:1>' , running = '<STR_LIT:0>' , numerator = '<STR_LIT:0>' , denominator = '<STR_LIT:1>' ) <EOL> return rx_conns , wr_conns , ex_conns <EOL> self . connection_manager . handle_connection_activity = retrieve_status <EOL> job_request = self . connection_manager . get_job_status ( single_request ) <EOL> request_status = job_request . status <EOL> self . failUnless ( request_status ) <EOL> self . assertTrue ( request_status [ '<STR_LIT>' ] ) <EOL> self . assertFalse ( request_status [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( request_status [ '<STR_LIT>' ] , <NUM_LIT:0> ) <EOL> self . assertEqual ( request_status [ '<STR_LIT>' ] , <NUM_LIT:1> ) <EOL> self . assertFalse ( job_request . timed_out ) <EOL> def test_get_job_status_unknown ( self ) : <EOL> single_request = self . generate_job_request ( ) <EOL> current_handle = single_request . job . handle <EOL> self . command_handler . recv_command ( GEARMAN_COMMAND_WORK_FAIL , job_handle = current_handle ) <EOL> def retrieve_status ( rx_conns , wr_conns , ex_conns ) : <EOL> self . command_handler . recv_command ( GEARMAN_COMMAND_STATUS_RES , job_handle = current_handle , known = '<STR_LIT:0>' , running = '<STR_LIT:0>' , numerator = '<STR_LIT:0>' , denominator = '<STR_LIT:1>' ) <EOL> return rx_conns , wr_conns , ex_conns <EOL> self . connection_manager . handle_connection_activity = retrieve_status <EOL> job_request = self . connection_manager . get_job_status ( single_request ) <EOL> request_status = job_request . status <EOL> self . failUnless ( request_status ) <EOL> self . assertFalse ( request_status [ '<STR_LIT>' ] ) <EOL> self . assertFalse ( request_status [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( request_status [ '<STR_LIT>' ] , <NUM_LIT:0> ) <EOL> self . assertEqual ( request_status [ '<STR_LIT>' ] , <NUM_LIT:1> ) <EOL> self . assertFalse ( job_request . timed_out ) <EOL> def test_get_job_status_timeout ( self ) : <EOL> single_request = self . generate_job_request ( ) <EOL> def retrieve_status_timeout ( rx_conns , wr_conns , ex_conns ) : <EOL> return rx_conns , wr_conns , ex_conns <EOL> self . connection_manager . handle_connection_activity = retrieve_status_timeout <EOL> job_request = self . connection_manager . get_job_status ( single_request , poll_timeout = <NUM_LIT> ) <EOL> self . assertTrue ( job_request . timed_out ) <EOL> class ClientCommandHandlerInterfaceTest ( _GearmanAbstractTest ) : <EOL> """<STR_LIT>""" <EOL> connection_manager_class = MockGearmanClient <EOL> command_handler_class = GearmanClientCommandHandler <EOL> def test_send_job_request ( self ) : <EOL> current_request = self . generate_job_request ( ) <EOL> gearman_job = current_request . job <EOL> for priority in ( PRIORITY_NONE , PRIORITY_HIGH , PRIORITY_LOW ) : <EOL> for background in ( False , True ) : <EOL> current_request . reset ( ) <EOL> current_request . priority = priority <EOL> current_request . background = background <EOL> self . command_handler . send_job_request ( current_request ) <EOL> queued_request = self . command_handler . requests_awaiting_handles . popleft ( ) <EOL> self . assertEqual ( queued_request , current_request ) <EOL> expected_cmd_type = submit_cmd_for_background_priority ( background , priority ) <EOL> self . assert_sent_command ( expected_cmd_type , task = gearman_job . task , data = gearman_job . data , unique = gearman_job . unique ) <EOL> def test_get_status_of_job ( self ) : <EOL> current_request = self . generate_job_request ( ) <EOL> self . command_handler . send_get_status_of_job ( current_request ) <EOL> self . assert_sent_command ( GEARMAN_COMMAND_GET_STATUS , job_handle = current_request . job . handle ) <EOL> class ClientCommandHandlerStateMachineTest ( _GearmanAbstractTest ) : <EOL> """<STR_LIT>""" <EOL> connection_manager_class = MockGearmanClient <EOL> command_handler_class = GearmanClientCommandHandler <EOL> def generate_job_request ( self , submitted = True , accepted = True ) : <EOL> current_request = super ( ClientCommandHandlerStateMachineTest , self ) . generate_job_request ( ) <EOL> if submitted or accepted : <EOL> self . command_handler . requests_awaiting_handles . append ( current_request ) <EOL> current_request . state = JOB_PENDING <EOL> if submitted and accepted : <EOL> self . command_handler . recv_command ( GEARMAN_COMMAND_JOB_CREATED , job_handle = current_request . job . handle ) <EOL> return current_request <EOL> def test_received_job_created ( self ) : <EOL> current_request = self . generate_job_request ( accepted = False ) <EOL> new_handle = str ( random . random ( ) ) <EOL> self . command_handler . recv_command ( GEARMAN_COMMAND_JOB_CREATED , job_handle = new_handle ) <EOL> self . assertEqual ( current_request . job . handle , new_handle ) <EOL> self . assertEqual ( current_request . state , JOB_CREATED ) <EOL> self . assertEqual ( self . command_handler . handle_to_request_map [ new_handle ] , current_request ) <EOL> def test_received_job_created_out_of_order ( self ) : <EOL> self . assertEqual ( self . command_handler . requests_awaiting_handles , collections . deque ( ) ) <EOL> self . assertRaises ( InvalidClientState , self . command_handler . recv_command , GEARMAN_COMMAND_JOB_CREATED , job_handle = None ) <EOL> def test_required_state_pending ( self ) : <EOL> current_request = self . generate_job_request ( submitted = False , accepted = False ) <EOL> new_handle = str ( random . random ( ) ) <EOL> invalid_states = [ JOB_UNKNOWN , JOB_CREATED , JOB_COMPLETE , JOB_FAILED ] <EOL> for bad_state in invalid_states : <EOL> current_request . state = bad_state <EOL> self . command_handler . requests_awaiting_handles . append ( current_request ) <EOL> self . assertRaises ( InvalidClientState , self . command_handler . recv_command , GEARMAN_COMMAND_JOB_CREATED , job_handle = new_handle ) <EOL> def test_required_state_queued ( self ) : <EOL> current_request = self . generate_job_request ( ) <EOL> job_handle = current_request . job . handle <EOL> new_data = str ( random . random ( ) ) <EOL> invalid_states = [ JOB_UNKNOWN , JOB_PENDING , JOB_COMPLETE , JOB_FAILED ] <EOL> for bad_state in invalid_states : <EOL> current_request . state = bad_state <EOL> self . assertRaises ( InvalidClientState , self . command_handler . recv_command , GEARMAN_COMMAND_WORK_DATA , job_handle = job_handle , data = new_data ) <EOL> self . assertRaises ( InvalidClientState , self . command_handler . recv_command , GEARMAN_COMMAND_WORK_WARNING , job_handle = job_handle , data = new_data ) <EOL> self . assertRaises ( InvalidClientState , self . command_handler . recv_command , GEARMAN_COMMAND_WORK_STATUS , job_handle = job_handle , numerator = <NUM_LIT:0> , denominator = <NUM_LIT:1> ) <EOL> self . assertRaises ( InvalidClientState , self . command_handler . recv_command , GEARMAN_COMMAND_WORK_COMPLETE , job_handle = job_handle , data = new_data ) <EOL> self . assertRaises ( InvalidClientState , self . command_handler . recv_command , GEARMAN_COMMAND_WORK_FAIL , job_handle = job_handle ) <EOL> def test_in_flight_work_updates ( self ) : <EOL> current_request = self . generate_job_request ( ) <EOL> job_handle = current_request . job . handle <EOL> new_data = str ( random . random ( ) ) <EOL> self . command_handler . recv_command ( GEARMAN_COMMAND_WORK_DATA , job_handle = job_handle , data = new_data ) <EOL> self . assertEqual ( current_request . data_updates . popleft ( ) , new_data ) <EOL> self . assertEqual ( current_request . state , JOB_CREATED ) <EOL> self . command_handler . recv_command ( GEARMAN_COMMAND_WORK_WARNING , job_handle = job_handle , data = new_data ) <EOL> self . assertEqual ( current_request . warning_updates . popleft ( ) , new_data ) <EOL> self . assertEqual ( current_request . state , JOB_CREATED ) <EOL> self . command_handler . recv_command ( GEARMAN_COMMAND_WORK_STATUS , job_handle = job_handle , numerator = <NUM_LIT:0> , denominator = <NUM_LIT:1> ) <EOL> self . assertEqual ( current_request . status_updates . popleft ( ) , ( <NUM_LIT:0> , <NUM_LIT:1> ) ) <EOL> self . assertEqual ( current_request . state , JOB_CREATED ) <EOL> def test_work_complete ( self ) : <EOL> current_request = self . generate_job_request ( ) <EOL> job_handle = current_request . job . handle <EOL> new_data = str ( random . random ( ) ) <EOL> self . command_handler . recv_command ( GEARMAN_COMMAND_WORK_COMPLETE , job_handle = job_handle , data = new_data ) <EOL> self . assertEqual ( current_request . result , new_data ) <EOL> self . assertEqual ( current_request . state , JOB_COMPLETE ) <EOL> def test_work_fail ( self ) : <EOL> current_request = self . generate_job_request ( ) <EOL> job_handle = current_request . job . handle <EOL> new_data = str ( random . random ( ) ) <EOL> self . command_handler . recv_command ( GEARMAN_COMMAND_WORK_FAIL , job_handle = job_handle ) <EOL> self . assertEqual ( current_request . state , JOB_FAILED ) <EOL> def test_status_request ( self ) : <EOL> current_request = self . generate_job_request ( ) <EOL> job_handle = current_request . job . handle <EOL> self . assertEqual ( current_request . status , { } ) <EOL> self . command_handler . recv_command ( GEARMAN_COMMAND_STATUS_RES , job_handle = job_handle , known = '<STR_LIT:1>' , running = '<STR_LIT:1>' , numerator = '<STR_LIT:0>' , denominator = '<STR_LIT:1>' ) <EOL> self . assertEqual ( current_request . status [ '<STR_LIT>' ] , job_handle ) <EOL> self . assertTrue ( current_request . status [ '<STR_LIT>' ] ) <EOL> self . assertTrue ( current_request . status [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( current_request . status [ '<STR_LIT>' ] , <NUM_LIT:0> ) <EOL> self . assertEqual ( current_request . status [ '<STR_LIT>' ] , <NUM_LIT:1> ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> unittest . main ( ) </s>
