GAE: unit testing taskqueue with testbed

Question

I'm using testbed to unit test my google app engine app, and my app uses a taskqueue.

When I submit a task to a taskqueue during a unit test, it appears that the task is in the queue, but the task does not execute.

How do I get the task to execute during a unit test?

Answer 1

Using Saxon's excellent answer, I was able to do the same thing using testbed instead of gaetestbed. Here is what I did.

Added this to my setUp():

    self.taskqueue_stub = apiproxy_stub_map.apiproxy.GetStub('taskqueue')

Then, in my test, I used the following:

    # Execute the task in the taskqueue
    tasks = self.taskqueue_stub.GetTasks("default")
    self.assertEqual(len(tasks), 1)
    task = tasks[0]
    params = base64.b64decode(task["body"])
    response = self.app.post(task["url"], params)

Somewhere along the line, the POST parameters get base64 encoded so had to undo that to get it to work.

I like this better than Saxon's answer since I can use the official testbed package and I can do it all within my own test code.

EDIT: I later wanted to do the same thing with tasks submitted using the deferred library, and it took a bit of headbanging to figure it, so I'm sharing here to ease other people's pain.

If your taskqueue contains only tasks submitted with deferred, then this will run all of the tasks and any tasks queued by those tasks:

def submit_deferred(taskq):
    tasks = taskq.GetTasks("default")
    taskq.FlushQueue("default")
    while tasks:
        for task in tasks:
            (func, args, opts) = pickle.loads(base64.b64decode(task["body"]))
            func(*args)
        tasks = taskq.GetTasks("default")
        taskq.FlushQueue("default")