Kolejki zadań testowania jednostkowego w AppEngine

Question

Od bardzo dawna używam kolejek zadań w AppEngine do planowania zadań, dokładnie tak, jak powinienem.

Ale zawsze zastanawiałem się, w jaki sposób można napisać testy? Do tej pory po prostu robiłem testy, aby upewnić się, że błąd nie występuje w interfejsie API, który umieszcza w kolejce zadanie, a następnie napisał bardziej odpowiednie testy interfejsu API wykonującego to zadanie.

Jednak ostatnio zacząłem czuć się trochę niezadowolony z tego i szukam sposobu, aby faktycznie sprawdzić, czy prawidłowe zadanie zostało dodane do właściwej kolejki. Mam nadzieję, że można to zrobić lepiej niż po prostu przez wdrożenie kodu i nadzieję na najlepsze.

Używam django-nonrel, jeśli ma to wpływ na odpowiedź.

Podsumowując: W jaki sposób można zapisać test jednostki, aby potwierdzić, że zadania zostały umieszczone w kolejce?

Answer 1

Użycie GAE Test Bed pozwoli ci wyodrębnić kolejkę zadań.

Jeśli dziedziczysz po FunctionalTestCase lub TaskQueueTestCase, otrzymasz metody takie jak get_tasks i assertTasksInQueue.

Możesz także uruchomić te zadania. Sposób wykonania różni się w zależności od tego, czy korzystasz z zadań, czy odroczonych.

Dla deferreds mam jakiś kod tak:

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

przebiegu zadania jest podobna, ale po sprowadzić zadanie, należy użyć WebTest (który Bed GAE test jest zbudowany na górze), aby przedstawić jako Żądanie POST do adresu URL zadania z jego parametrami.

Answer 2

Istnieje mały szkielet testowy o nazwie gaetestbed, który może odpowiadać potrzebom. Szczegółowe informacje znajdują się pod adresem: https://github.com/jgeewax/gaetestbed.

To środowisko testowe działa w połączeniu z wtyczką typu nos, nos-gae i pakietem WebTest. Biorąc pod uwagę mieszankę pakietów Pythona jest najlepszym sposobem, aby przetestować aplikacje GAE, o ile mi chodzi.

Answer 3

SDK 1.4.3Testbed API zapewnia łatwą konfigurację bibliotek pośredniczących do lokalnych testów integracyjnych.

Dostępna jest usługa Service stubs dla Task Queue.

Answer 4

Jeśli używasz google.appengine.ext.testbed zamiast GAE pilotowej (GAE Testbed jest teraz przestarzała i wnoszenia do ext.testbed), można wykonać następujące czynności:

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