Testing Celery tasks queued without executing the tasks
I am trying to write unit tests in a way that works nicely with Celery. I've read the documentation that suggests enabling CELERY_ALWAYS_EAGER. However, this isn't a good solution for me as always running the task is not a good idea. I want to test that tasks get queued, without actually running the tasks. This is because some tasks have external dependencies that are inappropriate for tests, but it is important to test the task was queued with correct arguments. For times that I do want to test the results of a task, running task.apply() seems to work well. To achieve this I've created a wrapper around Task that looks like:
enabled = True
tasks = 
LoggedTask = collections.namedtuple('LoggedTask', ['task', 'args', 'kwargs'])
abstract = True
def apply_async(self, args=None, kwargs=None, task_id=None, producer=None,
link=None, link_error=None, **options):
# Log for tests.
tasks.append(LoggedTask(self, args, kwargs))
app.Task = Task
This is inspired by Django's mail.outbox feature used in tests.
Is this a good approach? Is there a better or more preferred way to handle this?
You received this message because you are subscribed to the Google Groups "celery-users" group.
To unsubscribe from this group and stop receiving emails from it, send an email to celery-users+unsubscribe-/JYPxA39Uh5TLH3MbocFFw@public.gmane.org
To post to this group, send email to celery-users-/JYPxA39Uh5TLH3MbocFFw@public.gmane.org
Visit this group at http://groups.google.com/group/celery-users
For more options, visit https://groups.google.com/d/optout