Очереди задач тестирования модулей в AppEngine

17

В течение очень долгого времени я использую очереди задач в AppEngine для планирования задач, как раз так, как я должен.

Но то, что мне всегда было интересно, - это как писать тесты для этого? До сих пор я просто делал тесты, чтобы убедиться, что в API нет очереди, которая ставит в очередь задачу, а затем написала более подходящие тесты для API, выполняющего задачу.

Однако в последнее время я начал чувствовать себя немного неудовлетворенным этим, и я ищу способ проверить, что правильная задача была добавлена ​​в правильную очередь. Надеюсь, это можно сделать лучше, чем просто путем развертывания кода и надежды на лучшее.

Я использую django-nonrel, если это имеет какое-либо отношение к ответу.

Напомним: как можно записать единичный тест для подтверждения задач, поставленных в очередь?

    
задан Swizec Teller 07.02.2011 в 23:44
источник

4 ответа

13

Используя тестовую кровать GAE , вы можете закрыть очередь задач.

Если вы наследуете 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-адрес задачи с его параметрами.

    
ответ дан Luke Francl 08.02.2011 в 01:26
  • Похоже, что это будет трюк. Благодаря! –  Swizec Teller 10.02.2011 в 00:35
  • Вы можете использовать self.get_tasks и задачу ['decoded_body'] в качестве ярлыка, если вы используете все базовые тестовые примеры (github.com/jgeewax/gaetestbed/blob/master/gaetestbed/...). Кроме того, это, надеюсь, скоро станет частью google.appengine.ext.testbed. Существует метод get_filtered_tasks, который очень похож на get_tasks (code.google.com/p/googleappengine/source/browse/trunk/python/...) –  JJ Geewax 05.07.2012 в 03:11
  • См. мой ответ: эта библиотека теперь устарела в пользу ext.testbed (developers.google.com/appengine/docs/python/tools/...) –  JJ Geewax 17.06.2014 в 08:14
18

Если вы используете 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(...)
    
ответ дан JJ Geewax 05.07.2012 в 03:35
  • Это должен быть принятый ответ. благодаря –  Houman 15.06.2014 в 13:11
  • Возможно ли это сделать со своими очередями? Я использую отложенные как: deferred.defer (get_and_store_all_new_messages_async, user_id, query, next_page_token, _queue="emailFetch"), но он не распознает очередь, вероятно, потому, что он не читал файл queue.yaml для модульного теста –  Debnath Sinha 18.11.2014 в 14:18
1

Существует небольшая тестовая структура, называемая gaetestbed, которая может удовлетворить ваши потребности. Подробнее см. Ссылка .

Эта среда тестирования работает в сочетании с плагином для носа, носа-gae и пакетом WebTest. Данное сочетание пакетов python - лучший способ проверить приложения GAE, насколько я могу судить.     

ответ дан Piotr Duda 08.02.2011 в 00:21
  • Этот пакет фактически не рекомендуется, поскольку я объединил эти функции в google.appengine.ext.testbed (developers.google.com/appengine/docs/python/tools/...) –  JJ Geewax 17.06.2014 в 08:14
0

SDK 1.4.3 Testbed API обеспечивает легкую конфигурацию библиотек-заглушек для локальных тестов интеграции.

Доступны служебные заглушки для Task Queue .

    
ответ дан systempuntoout 31.03.2011 в 10:12