Writing unit tests for Appengine's taskqueue and deferred

19 Oct 2016

Writing unit tests for Appengine's taskqueue and deferred involves stubbing a queue and processing items in the queue.

It's a bit difficult to find the correct APIs to do these things. So, here's an example:

import pickle import unittest from google.appengine.ext import deferred from google.appengine.ext import testbed run = 0 def work(offset, limit): run += 1 if offset < limit: deferred.defer(work, offset+1, limit=limit) class DeferredTest(unittest.TestCase): def setUp(self): run = 0 self.testbed = testbed.Testbed() self.testbed.activate() self.testbed.init_taskqueue_stub() self.taskqueue = self.testbed.get_stub(testbed.TASKQUEUE_SERVICE_NAME) def tearDown(self): self.testbed.deactivate() def _run_tasks(self): """Dequeue and run tasks until there's no more task.""" while True: tasks = self.taskqueue.get_filtered_tasks() if not tasks: break deferred.run(tasks[0].payload) self.taskqueue.DeleteTask('default', tasks[0].name) def test_queue(self): work(8, 10) tasks = self.taskqueue.get_filtered_tasks() self.assertEqual(1, len(tasks)) func, args, kwargs = pickle.loads(tasks[0].payload) self.assertEqual('work', func.__name__) self.assertListEqual([9], args) self.assertDictEqual({'limit': 10}, kwargs) def test_end(self): work(10, 10) self.assertEqual(0, len(self.taskqueue.get_filtered_tasks()) def test_work_until_end(self): work(0, 10) self._run_tasks() self.assertEqual(10, run)

Give it a kudos