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?
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")
Another (cleaner) option to achieve this is to use the task queue stub within the testbed. To do this you first have to initialize the task queue stub by adding the following to your setUp() method:
self.testbed = init_testbed()
self.testbed.init_taskqueue_stub()
The tasks scheduler can be accessed using the following code:
taskq = self.testbed.get_stub(testbed.TASKQUEUE_SERVICE_NAME)
The interface for working with the queue stub is as follows:
GetQueues() #returns a list of dictionaries with information about the available queues
#returns a list of dictionaries with information about the tasks in a given queue
GetTasks(queue_name)
DeleteTask(queue_name, task_name) #removes the task task_name from the given queue
FlushQueue(queue_name) #removes all the tasks from the queue
#returns tasks filtered by name & url pointed to by the task from the given queues
get_filtered_tasks(url, name, queue_names)
StartBackgroundExecution() #Executes the queued tasks
Shutdown() #Requests the task scheduler to shutdown.
Also, as this uses App Engine SDK own facilities - it works just fine with the deferred library.
The dev app server is single-threaded, so it can't run tasks in the background while the foreground thread is running the tests.
I modified TaskQueueTestCase in taskqueue.py in gaetestbed to add the following function:
def execute_tasks(self, application):
"""
Executes all currently queued tasks, and also removes them from the
queue.
The tasks are execute against the provided web application.
"""
# Set up the application for webtest to use (resetting _app in case a
# different one has been used before).
self._app = None
self.APPLICATION = application
# Get all of the tasks, and then clear them.
tasks = self.get_tasks()
self.clear_task_queue()
# Run each of the tasks, checking that they succeeded.
for task in tasks:
response = self.post(task['url'], task['params'])
self.assertOK(response)
For this to work, I also had to change the base class of TaskQueueTestCase from BaseTestCase to WebTestCase.
My tests then do something like this:
# Do something which enqueues a task.
# Check that a task was enqueued, then execute it.
self.assertTrue(len(self.get_tasks()), 1)
self.execute_tasks(some_module.application)
# Now test that the task did what was expected.
This therefore executes the task directly from the foreground unit test. This is not quite the same as in production (ie, the task will get executed 'some time later' on a separate request), but it works well enough for me.
You might want to try the following code. Full explanation is here: http://www.geewax.org/task-queue-support-in-app-engines-ext-testbed/
import unittest
from google.appengine.api import taskqueue
from google.appengine.ext import testbed
class TaskQueueTestCase(unittest.TestCase):
def setUp(self):
self.testbed = testbed.Testbed()
self.testbed.activate()
self.testbed.init_taskqueue_stub()
self.task_queue_stub = self.testbed.get_stub(testbed.TASKQUEUE_SERVICE_NAME)
def tearDown(self):
self.testbed.deactivate()
def testTaskAdded(self):
taskqueue.add(url='/path/to/task')
tasks = self.taskqueue_stub.get_filtered_tasks(url='/path/to/task')
self.assertEqual(1, len(tasks))
self.assertEqual('/path/to/task', tasks[0].url)
unittest.main()
