Running a Celery worker in unittest

Question

I have the following setup:

• Django-Celery project A registers task foo
• Project B: Uses Celery's send_task to call foo
• Project A and project B have the same configuration: SQS, msgpack for serialization, gzip, etc.
• Each project lives on a different github repository

I've unit-tested calls to "foo" in project A, without using Celery at all, just foo(1,2,3) and assert the result. I know that it works.

I've unit-tested that send_task in project B sends the right parameters.

What I'm not testing, and need your advise on is the integration between the two projects. I would like to have a unittest that would:

• Start a worker in the context of project A
• Send a task using the code of project B
• Assert that the worker started in the first step gets the task, with the parameters I sent in the second step, and that the foo function returned the expected result.

It seems to be possible to hack this by using python's subprocess and parsing the output of the worker, but that's ugly. What's the recommended approach to unit-testing in cases like this? Any code snippet you could share? Thanks!

Answer 1

I'm not sure if it's worthwhile to explicitly test the transportation mechanism (i.e. the sending of the task parameters through celery) using a unit test. Personally, I would write my test as follows (can be split up in several unit tests):

• Use the code from project B to generate a task with sample parameters.
• Encode the task parameters using the same method used by Celery (i.e. pickling the parameters or encoding them as JSON).
• Decode the task parameters again, checking that no corruption occured.
• Call the task function, making sure that it produces the correct result.
• Perform the same encoding/decoding sequence for the results of the task function.

Using this method, you will be able to test that

• The task generation works as intended
• The encoding & decoding of the task parameters and results works as expected

If necessary, you can still independently test the functioning of the transportation mechanism using a system test.