Unit testing task queues in AppEngine

Question

For a very long time now I've been using task queues on AppEngine to schedule tasks, just the way I'm supposed to.

But what I've always been wondering is how does one write tests for that? Until now I've simply made tests to make sure an error doesn't occur on the API that queues a task and then wrote the more proper tests for the API executing the task.

However lately I've started feeling a bit unsatisfied by this and I'm searching for a way to actually test that the correct task has been added to the correct queue. Hopefully this can be done better than simply by deploying the code and hoping for the best.

I'm using django-nonrel, if that has any bearing on the answer.

To recap: How can a unit test be written to confirm tasks have been queued?

Answer 1

If you're using google.appengine.ext.testbed instead of GAE Testbed (GAE Testbed is now deprecated and being moved into ext.testbed), you can do the following:

import base64
import unittest2

from google.appengine.ext import deferred
from google.appengine.ext import testbed


class TestTasks(unittest2.TestCase):
  def setUp(self):
    self.testbed = testbed.Testbed()
    self.testbed.activate()
    self.testbed.init_taskqueue_stub()
    self.taskqueue_stub = self.testbed.get_stub(testbed.TASKQUEUE_SERVICE_NAME)

  def tearDown(self):
    self.testbed.deactivate()

  def test_send_contact_request(self):
    # Make the request to your app that "defers" something:
    response = ...
    self.assertEqual(response.status_int, 200)

    # Get the task out of the queue
    tasks = self.taskqueue_stub.get_filtered_tasks()
    self.assertEqual(1, len(tasks))

    # Run the task
    task = tasks[0]
    deferred.run(task.payload)

    # Assert that other things happened (ie, if the deferred was sending mail...)
    self.assertEqual(...)

Answer 2

Using GAE Test Bed will allow you to stub out a task queue.

If you inherit from FunctionalTestCase or TaskQueueTestCase, you'll get methods such as get_tasks and assertTasksInQueue.

You can actually run the tasks, too. How to do it differs depending on whether you use tasks or deferred.

For deferreds, I have some code like this:

from google.appengine.ext import deferred
import base64

# gets the most recent task -- since the task queue is reset between tests,
# this is usually what you want
def get_task(self):
    for task in self.get_task_queue_stub().GetTasks('default'):
        return task

# decode and execute the deferred method
def run_deferred(task):
    deferred.run(base64.b64decode(task['body']))

Running tasks is similar, but after you fetch the task, you use WebTest (which GAE Test Bed is built on top of) to submit as POST request to the task's URL with its parameters.

Answer 3

There is a little test framework called gaetestbed which may suit your need. For details please refer to: https://github.com/jgeewax/gaetestbed.

This testing environment works in connection with nose, nose-gae plugin and WebTest package. Given mix of python packages is the best way to test GAE applications as far as I'm concerned.