长期以来,我一直在使用AppEngine上的任务队列来安排任务,就像我应该做的那样。
但是我一直在想,如何为此编写测试呢?到目前为止,我只是编写了测试来确保在排队任务的API上不会发生错误,然后编写了更适当的测试来执行任务的API。
然而,最近我对此感到有些不满意,并正在寻找一种方法来实际测试已将正确的任务添加到正确的队列中。希望这可以比仅部署代码并希望一切顺利做得更好。
如果有任何影响答案的话,我正在使用django-nonrel。
总之:如何编写单元测试以确认已排队的任务?
长期以来,我一直在使用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可以让您创建任务队列的存根。
如果您从FunctionalTestCase
或TaskQueueTestCase
继承,您将获得诸如get_tasks
和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 Test Bed基于其构建)将其参数作为POST请求提交到任务的URL。
get_filtered_tasks
的方法,其工作原理类似于get_tasks
(https://code.google.com/p/googleappengine/source/browse/trunk/python/google/appengine/api/taskqueue/taskqueue_stub.py#2453)。 - JJ Geewax有一个小型的测试框架叫做gaetestbed,可能适合您的需求。详情请参考:https://github.com/jgeewax/gaetestbed。
这个测试环境与nose、nose-gae插件和WebTest包配合使用。在我看来,这些Python包的混合使用是测试GAE应用程序的最佳方式。