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?

Roberson

Answer 1

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.

Bedad

You can use self.get_tasks and task['decoded_body'] as a shortcut if you're using all the base test cases (github.com/jgeewax/gaetestbed/blob/master/gaetestbed/…). Additionally, this hopefully be a part of google.appengine.ext.testbed soon. There's a get_filtered_tasks method which works a lot like get_tasks (code.google.com/p/googleappengine/source/browse/trunk/python/…) – Hottempered

See my answer: this library is now deprecated in favor of ext.testbed (developers.google.com/appengine/docs/python/tools/…) – Hottempered

Answer 2

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(...)

Hottempered

Is it possible to do this with your own queues? I use deferred like: deferred.defer(get_and_store_all_new_messages_async, user_id, query, next_page_token, _queue="emailFetch") but it doesn't recognize the queue, probably because it didn't read the queue.yaml file for the unit test – Bonefish

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.

Torrance

This package is actually deprecated as I've merged the functionality into google.appengine.ext.testbed (developers.google.com/appengine/docs/python/tools/…) – Hottempered

Answer 4

The SDK 1.4.3 Testbed API provides easy configuration of stub libraries for local integration tests.

A Service stubs for Task Queue is available.

Levitation

This link is now broken. – Woodpile