Description: Include tests directory
 Include the tests directory from the git repository. This file isn't in the 
 upstream tarball, but we like to run tests during our build process, so add 
 it manually.
Origin: upstream, https://github.com/Yelp/python-gearman/tree/master/tests
Bug: https://github.com/Yelp/python-gearman/issues/12
Forwarded: yes
Last-Update: 2011-05-06

Index: python-gearman-2.0.2/tests/_core_testing.py
===================================================================
--- /dev/null	1970-01-01 00:00:00.000000000 +0000
+++ python-gearman-2.0.2/tests/_core_testing.py	2011-05-06 22:21:12.293469647 +0200
@@ -0,0 +1,106 @@
+import collections
+import random
+import unittest
+
+import gearman.util
+from gearman.command_handler import GearmanCommandHandler
+from gearman.connection import GearmanConnection
+from gearman.connection_manager import GearmanConnectionManager, NoopEncoder
+
+from gearman.constants import PRIORITY_NONE, PRIORITY_HIGH, PRIORITY_LOW, DEFAULT_GEARMAN_PORT, JOB_UNKNOWN, JOB_CREATED
+from gearman.errors import ConnectionError
+from gearman.job import GearmanJob, GearmanJobRequest
+from gearman.protocol import get_command_name
+
+class MockGearmanConnection(GearmanConnection):
+    def __init__(self, host=None, port=DEFAULT_GEARMAN_PORT):
+        host = host or '__testing_host__'
+        super(MockGearmanConnection, self).__init__(host=host, port=port)
+
+        self._fail_on_bind = False
+        self._fail_on_read = False
+        self._fail_on_write = False
+
+    def _create_client_socket(self):
+        if self._fail_on_bind:
+            self.throw_exception(message='mock bind failure')
+
+    def read_data_from_socket(self):
+        if self._fail_on_read:
+            self.throw_exception(message='mock read failure')
+
+    def send_data_to_socket(self):
+        if self._fail_on_write:
+            self.throw_exception(message='mock write failure')
+
+    def __repr__(self):
+        return ('<GearmanConnection %s:%d connected=%s> (%s)' %
+            (self.gearman_host, self.gearman_port, self.connected, id(self)))
+
+class MockGearmanConnectionManager(GearmanConnectionManager):
+    """Handy mock client base to test Worker/Client/Abstract ClientBases"""
+    def poll_connections_once(self, connections, timeout=None):
+        return set(), set(), set()
+
+class _GearmanAbstractTest(unittest.TestCase):
+    connection_class = MockGearmanConnection
+    connection_manager_class = MockGearmanConnectionManager
+    command_handler_class = None
+
+    job_class = GearmanJob
+    job_request_class = GearmanJobRequest
+
+    def setUp(self):
+        # Create a new MockGearmanTestClient on the fly
+        self.setup_connection_manager()
+        self.setup_connection()
+        self.setup_command_handler()
+
+    def setup_connection_manager(self):
+        testing_attributes = {'command_handler_class': self.command_handler_class, 'connection_class': self.connection_class}
+        testing_client_class = type('MockGearmanTestingClient', (self.connection_manager_class, ), testing_attributes)
+
+        self.connection_manager = testing_client_class()
+
+    def setup_connection(self):
+        self.connection = self.connection_class()
+        self.connection_manager.connection_list = [self.connection]
+
+    def setup_command_handler(self):
+        self.connection_manager.establish_connection(self.connection)
+        self.command_handler = self.connection_manager.connection_to_handler_map[self.connection]
+
+    def generate_job(self):
+        return self.job_class(self.connection, handle=str(random.random()), task='__test_ability__', unique=str(random.random()), data=str(random.random()))
+
+    def generate_job_dict(self):
+        current_job = self.generate_job()
+        return current_job.to_dict()
+
+    def generate_job_request(self, priority=PRIORITY_NONE, background=False):
+        job_handle = str(random.random())
+        current_job = self.job_class(connection=self.connection, handle=job_handle, task='__test_ability__', unique=str(random.random()), data=str(random.random()))
+        current_request = self.job_request_class(current_job, initial_priority=priority, background=background)
+
+        self.assertEqual(current_request.state, JOB_UNKNOWN)
+
+        return current_request
+
+    def assert_jobs_equal(self, job_actual, job_expected):
+        # Validates that GearmanJobs are essentially equal
+        self.assertEqual(job_actual.handle, job_expected.handle)
+        self.assertEqual(job_actual.task, job_expected.task)
+        self.assertEqual(job_actual.unique, job_expected.unique)
+        self.assertEqual(job_actual.data, job_expected.data)
+
+    def assert_sent_command(self, expected_cmd_type, **expected_cmd_args):
+        # Make sure any commands we're passing through the CommandHandler gets properly passed through to the client base
+        client_cmd_type, client_cmd_args = self.connection._outgoing_commands.popleft()
+        self.assert_commands_equal(client_cmd_type, expected_cmd_type)
+        self.assertEqual(client_cmd_args, expected_cmd_args)
+
+    def assert_no_pending_commands(self):
+        self.assertEqual(self.connection._outgoing_commands, collections.deque())
+
+    def assert_commands_equal(self, cmd_type_actual, cmd_type_expected):
+        self.assertEqual(get_command_name(cmd_type_actual), get_command_name(cmd_type_expected))
Index: python-gearman-2.0.2/tests/admin_client_tests.py
===================================================================
--- /dev/null	1970-01-01 00:00:00.000000000 +0000
+++ python-gearman-2.0.2/tests/admin_client_tests.py	2011-05-06 22:21:12.297469268 +0200
@@ -0,0 +1,153 @@
+import unittest
+
+from gearman.admin_client import GearmanAdminClient, ECHO_STRING
+from gearman.admin_client_handler import GearmanAdminClientCommandHandler
+
+from gearman.errors import InvalidAdminClientState, ProtocolError
+from gearman.protocol import GEARMAN_COMMAND_ECHO_RES, GEARMAN_COMMAND_ECHO_REQ, GEARMAN_COMMAND_TEXT_COMMAND, \
+    GEARMAN_SERVER_COMMAND_STATUS, GEARMAN_SERVER_COMMAND_VERSION, GEARMAN_SERVER_COMMAND_WORKERS, GEARMAN_SERVER_COMMAND_MAXQUEUE, GEARMAN_SERVER_COMMAND_SHUTDOWN
+
+from tests._core_testing import _GearmanAbstractTest, MockGearmanConnectionManager, MockGearmanConnection
+
+class MockGearmanAdminClient(GearmanAdminClient, MockGearmanConnectionManager):
+    pass
+
+class CommandHandlerStateMachineTest(_GearmanAbstractTest):
+    """Test the public interface a GearmanWorker may need to call in order to update state on a GearmanWorkerCommandHandler"""
+    connection_manager_class = MockGearmanAdminClient
+    command_handler_class = GearmanAdminClientCommandHandler
+
+    def setUp(self):
+        super(CommandHandlerStateMachineTest, self).setUp()
+        self.connection_manager.current_connection = self.connection
+        self.connection_manager.current_handler = self.command_handler
+
+    def test_send_illegal_server_commands(self):
+        self.assertRaises(ProtocolError, self.send_server_command, "This is not a server command")
+
+    def test_ping_server(self):
+        self.command_handler.send_echo_request(ECHO_STRING)
+        self.assert_sent_command(GEARMAN_COMMAND_ECHO_REQ, data=ECHO_STRING)
+        self.assertEqual(self.command_handler._sent_commands[0], GEARMAN_COMMAND_ECHO_REQ)
+
+        self.command_handler.recv_command(GEARMAN_COMMAND_ECHO_RES, data=ECHO_STRING)
+        server_response = self.pop_response(GEARMAN_COMMAND_ECHO_REQ)
+        self.assertEquals(server_response, ECHO_STRING)
+
+    def test_state_and_protocol_errors_for_status(self):
+        self.send_server_command(GEARMAN_SERVER_COMMAND_STATUS)
+
+        # Test premature popping as this we aren't until ready we see the '.'
+        self.assertRaises(InvalidAdminClientState, self.pop_response, GEARMAN_SERVER_COMMAND_STATUS)
+
+        # Test malformed server status
+        self.assertRaises(ProtocolError, self.recv_server_response, '\t'.join(['12', 'IP-A', 'CLIENT-A']))
+
+        self.recv_server_response('.')
+
+        server_response = self.pop_response(GEARMAN_SERVER_COMMAND_STATUS)
+        self.assertEquals(server_response, tuple())
+
+    def test_multiple_status(self):
+        self.send_server_command(GEARMAN_SERVER_COMMAND_STATUS)
+        self.recv_server_response('\t'.join(['test_function', '1', '5', '17']))
+        self.recv_server_response('\t'.join(['another_function', '2', '4', '23']))
+        self.recv_server_response('.')
+
+        server_response = self.pop_response(GEARMAN_SERVER_COMMAND_STATUS)
+        self.assertEquals(len(server_response), 2)
+
+        test_response, another_response = server_response
+        self.assertEquals(test_response['task'], 'test_function')
+        self.assertEquals(test_response['queued'], 1)
+        self.assertEquals(test_response['running'], 5)
+        self.assertEquals(test_response['workers'],  17)
+
+        self.assertEquals(another_response['task'], 'another_function')
+        self.assertEquals(another_response['queued'], 2)
+        self.assertEquals(another_response['running'], 4)
+        self.assertEquals(another_response['workers'],  23)
+
+    def test_version(self):
+        expected_version = '0.12345'
+
+        self.send_server_command(GEARMAN_SERVER_COMMAND_VERSION)
+        self.recv_server_response(expected_version)
+
+        server_response = self.pop_response(GEARMAN_SERVER_COMMAND_VERSION)
+        self.assertEquals(expected_version, server_response)
+
+    def test_state_and_protocol_errors_for_workers(self):
+        self.send_server_command(GEARMAN_SERVER_COMMAND_WORKERS)
+
+        # Test premature popping as this we aren't until ready we see the '.'
+        self.assertRaises(InvalidAdminClientState, self.pop_response, GEARMAN_SERVER_COMMAND_WORKERS)
+
+        # Test malformed responses
+        self.assertRaises(ProtocolError, self.recv_server_response, ' '.join(['12', 'IP-A', 'CLIENT-A']))
+        self.assertRaises(ProtocolError, self.recv_server_response, ' '.join(['12', 'IP-A', 'CLIENT-A', 'NOT:']))
+
+        self.recv_server_response('.')
+
+        server_response = self.pop_response(GEARMAN_SERVER_COMMAND_WORKERS)
+        self.assertEquals(server_response, tuple())
+
+    def test_multiple_workers(self):
+        self.send_server_command(GEARMAN_SERVER_COMMAND_WORKERS)
+        self.recv_server_response(' '.join(['12', 'IP-A', 'CLIENT-A', ':', 'function-A', 'function-B']))
+        self.recv_server_response(' '.join(['13', 'IP-B', 'CLIENT-B', ':', 'function-C']))
+        self.recv_server_response('.')
+
+        server_response = self.pop_response(GEARMAN_SERVER_COMMAND_WORKERS)
+        self.assertEquals(len(server_response), 2)
+
+        test_response, another_response = server_response
+        self.assertEquals(test_response['file_descriptor'], '12')
+        self.assertEquals(test_response['ip'], 'IP-A')
+        self.assertEquals(test_response['client_id'], 'CLIENT-A')
+        self.assertEquals(test_response['tasks'],  ('function-A', 'function-B'))
+
+        self.assertEquals(another_response['file_descriptor'], '13')
+        self.assertEquals(another_response['ip'], 'IP-B')
+        self.assertEquals(another_response['client_id'], 'CLIENT-B')
+        self.assertEquals(another_response['tasks'],  ('function-C', ))
+
+    def test_maxqueue(self):
+        self.send_server_command(GEARMAN_SERVER_COMMAND_MAXQUEUE)
+        self.assertRaises(ProtocolError, self.recv_server_response, 'NOT OK')
+
+        # Pop prematurely
+        self.assertRaises(InvalidAdminClientState, self.pop_response, GEARMAN_SERVER_COMMAND_MAXQUEUE)
+
+        self.recv_server_response('OK')
+        server_response = self.pop_response(GEARMAN_SERVER_COMMAND_MAXQUEUE)
+        self.assertEquals(server_response, 'OK')
+
+    def test_shutdown(self):
+        self.send_server_command(GEARMAN_SERVER_COMMAND_SHUTDOWN)
+
+        # Pop prematurely
+        self.assertRaises(InvalidAdminClientState, self.pop_response, GEARMAN_SERVER_COMMAND_SHUTDOWN)
+
+        self.recv_server_response(None)
+        server_response = self.pop_response(GEARMAN_SERVER_COMMAND_SHUTDOWN)
+        self.assertEquals(server_response, None)
+
+    def send_server_command(self, expected_command):
+        self.command_handler.send_text_command(expected_command)
+        expected_line = "%s\n" % expected_command
+        self.assert_sent_command(GEARMAN_COMMAND_TEXT_COMMAND, raw_text=expected_line)
+
+        self.assertEqual(self.command_handler._sent_commands[0], expected_command)
+
+    def recv_server_response(self, response_line):
+        self.command_handler.recv_command(GEARMAN_COMMAND_TEXT_COMMAND, raw_text=response_line)
+
+    def pop_response(self, expected_command):
+        server_cmd, server_response = self.command_handler.pop_response()
+        self.assertEquals(expected_command, server_cmd)
+
+        return server_response
+
+if __name__ == '__main__':
+    unittest.main()
Index: python-gearman-2.0.2/tests/client_tests.py
===================================================================
--- /dev/null	1970-01-01 00:00:00.000000000 +0000
+++ python-gearman-2.0.2/tests/client_tests.py	2011-05-06 22:21:12.297469268 +0200
@@ -0,0 +1,450 @@
+import collections
+import random
+import unittest
+
+from gearman.client import GearmanClient
+from gearman.client_handler import GearmanClientCommandHandler
+
+from gearman.constants import PRIORITY_NONE, PRIORITY_HIGH, PRIORITY_LOW, JOB_UNKNOWN, JOB_PENDING, JOB_CREATED, JOB_FAILED, JOB_COMPLETE
+from gearman.errors import ExceededConnectionAttempts, ServerUnavailable, InvalidClientState
+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
+
+from tests._core_testing import _GearmanAbstractTest, MockGearmanConnectionManager, MockGearmanConnection
+
+class MockGearmanClient(GearmanClient, MockGearmanConnectionManager):
+    pass
+
+class ClientTest(_GearmanAbstractTest):
+    """Test the public client interface"""
+    connection_manager_class = MockGearmanClient
+    command_handler_class = GearmanClientCommandHandler
+
+    def setUp(self):
+        super(ClientTest, self).setUp()
+        self.original_handle_connection_activity = self.connection_manager.handle_connection_activity
+
+    def tearDown(self):
+        super(ClientTest, self).tearDown()
+        self.connection_manager.handle_connection_activity = self.original_handle_connection_activity
+
+    def generate_job_request(self, submitted=True, accepted=True):
+        current_request = super(ClientTest, self).generate_job_request()
+        if submitted or accepted:
+            self.connection_manager.establish_request_connection(current_request)
+            self.command_handler.send_job_request(current_request)
+
+        if submitted and accepted:
+            self.command_handler.recv_command(GEARMAN_COMMAND_JOB_CREATED, job_handle=current_request.job.handle)
+            self.assert_(current_request.job.handle in self.command_handler.handle_to_request_map)
+
+        return current_request
+
+    def test_establish_request_connection_complex(self):
+        # Spin up a bunch of imaginary gearman connections
+        failed_connection = MockGearmanConnection()
+        failed_connection._fail_on_bind = True
+
+        failed_then_retried_connection = MockGearmanConnection()
+        failed_then_retried_connection._fail_on_bind = True
+
+        good_connection = MockGearmanConnection()
+        good_connection.connect()
+
+        # Register all our connections
+        self.connection_manager.connection_list = [failed_connection, failed_then_retried_connection, good_connection]
+
+        # When we first create our request, our client shouldn't know anything about it
+        current_request = self.generate_job_request(submitted=False, accepted=False)
+        self.failIf(current_request in self.connection_manager.request_to_rotating_connection_queue)
+
+        # Make sure that when we start up, we get our good connection
+        chosen_connection = self.connection_manager.establish_request_connection(current_request)
+        self.assertEqual(chosen_connection, good_connection)
+
+        self.assertFalse(failed_connection.connected)
+        self.assertFalse(failed_then_retried_connection.connected)
+        self.assertTrue(good_connection.connected)
+
+        # No state changed so we should still go to the correct connection
+        chosen_connection = self.connection_manager.establish_request_connection(current_request)
+        self.assertEqual(chosen_connection, good_connection)
+
+        # Pretend like our good connection died so we'll need to choose somethign else
+        good_connection._reset_connection()
+        good_connection._fail_on_bind = True
+
+        failed_then_retried_connection._fail_on_bind = False
+        failed_then_retried_connection.connect()
+
+        # Make sure we rotate good_connection and failed_connection out
+        chosen_connection = self.connection_manager.establish_request_connection(current_request)
+        self.assertEqual(chosen_connection, failed_then_retried_connection)
+        self.assertFalse(failed_connection.connected)
+        self.assertTrue(failed_then_retried_connection.connected)
+        self.assertFalse(good_connection.connected)
+
+    def test_establish_request_connection_dead(self):
+        self.connection_manager.connection_list = []
+        self.connection_manager.command_handlers = {}
+
+        current_request = self.generate_job_request(submitted=False, accepted=False)
+
+        # No connections == death
+        self.assertRaises(ServerUnavailable, self.connection_manager.establish_request_connection, current_request)
+
+        # Spin up a bunch of imaginary gearman connections
+        failed_connection = MockGearmanConnection()
+        failed_connection._fail_on_bind = True
+        self.connection_manager.connection_list.append(failed_connection)
+
+        # All failed connections == death
+        self.assertRaises(ServerUnavailable, self.connection_manager.establish_request_connection, current_request)
+
+    def test_auto_retry_behavior(self):
+        current_request = self.generate_job_request(submitted=False, accepted=False)
+
+        def fail_then_create_jobs(rx_conns, wr_conns, ex_conns):
+            if self.connection_manager.current_failures < self.connection_manager.expected_failures:
+                self.connection_manager.current_failures += 1
+
+                # We're going to down this connection and reset state
+                self.assertTrue(self.connection.connected)
+                self.connection_manager.handle_error(self.connection)
+                self.assertFalse(self.connection.connected)
+
+                # We're then going to IMMEDIATELY pull this connection back up
+                # So we don't bail out of the "self.connection_manager.poll_connections_until_stopped" loop
+                self.connection_manager.establish_connection(self.connection)
+            else:
+                self.assertEquals(current_request.state, JOB_PENDING)
+                self.command_handler.recv_command(GEARMAN_COMMAND_JOB_CREATED, job_handle=current_request.job.handle)
+
+            return rx_conns, wr_conns, ex_conns
+
+        self.connection_manager.handle_connection_activity = fail_then_create_jobs
+        self.connection_manager.expected_failures = 5
+
+        # Now that we've setup our rety behavior, we need to reset the entire state of our experiment
+        # First pass should succeed as we JUST touch our max attempts
+        self.connection_manager.current_failures = current_request.connection_attempts = 0
+        current_request.max_connection_attempts = self.connection_manager.expected_failures + 1
+        current_request.state = JOB_UNKNOWN
+
+        accepted_jobs = self.connection_manager.wait_until_jobs_accepted([current_request])
+        self.assertEquals(current_request.state, JOB_CREATED)
+        self.assertEquals(current_request.connection_attempts, current_request.max_connection_attempts)
+
+        # Second pass should fail as we JUST exceed our max attempts
+        self.connection_manager.current_failures = current_request.connection_attempts = 0
+        current_request.max_connection_attempts = self.connection_manager.expected_failures
+        current_request.state = JOB_UNKNOWN
+
+        self.assertRaises(ExceededConnectionAttempts, self.connection_manager.wait_until_jobs_accepted, [current_request])
+        self.assertEquals(current_request.state, JOB_UNKNOWN)
+        self.assertEquals(current_request.connection_attempts, current_request.max_connection_attempts)
+
+    def test_multiple_fg_job_submission(self):
+        submitted_job_count = 5
+        expected_job_list = [self.generate_job() for _ in xrange(submitted_job_count)]
+        def mark_jobs_created(rx_conns, wr_conns, ex_conns):
+            for current_job in expected_job_list:
+                self.command_handler.recv_command(GEARMAN_COMMAND_JOB_CREATED, job_handle=current_job.handle)
+
+            return rx_conns, wr_conns, ex_conns
+
+        self.connection_manager.handle_connection_activity = mark_jobs_created
+
+        job_dictionaries = [current_job.to_dict() for current_job in expected_job_list]
+
+        # Test multiple job submission
+        job_requests = self.connection_manager.submit_multiple_jobs(job_dictionaries, wait_until_complete=False)
+        for current_request, expected_job in zip(job_requests, expected_job_list):
+            current_job = current_request.job
+            self.assert_jobs_equal(current_job, expected_job)
+
+            self.assertEqual(current_request.priority, PRIORITY_NONE)
+            self.assertEqual(current_request.background, False)
+            self.assertEqual(current_request.state, JOB_CREATED)
+
+            self.assertFalse(current_request.complete)
+
+    def test_single_bg_job_submission(self):
+        expected_job = self.generate_job()
+        def mark_job_created(rx_conns, wr_conns, ex_conns):
+            self.command_handler.recv_command(GEARMAN_COMMAND_JOB_CREATED, job_handle=expected_job.handle)
+            return rx_conns, wr_conns, ex_conns
+
+        self.connection_manager.handle_connection_activity = mark_job_created
+        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)
+
+        current_job = job_request.job
+        self.assert_jobs_equal(current_job, expected_job)
+
+        self.assertEqual(job_request.priority, PRIORITY_LOW)
+        self.assertEqual(job_request.background, True)
+        self.assertEqual(job_request.state, JOB_CREATED)
+
+        self.assertTrue(job_request.complete)
+
+    def test_single_fg_job_submission_timeout(self):
+        expected_job = self.generate_job()
+        def job_failed_submission(rx_conns, wr_conns, ex_conns):
+            return rx_conns, wr_conns, ex_conns
+
+        self.connection_manager.handle_connection_activity = job_failed_submission
+        job_request = self.connection_manager.submit_job(expected_job.task, expected_job.data, unique=expected_job.unique, priority=PRIORITY_HIGH, poll_timeout=0.01)
+
+        self.assertEqual(job_request.priority, PRIORITY_HIGH)
+        self.assertEqual(job_request.background, False)
+        self.assertEqual(job_request.state, JOB_PENDING)
+
+        self.assertFalse(job_request.complete)
+        self.assertTrue(job_request.timed_out)
+
+    def test_wait_for_multiple_jobs_to_complete_or_timeout(self):
+        completed_request = self.generate_job_request()
+        failed_request = self.generate_job_request()
+        timeout_request = self.generate_job_request()
+
+        self.update_requests = True
+        def multiple_job_updates(rx_conns, wr_conns, ex_conns):
+            # Only give a single status update and have the 3rd job handle timeout
+            if self.update_requests:
+                self.command_handler.recv_command(GEARMAN_COMMAND_WORK_COMPLETE, job_handle=completed_request.job.handle, data='12345')
+                self.command_handler.recv_command(GEARMAN_COMMAND_WORK_FAIL, job_handle=failed_request.job.handle)
+                self.update_requests = False
+
+            return rx_conns, wr_conns, ex_conns
+
+        self.connection_manager.handle_connection_activity = multiple_job_updates
+
+        finished_requests = self.connection_manager.wait_until_jobs_completed([completed_request, failed_request, timeout_request], poll_timeout=0.01)
+        del self.update_requests
+
+        finished_completed_request, finished_failed_request, finished_timeout_request = finished_requests
+
+        self.assert_jobs_equal(finished_completed_request.job, completed_request.job)
+        self.assertEqual(finished_completed_request.state, JOB_COMPLETE)
+        self.assertEqual(finished_completed_request.result, '12345')
+        self.assertFalse(finished_completed_request.timed_out)
+        self.assert_(finished_completed_request.job.handle not in self.command_handler.handle_to_request_map)
+
+        self.assert_jobs_equal(finished_failed_request.job, failed_request.job)
+        self.assertEqual(finished_failed_request.state, JOB_FAILED)
+        self.assertEqual(finished_failed_request.result, None)
+        self.assertFalse(finished_failed_request.timed_out)
+        self.assert_(finished_failed_request.job.handle not in self.command_handler.handle_to_request_map)
+
+        self.assertEqual(finished_timeout_request.state, JOB_CREATED)
+        self.assertEqual(finished_timeout_request.result, None)
+        self.assertTrue(finished_timeout_request.timed_out)
+        self.assert_(finished_timeout_request.job.handle in self.command_handler.handle_to_request_map)
+
+    def test_get_job_status(self):
+        single_request = self.generate_job_request()
+
+        def retrieve_status(rx_conns, wr_conns, ex_conns):
+            self.command_handler.recv_command(GEARMAN_COMMAND_STATUS_RES, job_handle=single_request.job.handle, known='1', running='0', numerator='0', denominator='1')
+            return rx_conns, wr_conns, ex_conns
+
+        self.connection_manager.handle_connection_activity = retrieve_status
+
+        job_request = self.connection_manager.get_job_status(single_request)
+        request_status = job_request.status
+        self.failUnless(request_status)
+        self.assertTrue(request_status['known'])
+        self.assertFalse(request_status['running'])
+        self.assertEqual(request_status['numerator'], 0)
+        self.assertEqual(request_status['denominator'], 1)
+        self.assertFalse(job_request.timed_out)
+
+    def test_get_job_status_unknown(self):
+        single_request = self.generate_job_request()
+        current_handle = single_request.job.handle
+        self.command_handler.recv_command(GEARMAN_COMMAND_WORK_FAIL, job_handle=current_handle)
+
+        def retrieve_status(rx_conns, wr_conns, ex_conns):
+            self.command_handler.recv_command(GEARMAN_COMMAND_STATUS_RES, job_handle=current_handle, known='0', running='0', numerator='0', denominator='1')
+            return rx_conns, wr_conns, ex_conns
+
+        self.connection_manager.handle_connection_activity = retrieve_status
+
+        job_request = self.connection_manager.get_job_status(single_request)
+        request_status = job_request.status
+        self.failUnless(request_status)
+        self.assertFalse(request_status['known'])
+        self.assertFalse(request_status['running'])
+        self.assertEqual(request_status['numerator'], 0)
+        self.assertEqual(request_status['denominator'], 1)
+        self.assertFalse(job_request.timed_out)
+        self.assert_(current_handle not in self.command_handler.handle_to_request_map)
+
+    def test_get_job_status_timeout(self):
+        single_request = self.generate_job_request()
+
+        def retrieve_status_timeout(rx_conns, wr_conns, ex_conns):
+            pass
+
+        self.connection_manager.handle_connection_activity = retrieve_status_timeout
+
+        job_request = self.connection_manager.get_job_status(single_request, poll_timeout=0.01)
+        self.assertTrue(job_request.timed_out)
+
+
+class ClientCommandHandlerInterfaceTest(_GearmanAbstractTest):
+    """Test the public interface a GearmanClient may need to call in order to update state on a GearmanClientCommandHandler"""
+    connection_manager_class = MockGearmanClient
+    command_handler_class = GearmanClientCommandHandler
+
+    def test_send_job_request(self):
+        current_request = self.generate_job_request()
+        gearman_job = current_request.job
+
+        for priority in (PRIORITY_NONE, PRIORITY_HIGH, PRIORITY_LOW):
+            for background in (False, True):
+                current_request.reset()
+                current_request.priority = priority
+                current_request.background = background
+
+                self.command_handler.send_job_request(current_request)
+
+                queued_request = self.command_handler.requests_awaiting_handles.popleft()
+                self.assertEqual(queued_request, current_request)
+
+                expected_cmd_type = submit_cmd_for_background_priority(background, priority)
+                self.assert_sent_command(expected_cmd_type, task=gearman_job.task, data=gearman_job.data, unique=gearman_job.unique)
+
+    def test_get_status_of_job(self):
+        current_request = self.generate_job_request()
+
+        self.command_handler.send_get_status_of_job(current_request)
+
+        self.assert_sent_command(GEARMAN_COMMAND_GET_STATUS, job_handle=current_request.job.handle)
+
+
+class ClientCommandHandlerStateMachineTest(_GearmanAbstractTest):
+    """Test single state transitions within a GearmanWorkerCommandHandler"""
+    connection_manager_class = MockGearmanClient
+    command_handler_class = GearmanClientCommandHandler
+
+    def generate_job_request(self, submitted=True, accepted=True):
+        current_request = super(ClientCommandHandlerStateMachineTest, self).generate_job_request()
+        if submitted or accepted:
+            self.command_handler.requests_awaiting_handles.append(current_request)
+            current_request.state = JOB_PENDING
+
+        if submitted and accepted:
+            self.command_handler.recv_command(GEARMAN_COMMAND_JOB_CREATED, job_handle=current_request.job.handle)
+
+        return current_request
+
+    def test_received_job_created(self):
+        current_request = self.generate_job_request(accepted=False)
+
+        new_handle = str(random.random())
+        self.command_handler.recv_command(GEARMAN_COMMAND_JOB_CREATED, job_handle=new_handle)
+
+        self.assertEqual(current_request.job.handle, new_handle)
+        self.assertEqual(current_request.state, JOB_CREATED)
+        self.assertEqual(self.command_handler.handle_to_request_map[new_handle], current_request)
+
+    def test_received_job_created_out_of_order(self):
+        self.assertEqual(self.command_handler.requests_awaiting_handles, collections.deque())
+
+        # Make sure we bail cuz we have an empty queue
+        self.assertRaises(InvalidClientState, self.command_handler.recv_command, GEARMAN_COMMAND_JOB_CREATED, job_handle=None)
+
+    def test_required_state_pending(self):
+        current_request = self.generate_job_request(submitted=False, accepted=False)
+
+        new_handle = str(random.random())
+
+        invalid_states = [JOB_UNKNOWN, JOB_CREATED, JOB_COMPLETE, JOB_FAILED]
+        for bad_state in invalid_states:
+            current_request.state = bad_state
+
+            # We only want to check the state of request... not die if we don't have any pending requests
+            self.command_handler.requests_awaiting_handles.append(current_request)
+
+            self.assertRaises(InvalidClientState, self.command_handler.recv_command, GEARMAN_COMMAND_JOB_CREATED, job_handle=new_handle)
+
+    def test_required_state_queued(self):
+        current_request = self.generate_job_request()
+
+        job_handle = current_request.job.handle
+        new_data = str(random.random())
+
+        invalid_states = [JOB_UNKNOWN, JOB_PENDING, JOB_COMPLETE, JOB_FAILED]
+        for bad_state in invalid_states:
+            current_request.state = bad_state
+
+            # All these commands expect to be in JOB_CREATED
+            self.assertRaises(InvalidClientState, self.command_handler.recv_command, GEARMAN_COMMAND_WORK_DATA, job_handle=job_handle, data=new_data)
+
+            self.assertRaises(InvalidClientState, self.command_handler.recv_command, GEARMAN_COMMAND_WORK_WARNING, job_handle=job_handle, data=new_data)
+
+            self.assertRaises(InvalidClientState, self.command_handler.recv_command, GEARMAN_COMMAND_WORK_STATUS, job_handle=job_handle, numerator=0, denominator=1)
+
+            self.assertRaises(InvalidClientState, self.command_handler.recv_command, GEARMAN_COMMAND_WORK_COMPLETE, job_handle=job_handle, data=new_data)
+
+            self.assertRaises(InvalidClientState, self.command_handler.recv_command, GEARMAN_COMMAND_WORK_FAIL, job_handle=job_handle)
+
+    def test_in_flight_work_updates(self):
+        current_request = self.generate_job_request()
+
+        job_handle = current_request.job.handle
+        new_data = str(random.random())
+
+        # Test WORK_DATA
+        self.command_handler.recv_command(GEARMAN_COMMAND_WORK_DATA, job_handle=job_handle, data=new_data)
+        self.assertEqual(current_request.data_updates.popleft(), new_data)
+        self.assertEqual(current_request.state, JOB_CREATED)
+
+        # Test WORK_WARNING
+        self.command_handler.recv_command(GEARMAN_COMMAND_WORK_WARNING, job_handle=job_handle, data=new_data)
+        self.assertEqual(current_request.warning_updates.popleft(), new_data)
+        self.assertEqual(current_request.state, JOB_CREATED)
+
+        # Test WORK_STATUS
+        self.command_handler.recv_command(GEARMAN_COMMAND_WORK_STATUS, job_handle=job_handle, numerator=0, denominator=1)
+
+        self.assertEqual(current_request.status_updates.popleft(), (0, 1))
+        self.assertEqual(current_request.state, JOB_CREATED)
+
+    def test_work_complete(self):
+        current_request = self.generate_job_request()
+
+        job_handle = current_request.job.handle
+        new_data = str(random.random())
+        self.command_handler.recv_command(GEARMAN_COMMAND_WORK_COMPLETE, job_handle=job_handle, data=new_data)
+
+        self.assertEqual(current_request.result, new_data)
+        self.assertEqual(current_request.state, JOB_COMPLETE)
+
+    def test_work_fail(self):
+        current_request = self.generate_job_request()
+
+        job_handle = current_request.job.handle
+        new_data = str(random.random())
+        self.command_handler.recv_command(GEARMAN_COMMAND_WORK_FAIL, job_handle=job_handle)
+
+        self.assertEqual(current_request.state, JOB_FAILED)
+
+    def test_status_request(self):
+        current_request = self.generate_job_request()
+
+        job_handle = current_request.job.handle
+
+        self.assertEqual(current_request.status, {})
+
+        self.command_handler.recv_command(GEARMAN_COMMAND_STATUS_RES, job_handle=job_handle, known='1', running='1', numerator='0', denominator='1')
+
+        self.assertEqual(current_request.status['handle'], job_handle)
+        self.assertTrue(current_request.status['known'])
+        self.assertTrue(current_request.status['running'])
+        self.assertEqual(current_request.status['numerator'], 0)
+        self.assertEqual(current_request.status['denominator'], 1)
+
+if __name__ == '__main__':
+    unittest.main()
Index: python-gearman-2.0.2/tests/protocol_tests.py
===================================================================
--- /dev/null	1970-01-01 00:00:00.000000000 +0000
+++ python-gearman-2.0.2/tests/protocol_tests.py	2011-05-06 22:21:12.297469268 +0200
@@ -0,0 +1,275 @@
+import struct
+import unittest
+
+from gearman import protocol
+
+from gearman.connection import GearmanConnection
+from gearman.constants import JOB_PENDING, JOB_CREATED, JOB_FAILED, JOB_COMPLETE
+from gearman.errors import ConnectionError, ServerUnavailable, ProtocolError
+
+from tests._core_testing import _GearmanAbstractTest
+
+class ProtocolBinaryCommandsTest(unittest.TestCase):
+    #######################
+    # Begin parsing tests #
+    #######################
+    def test_parsing_errors(self):
+        malformed_command_buffer = "%sAAAABBBBCCCC"
+
+        # Raise malformed magic exceptions
+        self.assertRaises(ProtocolError, protocol.parse_binary_command, malformed_command_buffer % "DDDD")
+        self.assertRaises(ProtocolError, protocol.parse_binary_command, malformed_command_buffer % protocol.MAGIC_RES_STRING, is_response=False)
+        self.assertRaises(ProtocolError, protocol.parse_binary_command, malformed_command_buffer % protocol.MAGIC_REQ_STRING, is_response=True)
+
+        # Raise unknown command errors
+        unassigned_gearman_command = 1234
+        unknown_command_buffer = struct.pack('!4sII', protocol.MAGIC_RES_STRING, unassigned_gearman_command, 0)
+        self.assertRaises(ProtocolError, protocol.parse_binary_command, unknown_command_buffer)
+
+        # Raise an error on our imaginary GEARMAN_COMMAND_TEXT_COMMAND
+        imaginary_command_buffer = struct.pack('!4sII4s', protocol.MAGIC_RES_STRING, protocol.GEARMAN_COMMAND_TEXT_COMMAND, 4, 'ABCD')
+        self.assertRaises(ProtocolError, protocol.parse_binary_command, imaginary_command_buffer)
+
+        # Raise an error on receiving an unexpected payload
+        unexpected_payload_command_buffer = struct.pack('!4sII4s', protocol.MAGIC_RES_STRING, protocol.GEARMAN_COMMAND_NOOP, 4, 'ABCD')
+        self.assertRaises(ProtocolError, protocol.parse_binary_command, unexpected_payload_command_buffer)
+
+    def test_parsing_request(self):
+        # Test parsing a request for a job (server side parsing)
+        grab_job_command_buffer = struct.pack('!4sII', protocol.MAGIC_REQ_STRING, protocol.GEARMAN_COMMAND_GRAB_JOB_UNIQ, 0)
+        cmd_type, cmd_args, cmd_len = protocol.parse_binary_command(grab_job_command_buffer, is_response=False)
+        self.assertEquals(cmd_type, protocol.GEARMAN_COMMAND_GRAB_JOB_UNIQ)
+        self.assertEquals(cmd_args, dict())
+        self.assertEquals(cmd_len, len(grab_job_command_buffer))
+
+    def test_parsing_without_enough_data(self):
+        # Test that we return with nothing to do... received a partial packet
+        not_enough_data_command_buffer = struct.pack('!4s', protocol.MAGIC_RES_STRING)
+        cmd_type, cmd_args, cmd_len = protocol.parse_binary_command(not_enough_data_command_buffer)
+        self.assertEquals(cmd_type, None)
+        self.assertEquals(cmd_args, None)
+        self.assertEquals(cmd_len, 0)
+
+        # Test that we return with nothing to do... received a partial packet (expected binary payload of size 4, got 0)
+        not_enough_data_command_buffer = struct.pack('!4sII', protocol.MAGIC_RES_STRING, protocol.GEARMAN_COMMAND_ECHO_RES, 4)
+        cmd_type, cmd_args, cmd_len = protocol.parse_binary_command(not_enough_data_command_buffer)
+        self.assertEquals(cmd_type, None)
+        self.assertEquals(cmd_args, None)
+        self.assertEquals(cmd_len, 0)
+
+    def test_parsing_no_args(self):
+        noop_command_buffer = struct.pack('!4sII', protocol.MAGIC_RES_STRING, protocol.GEARMAN_COMMAND_NOOP, 0)
+        cmd_type, cmd_args, cmd_len = protocol.parse_binary_command(noop_command_buffer)
+        self.assertEquals(cmd_type, protocol.GEARMAN_COMMAND_NOOP)
+        self.assertEquals(cmd_args, dict())
+        self.assertEquals(cmd_len, len(noop_command_buffer))
+
+    def test_parsing_single_arg(self):
+        echoed_string = 'abcd'
+        echo_command_buffer = struct.pack('!4sII4s', protocol.MAGIC_RES_STRING, protocol.GEARMAN_COMMAND_ECHO_RES, 4, echoed_string)
+        cmd_type, cmd_args, cmd_len = protocol.parse_binary_command(echo_command_buffer)
+        self.assertEquals(cmd_type, protocol.GEARMAN_COMMAND_ECHO_RES)
+        self.assertEquals(cmd_args, dict(data=echoed_string))
+        self.assertEquals(cmd_len, len(echo_command_buffer))
+
+    def test_parsing_single_arg_with_extra_data(self):
+        echoed_string = 'abcd'
+        excess_bytes = 5
+        excess_data = echoed_string + (protocol.NULL_CHAR * excess_bytes)
+        excess_echo_command_buffer = struct.pack('!4sII9s', protocol.MAGIC_RES_STRING, protocol.GEARMAN_COMMAND_ECHO_RES, 4, excess_data)
+        cmd_type, cmd_args, cmd_len = protocol.parse_binary_command(excess_echo_command_buffer)
+        self.assertEquals(cmd_type, protocol.GEARMAN_COMMAND_ECHO_RES)
+        self.assertEquals(cmd_args, dict(data=echoed_string))
+        self.assertEquals(cmd_len, len(excess_echo_command_buffer) - excess_bytes)
+
+    def test_parsing_multiple_args(self):
+        # Tests ordered argument processing and proper NULL_CHAR splitting
+        expected_data = protocol.NULL_CHAR * 4
+        binary_payload = protocol.NULL_CHAR.join(['test', 'function', 'identifier', expected_data])
+        payload_size = len(binary_payload)
+
+        uniq_command_buffer = struct.pack('!4sII%ds' % payload_size, protocol.MAGIC_RES_STRING, protocol.GEARMAN_COMMAND_JOB_ASSIGN_UNIQ, payload_size, binary_payload)
+        cmd_type, cmd_args, cmd_len = protocol.parse_binary_command(uniq_command_buffer)
+        self.assertEquals(cmd_type, protocol.GEARMAN_COMMAND_JOB_ASSIGN_UNIQ)
+        self.assertEquals(cmd_args, dict(job_handle='test', task='function', unique='identifier', data=expected_data))
+        self.assertEquals(cmd_len, len(uniq_command_buffer))
+
+    #######################
+    # Begin packing tests #
+    #######################
+    def test_packing_errors(self):
+        # Assert we get an unknown command
+        cmd_type = 1234
+        cmd_args = dict()
+        self.assertRaises(ProtocolError, protocol.pack_binary_command, cmd_type, cmd_args)
+
+        # Assert we get a fake command
+        cmd_type = protocol.GEARMAN_COMMAND_TEXT_COMMAND
+        cmd_args = dict()
+        self.assertRaises(ProtocolError, protocol.pack_binary_command, cmd_type, cmd_args)
+
+        # Assert we get arg mismatch, got 1, expecting 0
+        cmd_type = protocol.GEARMAN_COMMAND_GRAB_JOB
+        cmd_args = dict(extra='arguments')
+        self.assertRaises(ProtocolError, protocol.pack_binary_command, cmd_type, cmd_args)
+
+        # Assert we get arg mismatch, got 0, expecting 1
+        cmd_type = protocol.GEARMAN_COMMAND_JOB_CREATED
+        cmd_args = dict()
+        self.assertRaises(ProtocolError, protocol.pack_binary_command, cmd_type, cmd_args)
+
+        # Assert we get arg mismatch (name), got 1, expecting 1
+        cmd_type = protocol.GEARMAN_COMMAND_JOB_CREATED
+        cmd_args = dict(extra='arguments')
+        self.assertRaises(ProtocolError, protocol.pack_binary_command, cmd_type, cmd_args)
+
+        # Assert we get a non-string argument
+        cmd_type = protocol.GEARMAN_COMMAND_JOB_CREATED
+        cmd_args = dict(job_handle=12345)
+        self.assertRaises(ProtocolError, protocol.pack_binary_command, cmd_type, cmd_args)
+
+        # Assert we get a non-string argument (expecting BYTES)
+        cmd_type = protocol.GEARMAN_COMMAND_JOB_CREATED
+        cmd_args = dict(job_handle=unicode(12345))
+        self.assertRaises(ProtocolError, protocol.pack_binary_command, cmd_type, cmd_args)
+
+    def test_packing_response(self):
+        # Test packing a response for a job (server side packing)
+        cmd_type = protocol.GEARMAN_COMMAND_NO_JOB
+        cmd_args = dict()
+
+        expected_command_buffer = struct.pack('!4sII', protocol.MAGIC_RES_STRING, cmd_type, 0)
+        packed_command_buffer = protocol.pack_binary_command(cmd_type, cmd_args, is_response=True)
+        self.assertEquals(packed_command_buffer, expected_command_buffer)
+
+    def test_packing_no_arg(self):
+        cmd_type = protocol.GEARMAN_COMMAND_NOOP
+        cmd_args = dict()
+
+        expected_command_buffer = struct.pack('!4sII', protocol.MAGIC_REQ_STRING, cmd_type, 0)
+        packed_command_buffer = protocol.pack_binary_command(cmd_type, cmd_args)
+        self.assertEquals(packed_command_buffer, expected_command_buffer)
+
+    def test_packing_single_arg(self):
+        cmd_type = protocol.GEARMAN_COMMAND_ECHO_REQ
+        cmd_args = dict(data='abcde')
+
+        expected_payload_size = len(cmd_args['data'])
+        expected_format = '!4sII%ds' % expected_payload_size
+
+        expected_command_buffer = struct.pack(expected_format, protocol.MAGIC_REQ_STRING, cmd_type, expected_payload_size, cmd_args['data'])
+        packed_command_buffer = protocol.pack_binary_command(cmd_type, cmd_args)
+        self.assertEquals(packed_command_buffer, expected_command_buffer)
+
+    def test_packing_multiple_args(self):
+        cmd_type = protocol.GEARMAN_COMMAND_SUBMIT_JOB
+        cmd_args = dict(task='function', unique='12345', data='abcd')
+
+        ordered_parameters = [cmd_args['task'], cmd_args['unique'], cmd_args['data']]
+
+        expected_payload = protocol.NULL_CHAR.join(ordered_parameters)
+        expected_payload_size = len(expected_payload)
+        expected_format = '!4sII%ds' % expected_payload_size
+        expected_command_buffer = struct.pack(expected_format, protocol.MAGIC_REQ_STRING, cmd_type, expected_payload_size, expected_payload)
+
+        packed_command_buffer = protocol.pack_binary_command(cmd_type, cmd_args)
+        self.assertEquals(packed_command_buffer, expected_command_buffer)
+
+class ProtocolTextCommandsTest(unittest.TestCase):
+	#######################
+    # Begin parsing tests #
+    #######################
+    def test_parsing_errors(self):
+        received_data = "Hello\x00there\n"
+        self.assertRaises(ProtocolError, protocol.parse_text_command, received_data)
+
+    def test_parsing_without_enough_data(self):
+        received_data = "Hello there"
+        cmd_type, cmd_response, cmd_len = protocol.parse_text_command(received_data)
+        self.assertEquals(cmd_type, None)
+        self.assertEquals(cmd_response, None)
+        self.assertEquals(cmd_len, 0)
+
+    def test_parsing_single_line(self):
+        received_data = "Hello there\n"
+        cmd_type, cmd_response, cmd_len = protocol.parse_text_command(received_data)
+        self.assertEquals(cmd_type, protocol.GEARMAN_COMMAND_TEXT_COMMAND)
+        self.assertEquals(cmd_response, dict(raw_text=received_data.strip()))
+        self.assertEquals(cmd_len, len(received_data))
+
+    def test_parsing_multi_line(self):
+        sentence_one = "Hello there\n"
+        sentence_two = "My name is bob\n"
+        received_data = sentence_one + sentence_two
+
+        cmd_type, cmd_response, cmd_len = protocol.parse_text_command(received_data)
+        self.assertEquals(cmd_type, protocol.GEARMAN_COMMAND_TEXT_COMMAND)
+        self.assertEquals(cmd_response, dict(raw_text=sentence_one.strip()))
+        self.assertEquals(cmd_len, len(sentence_one))
+
+    def test_packing_errors(self):
+        # Test bad command type
+        cmd_type = protocol.GEARMAN_COMMAND_NOOP
+        cmd_args = dict()
+        self.assertRaises(ProtocolError, protocol.pack_text_command, cmd_type, cmd_args)
+
+        # Test missing args
+        cmd_type = protocol.GEARMAN_COMMAND_TEXT_COMMAND
+        cmd_args = dict()
+        self.assertRaises(ProtocolError, protocol.pack_text_command, cmd_type, cmd_args)
+
+        # Test misnamed parameter dict
+        cmd_type = protocol.GEARMAN_COMMAND_TEXT_COMMAND
+        cmd_args = dict(bad_text='abcdefghij')
+        self.assertRaises(ProtocolError, protocol.pack_text_command, cmd_type, cmd_args)
+
+    #######################
+    # Begin packing tests #
+    #######################
+    def test_packing_single_line(self):
+        expected_string = 'Hello world'
+        cmd_type = protocol.GEARMAN_COMMAND_TEXT_COMMAND
+        cmd_args = dict(raw_text=expected_string)
+
+        packed_command = protocol.pack_text_command(cmd_type, cmd_args)
+        self.assertEquals(packed_command, expected_string)
+
+class GearmanConnectionTest(unittest.TestCase):
+    """Tests the base CommandHandler class that underpins all other CommandHandlerTests"""
+    def test_recv_command(self):
+        pass
+
+class GearmanCommandHandlerTest(_GearmanAbstractTest):
+    """Tests the base CommandHandler class that underpins all other CommandHandlerTests"""
+    def _test_recv_command(self):
+        # recv_echo_res and recv_error are predefined on the CommandHandler
+        self.command_handler.recv_command(protocol.GEARMAN_COMMAND_NOOP)
+        self.assert_recv_command(protocol.GEARMAN_COMMAND_NOOP)
+
+        # The mock handler never implemented 'recv_all_yours' so we should get an attribute error here
+        self.assertRaises(ValueError, self.command_handler.recv_command, protocol.GEARMAN_COMMAND_ALL_YOURS)
+
+    def _test_send_command(self):
+        self.command_handler.send_command(protocol.GEARMAN_COMMAND_NOOP)
+        self.assert_sent_command(protocol.GEARMAN_COMMAND_NOOP)
+
+        # The mock handler never implemented 'recv_all_yours' so we should get an attribute error here
+        self.command_handler.send_command(protocol.GEARMAN_COMMAND_ECHO_REQ, text='hello world')
+        self.assert_sent_command(protocol.GEARMAN_COMMAND_ECHO_REQ, text='hello world')
+
+    def assert_recv_command(self, expected_cmd_type, **expected_cmd_args):
+        cmd_type, cmd_args = self.command_handler.recv_command_queue.popleft()
+        self.assert_commands_equal(cmd_type, expected_cmd_type)
+        self.assertEqual(cmd_args, expected_cmd_args)
+
+    def assert_sent_command(self, expected_cmd_type, **expected_cmd_args):
+        # All commands should be sent via the CommandHandler
+        handler_cmd_type, handler_cmd_args = self.command_handler.sent_command_queue.popleft()
+        self.assert_commands_equal(handler_cmd_type, expected_cmd_type)
+        self.assertEqual(handler_cmd_args, expected_cmd_args)
+
+        super(GearmanCommandHandlerTest, self).assert_sent_command(expected_cmd_type, **expected_cmd_args)
+
+
+if __name__ == '__main__':
+    unittest.main()
\ No newline at end of file
Index: python-gearman-2.0.2/tests/worker_tests.py
===================================================================
--- /dev/null	1970-01-01 00:00:00.000000000 +0000
+++ python-gearman-2.0.2/tests/worker_tests.py	2011-05-06 22:21:12.301468889 +0200
@@ -0,0 +1,365 @@
+import collections
+from gearman import compat
+import unittest
+
+from gearman.worker import GearmanWorker
+from gearman.worker_handler import GearmanWorkerCommandHandler
+
+from gearman.errors import ServerUnavailable, InvalidWorkerState
+from gearman.protocol import get_command_name, GEARMAN_COMMAND_RESET_ABILITIES, GEARMAN_COMMAND_CAN_DO, GEARMAN_COMMAND_SET_CLIENT_ID, \
+    GEARMAN_COMMAND_NOOP, GEARMAN_COMMAND_PRE_SLEEP, GEARMAN_COMMAND_NO_JOB, GEARMAN_COMMAND_GRAB_JOB_UNIQ, GEARMAN_COMMAND_JOB_ASSIGN_UNIQ, \
+    GEARMAN_COMMAND_WORK_STATUS, GEARMAN_COMMAND_WORK_FAIL, GEARMAN_COMMAND_WORK_COMPLETE, GEARMAN_COMMAND_WORK_DATA, GEARMAN_COMMAND_WORK_EXCEPTION, GEARMAN_COMMAND_WORK_WARNING
+
+from tests._core_testing import _GearmanAbstractTest, MockGearmanConnectionManager, MockGearmanConnection
+
+class MockGearmanWorker(MockGearmanConnectionManager, GearmanWorker):
+    def __init__(self, *largs, **kwargs):
+        super(MockGearmanWorker, self).__init__(*largs, **kwargs)
+        self.worker_job_queues = compat.defaultdict(collections.deque)
+
+    def on_job_execute(self, current_job):
+        current_handler = self.connection_to_handler_map[current_job.connection]
+        self.worker_job_queues[current_handler].append(current_job)
+
+class _GearmanAbstractWorkerTest(_GearmanAbstractTest):
+    connection_manager_class = MockGearmanWorker
+    command_handler_class = GearmanWorkerCommandHandler
+
+    def setup_command_handler(self):
+        super(_GearmanAbstractWorkerTest, self).setup_command_handler()
+        self.assert_sent_abilities([])
+        self.assert_sent_command(GEARMAN_COMMAND_PRE_SLEEP)
+
+    def assert_sent_abilities(self, expected_abilities):
+        observed_abilities = set()
+
+        self.assert_sent_command(GEARMAN_COMMAND_RESET_ABILITIES)
+        for ability in expected_abilities:
+            cmd_type, cmd_args = self.connection._outgoing_commands.popleft()
+
+            self.assertEqual(get_command_name(cmd_type), get_command_name(GEARMAN_COMMAND_CAN_DO))
+            observed_abilities.add(cmd_args['task'])
+
+        self.assertEqual(observed_abilities, set(expected_abilities))
+
+    def assert_sent_client_id(self, expected_client_id):
+        self.assert_sent_command(GEARMAN_COMMAND_SET_CLIENT_ID, client_id=expected_client_id)
+
+class WorkerTest(_GearmanAbstractWorkerTest):
+    """Test the public worker interface"""
+    def test_registering_functions(self):
+        # Tests that the abilities were set on the GearmanWorker AND the GearmanWorkerCommandHandler
+        # Does NOT test that commands were actually sent out as that is tested in GearmanWorkerCommandHandlerInterfaceTest.test_set_abilities
+        def fake_callback_one(worker_command_handler, current_job):
+            pass
+
+        def fake_callback_two(worker_command_handler, current_job):
+            pass
+
+        # Register a single callback
+        self.connection_manager.register_task('fake_callback_one', fake_callback_one)
+        self.failUnless('fake_callback_one' in self.connection_manager.worker_abilities)
+        self.failIf('fake_callback_two' in self.connection_manager.worker_abilities)
+        self.assertEqual(self.connection_manager.worker_abilities['fake_callback_one'], fake_callback_one)
+        self.assertEqual(self.command_handler._handler_abilities, ['fake_callback_one'])
+
+        # Register another callback and make sure the command_handler sees the same functions
+        self.connection_manager.register_task('fake_callback_two', fake_callback_two)
+        self.failUnless('fake_callback_one' in self.connection_manager.worker_abilities)
+        self.failUnless('fake_callback_two' in self.connection_manager.worker_abilities)
+        self.assertEqual(self.connection_manager.worker_abilities['fake_callback_one'], fake_callback_one)
+        self.assertEqual(self.connection_manager.worker_abilities['fake_callback_two'], fake_callback_two)
+        self.assertEqual(self.command_handler._handler_abilities, ['fake_callback_one', 'fake_callback_two'])
+
+        # Unregister a callback and make sure the command_handler sees the same functions
+        self.connection_manager.unregister_task('fake_callback_one')
+        self.failIf('fake_callback_one' in self.connection_manager.worker_abilities)
+        self.failUnless('fake_callback_two' in self.connection_manager.worker_abilities)
+        self.assertEqual(self.connection_manager.worker_abilities['fake_callback_two'], fake_callback_two)
+        self.assertEqual(self.command_handler._handler_abilities, ['fake_callback_two'])
+
+    def test_setting_client_id(self):
+        new_client_id = 'HELLO'
+
+        # Make sure nothing is set
+        self.assertEqual(self.connection_manager.worker_client_id, None)
+        self.assertEqual(self.command_handler._client_id, None)
+
+        self.connection_manager.set_client_id(new_client_id)
+
+        # Make sure both the client and the connection handler reflect the new state
+        self.assertEqual(self.connection_manager.worker_client_id, new_client_id)
+        self.assertEqual(self.command_handler._client_id, new_client_id)
+
+    def test_establish_worker_connections(self):
+        self.connection_manager.connection_list = []
+        self.connection_manager.command_handlers = {}
+
+        # Spin up a bunch of imaginary gearman connections
+        good_connection = MockGearmanConnection()
+        good_connection.connect()
+        good_connection._fail_on_bind = False
+
+        failed_then_retried_connection = MockGearmanConnection()
+        failed_then_retried_connection._fail_on_bind = False
+
+        failed_connection = MockGearmanConnection()
+        failed_connection._fail_on_bind = True
+
+        # Register all our connections
+        self.connection_manager.connection_list = [good_connection, failed_then_retried_connection, failed_connection]
+
+        # The only alive connections should be the ones that ultimately be connection.connected
+        alive_connections = self.connection_manager.establish_worker_connections()
+        self.assertTrue(good_connection in alive_connections)
+        self.assertTrue(failed_then_retried_connection in alive_connections)
+        self.assertFalse(failed_connection in alive_connections)
+
+    def test_establish_worker_connections_dead(self):
+        self.connection_manager.connection_list = []
+        self.connection_manager.command_handlers = {}
+
+        # We have no connections so there will never be any work to do
+        self.assertRaises(ServerUnavailable, self.connection_manager.work)
+
+        # We were started with a dead connection, make sure we bail again
+        dead_connection = MockGearmanConnection()
+        dead_connection._fail_on_bind = True
+        dead_connection.connected = False
+        self.connection_manager.connection_list = [dead_connection]
+
+        self.assertRaises(ServerUnavailable, self.connection_manager.work)
+
+
+class WorkerCommandHandlerInterfaceTest(_GearmanAbstractWorkerTest):
+    """Test the public interface a GearmanWorker may need to call in order to update state on a GearmanWorkerCommandHandler"""
+
+    def test_on_connect(self):
+        expected_abilities = ['function_one', 'function_two', 'function_three']
+        expected_client_id = 'my_client_id'
+
+        self.connection.connected = False
+
+        self.connection_manager.set_client_id(expected_client_id)
+        self.connection_manager.unregister_task('__test_ability__')
+        for task in expected_abilities:
+            self.connection_manager.register_task(task, None)
+
+        # We were disconnected, connect and wipe pending commands
+        self.connection_manager.establish_connection(self.connection)
+
+        # When we attempt a new connection, make sure we get a new command handler
+        self.assertNotEquals(self.command_handler, self.connection_manager.connection_to_handler_map[self.connection])
+
+        self.assert_sent_client_id(expected_client_id)
+        self.assert_sent_abilities(expected_abilities)
+        self.assert_sent_command(GEARMAN_COMMAND_PRE_SLEEP)
+        self.assert_no_pending_commands()
+
+    def test_set_abilities(self):
+        expected_abilities = ['function_one', 'function_two', 'function_three']
+
+        # We were disconnected, connect and wipe pending commands
+        self.command_handler.set_abilities(expected_abilities)
+        self.assert_sent_abilities(expected_abilities)
+        self.assert_no_pending_commands()
+
+    def test_set_client_id(self):
+        expected_client_id = 'my_client_id'
+
+        handler_initial_state = {}
+        handler_initial_state['abilities'] = []
+        handler_initial_state['client_id'] = None
+
+        # We were disconnected, connect and wipe pending commands
+        self.command_handler.set_client_id(expected_client_id)
+        self.assert_sent_client_id(expected_client_id)
+        self.assert_no_pending_commands()
+
+    def test_send_functions(self):
+        current_job = self.generate_job()
+
+        # Test GEARMAN_COMMAND_WORK_STATUS
+        self.command_handler.send_job_status(current_job, 0, 1)
+        self.assert_sent_command(GEARMAN_COMMAND_WORK_STATUS, job_handle=current_job.handle, numerator='0', denominator='1')
+
+        # Test GEARMAN_COMMAND_WORK_COMPLETE
+        self.command_handler.send_job_complete(current_job, 'completion data')
+        self.assert_sent_command(GEARMAN_COMMAND_WORK_COMPLETE, job_handle=current_job.handle, data='completion data')
+
+        # Test GEARMAN_COMMAND_WORK_FAIL
+        self.command_handler.send_job_failure(current_job)
+        self.assert_sent_command(GEARMAN_COMMAND_WORK_FAIL, job_handle=current_job.handle)
+
+        # Test GEARMAN_COMMAND_WORK_EXCEPTION
+        self.command_handler.send_job_exception(current_job, 'exception data')
+        self.assert_sent_command(GEARMAN_COMMAND_WORK_EXCEPTION, job_handle=current_job.handle, data='exception data')
+
+        # Test GEARMAN_COMMAND_WORK_DATA
+        self.command_handler.send_job_data(current_job, 'job data')
+        self.assert_sent_command(GEARMAN_COMMAND_WORK_DATA, job_handle=current_job.handle, data='job data')
+
+        # Test GEARMAN_COMMAND_WORK_WARNING
+        self.command_handler.send_job_warning(current_job, 'job warning')
+        self.assert_sent_command(GEARMAN_COMMAND_WORK_WARNING, job_handle=current_job.handle, data='job warning')
+
+class WorkerCommandHandlerStateMachineTest(_GearmanAbstractWorkerTest):
+    """Test multiple state transitions within a GearmanWorkerCommandHandler
+
+    End to end tests without a server
+    """
+    connection_manager_class = MockGearmanWorker
+    command_handler_class = GearmanWorkerCommandHandler
+
+    def setup_connection_manager(self):
+        super(WorkerCommandHandlerStateMachineTest, self).setup_connection_manager()
+        self.connection_manager.register_task('__test_ability__', None)
+
+    def setup_command_handler(self):
+        super(_GearmanAbstractWorkerTest, self).setup_command_handler()
+        self.assert_sent_abilities(['__test_ability__'])
+        self.assert_sent_command(GEARMAN_COMMAND_PRE_SLEEP)
+
+    def test_wakeup_work(self):
+        self.move_to_state_wakeup()
+
+        self.move_to_state_job_assign_uniq(self.generate_job_dict())
+
+        self.move_to_state_wakeup()
+
+        self.move_to_state_no_job()
+
+    def test_wakeup_sleep_wakup_work(self):
+        self.move_to_state_wakeup()
+
+        self.move_to_state_no_job()
+
+        self.move_to_state_wakeup()
+
+        self.move_to_state_job_assign_uniq(self.generate_job_dict())
+
+        self.move_to_state_wakeup()
+
+        self.move_to_state_no_job()
+
+    def test_multiple_wakeup_then_no_work(self):
+        # Awaken the state machine... then give it no work
+        self.move_to_state_wakeup()
+
+        for _ in range(5):
+            self.command_handler.recv_command(GEARMAN_COMMAND_NOOP)
+
+        self.assert_job_lock(is_locked=True)
+
+        # Pretend like the server has no work... do nothing
+        # Moving to state NO_JOB will make sure there's only 1 item on the queue
+        self.move_to_state_no_job()
+
+    def test_multiple_work(self):
+        self.move_to_state_wakeup()
+
+        self.move_to_state_job_assign_uniq(self.generate_job_dict())
+
+        self.move_to_state_wakeup()
+
+        self.move_to_state_job_assign_uniq(self.generate_job_dict())
+
+        self.move_to_state_wakeup()
+
+        self.move_to_state_job_assign_uniq(self.generate_job_dict())
+
+        self.move_to_state_wakeup()
+
+        # After this job completes, we're going to greedily ask for more jobs
+        self.move_to_state_no_job()
+
+    def test_worker_already_locked(self):
+        other_connection = MockGearmanConnection()
+        self.connection_manager.connection_list.append(other_connection)
+        self.connection_manager.establish_connection(other_connection)
+
+        other_handler = self.connection_manager.connection_to_handler_map[other_connection]
+        other_handler.recv_command(GEARMAN_COMMAND_NOOP)
+
+        # Make sure other handler has a lock
+        self.assertEqual(self.connection_manager.command_handler_holding_job_lock, other_handler)
+
+        # Make sure OUR handler has nothing incoming
+        self.assert_no_pending_commands()
+
+        # Make sure we try to grab a job but fail...so go back to sleep
+        self.command_handler.recv_command(GEARMAN_COMMAND_NOOP)
+        self.assert_sent_command(GEARMAN_COMMAND_PRE_SLEEP)
+
+        # Make sure other handler still has lock
+        self.assertEqual(self.connection_manager.command_handler_holding_job_lock, other_handler)
+
+        # Make the other handler release its lock
+        other_handler.recv_command(GEARMAN_COMMAND_NO_JOB)
+
+        # Ensure that the lock has been freed
+        self.assert_job_lock(is_locked=False)
+
+        # Try to do work after we have our lock released
+        self.move_to_state_wakeup()
+
+        self.move_to_state_job_assign_uniq(self.generate_job_dict())
+
+        self.move_to_state_wakeup()
+
+        self.move_to_state_no_job()
+
+    def move_to_state_wakeup(self):
+        self.assert_no_pending_commands()
+        self.assert_job_lock(is_locked=False)
+
+        self.command_handler.recv_command(GEARMAN_COMMAND_NOOP)
+
+    def move_to_state_no_job(self):
+        """Move us to the NO_JOB state...
+
+        1) We should've most recently sent only a single GEARMAN_COMMAND_GRAB_JOB_UNIQ
+        2) We should be awaiting job assignment
+        3) Once we receive a NO_JOB, we should say we're going back to sleep"""
+        self.assert_awaiting_job()
+
+        self.command_handler.recv_command(GEARMAN_COMMAND_NO_JOB)
+
+        # We should be asleep... which means no pending jobs and we're not awaiting job assignment
+        self.assert_sent_command(GEARMAN_COMMAND_PRE_SLEEP)
+        self.assert_no_pending_commands()
+        self.assert_job_lock(is_locked=False)
+
+    def move_to_state_job_assign_uniq(self, fake_job):
+        """Move us to the JOB_ASSIGN_UNIQ state...
+
+        1) We should've most recently sent only a single GEARMAN_COMMAND_GRAB_JOB_UNIQ
+        2) We should be awaiting job assignment
+        3) The job we receive should be the one we expected"""
+        self.assert_awaiting_job()
+
+        ### NOTE: This recv_command does NOT send out a GEARMAN_COMMAND_JOB_COMPLETE or GEARMAN_COMMAND_JOB_FAIL
+        ###           as we're using a MockGearmanConnectionManager with a method that only queues the job
+        self.command_handler.recv_command(GEARMAN_COMMAND_JOB_ASSIGN_UNIQ, **fake_job)
+
+        current_job = self.connection_manager.worker_job_queues[self.command_handler].popleft()
+        self.assertEqual(current_job.handle, fake_job['job_handle'])
+        self.assertEqual(current_job.task, fake_job['task'])
+        self.assertEqual(current_job.unique, fake_job['unique'])
+        self.assertEqual(current_job.data, fake_job['data'])
+
+        # At the end of recv_command(GEARMAN_COMMAND_JOB_ASSIGN_UNIQ)
+        self.assert_job_lock(is_locked=False)
+        self.assert_sent_command(GEARMAN_COMMAND_PRE_SLEEP)
+
+    def assert_awaiting_job(self):
+        self.assert_sent_command(GEARMAN_COMMAND_GRAB_JOB_UNIQ)
+        self.assert_no_pending_commands()
+
+    def assert_job_lock(self, is_locked):
+        expected_value = (is_locked and self.command_handler) or None
+        self.assertEqual(self.connection_manager.command_handler_holding_job_lock, expected_value)
+
+if __name__ == '__main__':
+    unittest.main()
+
Index: python-gearman-2.0.2/tests/__init__.py
===================================================================
--- /dev/null	1970-01-01 00:00:00.000000000 +0000
+++ python-gearman-2.0.2/tests/__init__.py	2011-05-06 22:21:37.939040391 +0200
@@ -0,0 +1 @@
+# quilt doesn't like empty files, so add a comment :)