def test_no_entry_for_worker_does_not_raise_exception(self):
     self.mock_resources.filter_workers.return_value = []
     try:
         tasks._delete_worker('worker1')
     except Exception:
         self.fail('_delete_worker() on a Worker that is not in the database caused an '
                   'Exception')

 def test_no_entry_for_worker_does_not_raise_exception(self, mock_worker_objects):
     mock_worker_objects.get.return_value = []
     try:
         tasks._delete_worker('worker1')
     except Exception:
         self.fail('_delete_worker() on a Worker that is not in the database caused an '
                   'Exception')

def initialize_worker(sender, instance, **kwargs):
    """
    This function performs all the necessary initialization of the Celery worker.

    It starts by cleaning up old state if this worker was previously running, but died unexpectedly.
    In such cases, any Pulp tasks that were running or waiting on this worker will show incorrect
    state. Any reserved_resource reservations associated with the previous worker will also be
    removed along with the worker entry in the database itself. The working directory specified in
    /etc/pulp/server.conf (/var/cache/pulp/<worker_name>) by default is removed and recreated. This
    is called early in the worker start process, and later when it's fully online, pulp_celerybeat
    will discover the worker as usual to allow new work to arrive at this worker. If there is no
    previous work to cleanup, this method still runs, but has no effect on the database.

    After cleaning up old state, it ensures the existence of the worker's working directory.

    Lastly, this function makes the call to Pulp's initialization code.

    It uses the celeryd_after_setup signal[0] so that it gets called by Celery after logging is
    initialized, but before Celery starts to run tasks.

    [0] http://celery.readthedocs.org/en/latest/userguide/signals.html#celeryd-after-setup

    :param sender:   The hostname of the worker
    :type  sender:   basestring
    :param instance: The Worker instance to be initialized (unused)
    :type  instance: celery.apps.worker.Worker
    :param kwargs:   Other params (unused)
    :type  kwargs:   dict
    """
    initialization.initialize()

    tasks._delete_worker(sender, normal_shutdown=True)

    # Create a new working directory for worker that is starting now
    common_utils.create_worker_working_directory(sender)

    def test_cancels_all_found_task_status_objects(self):
        mock_task_id_a = mock.Mock()
        mock_task_id_b = mock.Mock()
        self.mock_task_status.objects.return_value = [{'task_id': mock_task_id_a},
                                                      {'task_id': mock_task_id_b}]
        tasks._delete_worker('worker1')

        self.mock_cancel.assert_has_calls([mock.call(mock_task_id_a), mock.call(mock_task_id_b)])

 def test__delete_worker_normal_shutdown_true(self, mock_logger, mock_underscore):
     """
     Call _delete_worker() with the normal_shutdown keyword argument set to True. This should
     not make any calls to _() or logger().
     """
     tasks._delete_worker('does not exist Worker name')
     self.assertTrue(not mock_underscore.called)
     self.assertTrue(not mock_logger.called)

    def test_removes_the_worker(self, mock_worker_objects):
        mock_document = mock.Mock()
        mock_get = mock.Mock()
        mock_get.get.return_value = [mock_document]
        mock_worker_objects.return_value = mock_get

        tasks._delete_worker('worker1')

        mock_document.delete.assert_called_once()

 def test__delete_worker_no_database_entry(self):
     """
     Call _delete_worker() with a Worker that is not in the database. _delete_worker() relies on
     the database information, so it should return without error when called in this way.
     """
     try:
         tasks._delete_worker('does not exist Worker name')
     except Exception:
         self.fail('_delete_worker() on a Worker that is not in the database caused an '
                   'Exception')

    def shutdown(self, consumer):
        """
        Called when the consumer is shut down.

        So far, this just cleans up the database by removing the worker's record in
        the workers collection.

        :param consumer: The consumer instance
        :type  consumer: celery.worker.consumer.Consumer
        """
        tasks._delete_worker(consumer.hostname, normal_shutdown=True)

    def test_cancels_all_found_task_status_objects(self):
        mock_task_id_a = mock.Mock()
        mock_task_id_b = mock.Mock()
        self.mock_task_status_manager.find_by_criteria.return_value = [{'task_id': mock_task_id_a},
                                                                       {'task_id': mock_task_id_b}]
        tasks._delete_worker('worker1')

        find_by_criteria = self.mock_task_status_manager.find_by_criteria
        find_by_criteria.assert_called_once_with(self.mock_criteria.return_value)

        self.mock_cancel.assert_has_calls([mock.call(mock_task_id_a), mock.call(mock_task_id_b)])

def shutdown_worker(signal, sender, **kwargs):
    """
    Called when a worker is shutdown.
    So far, this just cleans up the database by removing the worker's record in
    the workers collection.
    :param signal:   The signal being sent to the worker
    :type  signal:   int
    :param instance: The hostname of the worker
    :type  instance: celery.apps.worker.Worker
    :param kwargs:   Other params (unused)
    :type  kwargs:   dict
    """
    tasks._delete_worker(sender.hostname, normal_shutdown=True)

