【问题标题】:Unit testing task queues in AppEngineAppEngine 中的单元测试任务队列
【发布时间】:2011-06-23 02:07:32
【问题描述】:

很长一段时间以来,我一直在使用 AppEngine 上的任务队列来安排任务,这正是我应该做的。

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

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

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

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

【问题讨论】:

    标签: unit-testing google-app-engine django-nonrel


    【解决方案1】:

    如果您使用 google.appengine.ext.testbed 而不是 GAE 测试平台(GAE 测试平台现已弃用并移至 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(...)
    

    【讨论】:

    • 是否可以使用您自己的队列来执行此操作?我使用 deferred 之类的: deferred.defer(get_and_store_all_new_messages_async, user_id, query, next_page_token, _queue="emailFetch") 但它无法识别队列,可能是因为它没有读取用于单元测试的 queue.yaml 文件跨度>
    【解决方案2】:

    使用GAE Test Bed 将允许您存根任务队列。

    如果你从FunctionalTestCaseTaskQueueTestCase继承,你会得到get_tasksassertTasksInQueue这样的方法。

    您也可以实际运行这些任务。如何做到这一点取决于您是使用任务还是延迟。

    对于延期,我有一些这样的代码:

    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 及其参数。

    【讨论】:

    【解决方案3】:

    有一个名为 gaetestbed 的小测试框架可能适合您的需要。详情请参考:https://github.com/jgeewax/gaetestbed

    此测试环境与nose、nose-gae 插件和WebTest 包配合使用。就我而言,鉴于 python 包的混合是测试 GAE 应用程序的最佳方式。

    【讨论】:

    【解决方案4】:

    SDK 1.4.3Testbed API 为本地集成测试提供了简单的存根库配置。

    Task Queue 的服务存根可用。

    【讨论】:

    • 此链接现已损坏。
    猜你喜欢
    • 1970-01-01
    • 1970-01-01
    • 1970-01-01
    • 2017-04-17
    • 1970-01-01
    • 1970-01-01
    • 1970-01-01
    • 2013-09-18
    • 1970-01-01
    相关资源
    最近更新 更多