AppEngine 中的单元测试任务队列

发布于 2024-10-16 14:35:15 字数 297 浏览 4 评论 0原文

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

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

然而最近我开始对此感到有点不满意，我正在寻找一种方法来实际测试正确的任务是否已添加到正确的队列中。希望这可以比简单地部署代码并希望得到最好的结果更好。

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

回顾一下：如何编写单元测试来确认任务已排队？

原文

分享到QQ

分享到微博

如果你对这篇内容有疑问，欢迎到本站社区发帖提问参与讨论，获取更多帮助，或者扫码二维码加入 Web 技术交流群。

发布评论

需要登录才能够评论，你可以免费注册一个本站的账号。

丶视觉 2024-10-23 14:35:15

如果您使用 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(...)

If you're using google.appengine.ext.testbed instead of GAE Testbed (GAE Testbed is now deprecated and being moved into ext.testbed), you can do the following:

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(...)

回复收藏 0 原文

盗梦空间 2024-10-23 14:35:15

使用 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 测试床构建于其之上）以 POST 请求及其参数提交到任务的 URL。

Using GAE Test Bed will allow you to stub out a task queue.

If you inherit from FunctionalTestCase or TaskQueueTestCase, you'll get methods such as get_tasks and assertTasksInQueue.

You can actually run the tasks, too. How to do it differs depending on whether you use tasks or deferred.

For deferreds, I have some code like this:

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']))

Running tasks is similar, but after you fetch the task, you use WebTest (which GAE Test Bed is built on top of) to submit as POST request to the task's URL with its parameters.

回复收藏 0 原文