在AppEngine中对任务队列进行单元测试任务

Question

长期以来，我一直在使用AppEngine上的任务队列来安排任务，就像我应该做的那样。

但是我一直在想，如何为此编写测试呢？到目前为止，我只是编写了测试来确保在排队任务的API上不会发生错误，然后编写了更适当的测试来执行任务的API。

然而，最近我对此感到有些不满意，并正在寻找一种方法来实际测试已将正确的任务添加到正确的队列中。希望这可以比仅部署代码并希望一切顺利做得更好。

如果有任何影响答案的话，我正在使用django-nonrel。

总之：如何编写单元测试以确认已排队的任务？

- Swizec Teller

Answer 1

如果你正在使用 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(...)

- JJ Geewax

---

你能用自己的队列来实现这个吗？我使用了deferred，像这样： deferred.defer(get_and_store_all_new_messages_async, user_id, query, next_page_token, _queue="emailFetch") 但它无法识别队列，可能是因为它没有读取单元测试的queue.yaml文件。 - Debnath Sinha

Answer 2

使用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。

- Luke Francl

---

如果您正在使用所有基本测试用例（https://github.com/jgeewax/gaetestbed/blob/master/gaetestbed/taskqueue.py），则可以使用`self.get_tasks`和`task['decoded_body']`作为快捷方式。此外，这个功能希望很快成为google.appengine.ext.testbed的一部分。有一个名为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

请查看我的回答：这个库现在已经被弃用，推荐使用 ext.testbed（https://developers.google.com/appengine/docs/python/tools/localunittesting）。 - JJ Geewax

Answer 3

有一个小型的测试框架叫做gaetestbed，可能适合您的需求。详情请参考：https://github.com/jgeewax/gaetestbed。

这个测试环境与nose、nose-gae插件和WebTest包配合使用。在我看来，这些Python包的混合使用是测试GAE应用程序的最佳方式。

- Piotr Duda

---

这个包已经过时了，因为我已经将其功能合并到了 google.appengine.ext.testbed 中 (https://developers.google.com/appengine/docs/python/tools/localunittesting)。 - JJ Geewax

Answer 4

SDK 1.4.3 Testbed API 提供了易于配置本地集成测试存根库的功能。

可用的任务队列服务存根。

- systempuntoout

---

此链接已经失效。 - pdoherty926