def initialize_worker(sender, instance, **kwargs):
    """
    This function performs all the necessary initialization of the Celery worker.

    We clean up old state in case this worker was previously running, but died unexpectedly.
    In such cases, any Pulp tasks that were running or waiting on this worker will show incorrect
    state. Any reserved_resource reservations associated with the previous worker will also be
    removed along with the worker entry in the database itself. The working directory specified in
    /etc/pulp/server.conf (/var/cache/pulp/<worker_name>) by default is removed and recreated. This
    is called early in the worker start process, and later when it's fully online, pulp_celerybeat
    will discover the worker as usual to allow new work to arrive at this worker. If there is no
    previous work to cleanup, this method still runs, but has no effect on the database.

    After cleaning up old state, it ensures the existence of the worker's working directory.

    Lastly, this function makes the call to Pulp's initialization code.

    It uses the celeryd_after_setup signal[0] so that it gets called by Celery after logging is
    initialized, but before Celery starts to run tasks.

    If the worker is a resource manager, it tries to acquire a lock stored within the database.
    If the lock cannot be acquired immediately, it will wait until the currently active instance
    becomes unavailable, at which point the worker cleanup routine will clear the lock for us to
    acquire. While the worker remains in this waiting state, it is not connected to the broker and
    will not attempt to do any work. A side effect of this is that, if terminated while in this
    state, the process will not send the "worker-offline" signal used by the EventMonitor to
    immediately clean up terminated workers. Therefore, we override the SIGTERM signal handler
    while in this state so that cleanup is done properly.

    [0] http://celery.readthedocs.org/en/latest/userguide/signals.html#celeryd-after-setup

    :param sender:   The hostname of the worker
    :type  sender:   basestring
    :param instance: The Worker instance to be initialized (unused)
    :type  instance: celery.apps.worker.Worker
    :param kwargs:   Other params (unused)
    :type  kwargs:   dict
    """
    initialization.initialize()

    # Delete any potential old state
    tasks._delete_worker(sender, normal_shutdown=True)

    # Create a new working directory for worker that is starting now
    common_utils.delete_worker_working_directory(sender)
    common_utils.create_worker_working_directory(sender)

    # If the worker is a resource manager, try to acquire the lock, or wait until it
    # can be acquired
    if sender.startswith(constants.RESOURCE_MANAGER_WORKER_NAME):
        get_resource_manager_lock(sender)

 def test_criteria_is_used_in_filter_workers(self):
     tasks._delete_worker('worker1')
     self.mock_resources.filter_workers.assert_called_once_with(self.mock_criteria.return_value)

 def test_removes_all_associated_reserved_resource_entries(self):
     tasks._delete_worker('worker1')
     self.assertTrue(self.mock_reserved_resource.get_collection.called)
     remove = self.mock_reserved_resource.get_collection.return_value.remove
     remove.assert_called_once_with({'worker_name': 'worker1'})

 def test_removes_the_worker(self):
     mock_worker = mock.Mock()
     self.mock_resources.filter_workers.return_value = tuple([mock_worker])
     tasks._delete_worker('worker1')
     mock_worker.delete.assert_called_once_with()

 def test_normal_shutdown_true_logs_correctly(self):
     tasks._delete_worker('worker1', normal_shutdown=True)
     self.assertTrue(not self.mock_gettext.called)
     self.assertTrue(not self.mock_logger.error.called)

 def test_normal_shutdown_not_specified_logs(self):
     self.mock_gettext.return_value = 'asdf %(name)s asdf'
     tasks._delete_worker('worker1')
     self.mock_gettext.assert_called_once_with(
         'The worker named %(name)s is missing. Canceling the tasks in its queue.')
     self.mock_logger.error.assert_called_once_with('asdf worker1 asdf')

 def test_removes_all_associated_reserved_resource_entries(self):
     tasks._delete_worker('worker1')
     self.assertTrue(self.mock_reserved_resource.objects.called)
     remove = self.mock_reserved_resource.objects.return_value.delete
     remove.assert_called_once_with()

 def test_criteria_to_find_task_status_is_correct(self):
     tasks._delete_worker('worker1')
     expected_call = mock.call(
         filters={'worker_name': self.mock_worker.from_bson.return_value.name,
                  'state': {'$in': self.mock_constants.CALL_INCOMPLETE_STATES}})
     self.assertEqual(self.mock_criteria.mock_calls[1], expected_call)

 def test_criteria_to_find_all_worker_is_correct(self):
     tasks._delete_worker('worker1')
     self.assertEqual(self.mock_criteria.mock_calls[0], mock.call(filters={'_id': 'worker1'}))

 def sigterm_handler(_signo, _stack_frame):
     msg = _("Worker '%s' shutdown" % name)
     _logger.info(msg)
     tasks._delete_worker(name, normal_shutdown=True)
     sys.exit(0)