Code attività di verifica unità in AppEngine

Question

Per molto tempo ho utilizzato le code delle attività su AppEngine per pianificare le attività, nel modo in cui dovrei.

Ma quello che mi sono sempre chiesto è come si scrivono i test per quello? Fino ad ora ho semplicemente fatto dei test per assicurarmi che non si verificasse un errore sull'API che accoda un task e poi scrissi i test più appropriati per l'API che esegue l'attività.

Tuttavia ultimamente ho iniziato a sentirmi un po 'insoddisfatto di questo e sto cercando un modo per testare effettivamente che l'attività corretta è stata aggiunta alla coda corretta. Speriamo che questo possa essere fatto meglio del semplice dispiegamento del codice e sperando per il meglio.

Uso il django-nonrel, se questo ha qualche influenza sulla risposta.

Ricapitolare: come si può scrivere un test unitario per confermare le attività sono state messe in coda?

Answer 1

L'utilizzo di GAE Test Bed consente di eseguire lo stub di una coda di attività.

Se si eredita da FunctionalTestCase o TaskQueueTestCase, si otterranno metodi come get_tasks e assertTasksInQueue.

È possibile eseguire anche le attività. Come farlo differisce a seconda che si utilizzino attività o differite.

Per deferreds, ho qualche codice come questo:

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

attività in corso è simile, ma dopo si recupera l'operazione, è possibile utilizzare WebTest (che GAE Test Bed è costruito in cima) a presentare come Richiesta POST all'URL dell'attività con i suoi parametri.

Answer 2

C'è un piccolo quadro di prova chiamato gaetestbed che può soddisfare le vostre necessità. Per dettagli, fare riferimento a: https://github.com/jgeewax/gaetestbed.

Questo ambiente di test funziona in combinazione con il plugin nose, nose-gae e il pacchetto WebTest. Dato il mix di pacchetti Python è il modo migliore per testare le applicazioni GAE per quanto mi riguarda.

Answer 3

Il SDK 1.4.3Testbed API fornisce una facile configurazione delle librerie di stub per i test di integrazione locali.

Un modulo di servizio per Task Queue è disponibile.

Answer 4

Se stai usando google.appengine.ext.testbed invece di GAE Testbed (GAE Testbed è ormai deprecato e di essere spostato in ext.testbed), è possibile effettuare le seguenti operazioni:

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