很长一段时间以来，我一直在使用 AppEngine 上的任务队列来安排任务，就像我应该做的那样。
但我一直想知道的是如何为此编写测试？到目前为止，我只是进行了测试以确保在排队任务的 API 上不会发生错误，然后为执行任务的 API 编写更合适的测试。
但是最近我开始对此感到有些不满意，我正在寻找一种方法来实际测试正确的任务是否已添加到正确的队列中。希望这可以比简单地通过部署代码并希望最好的方式做得更好。
我正在使用 django-nonrel，如果这对答案有任何影响。
回顾一下：如何编写单元测试来确认任务已排队？
如果您使用google.appengine.ext.testbed而不是 GAE Testbed（GAE Testbed 现在已弃用并移至ext.testbed），您可以执行以下操作：
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(...)
使用GAE Test Bed将允许您存根任务队列。
如果您从FunctionalTestCaseor继承TaskQueueTestCase，您将获得诸如get_tasksand之类的方法assertTasksInQueue。
您实际上也可以运行这些任务。如何做到这一点取决于您是使用任务还是延迟。
对于延期，我有一些这样的代码：
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']))
运行任务是类似的，但是在您获取任务之后，您使用 WebTest（GAE 测试床建立在其之上）作为 POST 请求提交到任务的 URL 及其参数。
有一个名为 gaetestbed 的小测试框架可能适合您的需要。详情请参考：https ://github.com/jgeewax/gaetestbed 。
这个测试环境与nose、nose-gae 插件和WebTest 包一起工作。就我而言，鉴于 python 包的混合是测试 GAE 应用程序的最佳方式。
SDK 1.4.3 Testbed API为本地集成测试提供了简单的存根库配置。
一个服务存根Task Queue可用。