GAE: очередь задач модульного тестирования с испытательным стендом - PullRequest
Новая
       0

GAE: очередь задач модульного тестирования с испытательным стендом

17 голосов
/ 09 июля 2011

Я использую тестовый стенд для модульного тестирования моего приложения Google App Engine, а мое приложение использует очередь задач.

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

Как мне выполнить задачу во время юнит-теста?

Ответы [ 4 ]

24 голосов
/ 09 июля 2011

Используя превосходный ответ Саксона, я смог сделать то же самое, используя испытательный стенд вместо gaetestbed. Вот что я сделал.

Добавил это в мой setUp(): 

    self.taskqueue_stub = apiproxy_stub_map.apiproxy.GetStub('taskqueue')

Затем в своем тесте я использовал следующее: 

    # Execute the task in the taskqueue
    tasks = self.taskqueue_stub.GetTasks("default")
    self.assertEqual(len(tasks), 1)
    task = tasks[0]
    params = base64.b64decode(task["body"])
    response = self.app.post(task["url"], params)

Где-то вдоль линии параметры POST кодируются в base64, поэтому пришлось отменить это, чтобы заставить его работать.

Мне нравится этот ответ лучше, чем ответ Саксона, поскольку я могу использовать официальный пакет тестового стенда и могу делать все это в своем собственном тестовом коде.

РЕДАКТИРОВАТЬ: Позже я захотел сделать то же самое с задачами, переданными с использованием отложенной библиотеки, и потребовалось немало потрясений, чтобы понять это, поэтому я делюсь здесь, чтобы облегчить боль других людей.

Если ваша очередь задач содержит только задачи, отправленные с отложенным выполнением, тогда будут запущены все задачи и любые задачи, поставленные в очередь этими задачами: 

def submit_deferred(taskq):
    tasks = taskq.GetTasks("default")
    taskq.FlushQueue("default")
    while tasks:
        for task in tasks:
            (func, args, opts) = pickle.loads(base64.b64decode(task["body"]))
            func(*args)
        tasks = taskq.GetTasks("default")
        taskq.FlushQueue("default")
14 голосов
/ 07 сентября 2011

Другим (более чистым) вариантом для достижения этой цели является использование заглушки очереди задач на испытательном стенде. Для этого сначала нужно инициализировать заглушку очереди задач, добавив в метод setUp() следующее: 

self.testbed = init_testbed()
self.testbed.init_taskqueue_stub()

Доступ к планировщику задач можно получить с помощью следующего кода: 

taskq = self.testbed.get_stub(testbed.TASKQUEUE_SERVICE_NAME)

Интерфейс для работы с заглушкой очереди выглядит следующим образом: 

GetQueues() #returns a list of dictionaries with information about the available queues

#returns a list of dictionaries with information about the tasks in a given queue
GetTasks(queue_name)

DeleteTask(queue_name, task_name) #removes the task task_name from the given queue

FlushQueue(queue_name) #removes all the tasks from the queue

#returns tasks filtered by name & url pointed to by the task from the given queues
get_filtered_tasks(url, name, queue_names)

StartBackgroundExecution() #Executes the queued tasks

Shutdown() #Requests the task scheduler to shutdown.

Кроме того, поскольку он использует собственные средства App Engine SDK - он отлично работает с отложенной библиотекой.

8 голосов
/ 09 июля 2011

Сервер приложений dev является однопоточным, поэтому он не может запускать задачи в фоновом режиме, пока поток переднего плана выполняет тесты.

Я изменил TaskQueueTestCase в taskqueue.py в gaetestbed, чтобы добавить следующую функцию: 

def execute_tasks(self, application):
    """
    Executes all currently queued tasks, and also removes them from the 
    queue.
    The tasks are execute against the provided web application.
    """

    # Set up the application for webtest to use (resetting _app in case a
    # different one has been used before). 
    self._app = None
    self.APPLICATION = application

    # Get all of the tasks, and then clear them.
    tasks = self.get_tasks()
    self.clear_task_queue()

    # Run each of the tasks, checking that they succeeded.
    for task in tasks:
        response = self.post(task['url'], task['params'])
        self.assertOK(response)

Чтобы это работало, мне также пришлось изменить базовый класс TaskQueueTestCase с BaseTestCase на WebTestCase.

Мои тесты затем делают что-то вроде этого: 

# Do something which enqueues a task.

# Check that a task was enqueued, then execute it.
self.assertTrue(len(self.get_tasks()), 1)
self.execute_tasks(some_module.application)

# Now test that the task did what was expected.

Таким образом, эта задача выполняется непосредственно из модульного теста переднего плана. Это не совсем то же самое, что и в производственном процессе (т. Е. Задача будет выполнена «через некоторое время» по отдельному запросу), но для меня это работает достаточно хорошо.

3 голосов
/ 11 июля 2013

Возможно, вы захотите попробовать следующий код. Полное объяснение здесь: http://www.geewax.org/task-queue-support-in-app-engines-ext-testbed/

import unittest
from google.appengine.api import taskqueue
from google.appengine.ext import testbed


class TaskQueueTestCase(unittest.TestCase):

  def setUp(self):
    self.testbed = testbed.Testbed()
    self.testbed.activate()
    self.testbed.init_taskqueue_stub()
    self.task_queue_stub = self.testbed.get_stub(testbed.TASKQUEUE_SERVICE_NAME)

  def tearDown(self):
    self.testbed.deactivate()

  def testTaskAdded(self):
    taskqueue.add(url='/path/to/task')

    tasks = self.taskqueue_stub.get_filtered_tasks(url='/path/to/task')
    self.assertEqual(1, len(tasks))
    self.assertEqual('/path/to/task', tasks[0].url)

unittest.main()
...