python - GAE: cola de prueba de pruebas unitarias con testbed

unit-testing google-app-engine

Question

Estoy usando testbed para testeo unitario de mi aplicación de motor de Google, y mi aplicación usa una cola de tareas.

Cuando envío una tarea a una cola de tareas durante una prueba de unidad, parece que la tarea está en la cola, pero la tarea no se ejecuta.

¿Cómo hago que la tarea se ejecute durante una prueba de unidad?

Answer 1

El servidor de aplicaciones dev es de un solo subproceso, por lo que no puede ejecutar tareas en segundo plano mientras el subproceso en primer plano está ejecutando las pruebas.

Modifiqué TaskQueueTestCase en taskqueue.py en gaetestbed para agregar la siguiente función:

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)

Para que esto funcione, también tuve que cambiar la clase base de TaskQueueTestCase de BaseTestCase a WebTestCase.

Mis pruebas luego hacen algo como esto:

# 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.

Por lo tanto, esto ejecuta la tarea directamente desde la prueba de unidad de primer plano. Esto no es exactamente lo mismo que en producción (es decir, la tarea se ejecutará ''algún tiempo después'' en una solicitud por separado), pero funciona lo suficientemente bien para mí.

Answer 2

Es posible que desee probar el siguiente código. La explicación completa está aquí: 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()

Answer 3

Otra opción (más limpia) para lograr esto es utilizar el talón de la cola de tareas dentro del banco de pruebas. Para hacer esto, primero debe inicializar el setUp() la cola de tareas agregando lo siguiente a su método setUp() :

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

Se puede acceder al programador de tareas utilizando el siguiente código:

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

La interfaz para trabajar con el apéndice de cola es la siguiente:

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.

Además, dado que utiliza las propias instalaciones de App Engine SDK, funciona bien con la biblioteca diferida.

Answer 4

Usando la excelente respuesta de Saxon, pude hacer lo mismo usando testbed en lugar de gaetestbed. Aquí esta lo que hice.

Agregado esto a mi setUp() :

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

Luego, en mi prueba, usé lo siguiente:

 # 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)

En algún lugar a lo largo de la línea, los parámetros POST se codifican en base64, por lo que tuve que deshacerlos para que funcionen.

Me gusta esto mejor que la respuesta de Saxon, ya que puedo usar el paquete oficial de testbed y puedo hacerlo todo dentro de mi propio código de prueba.

EDITAR: Más tarde quise hacer lo mismo con las tareas enviadas utilizando la biblioteca diferida, y me costó un poco de cabeza para entenderlo, así que estoy compartiendo aquí para aliviar el dolor de otras personas.

Si su cola de tareas contiene solo las tareas enviadas con aplazado, esto ejecutará todas las tareas y las tareas en cola de esas tareas:

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")