Test locali delle unità per Python 2

I test di unità ti consentono di verificare la qualità del codice dopo averlo scritto, ma puoi anche utilizzarli per migliorare il processo di sviluppo man mano che procedi. Anziché scrivere i test dopo aver completato lo sviluppo dell'applicazione, ti consigliamo di scriverli man mano. In questo modo, puoi progettare unità di codice piccole, manutenibili e riutilizzabili. Inoltre, ti consente di testare il codice in modo accurato e rapido.

Quando esegui test di unità locale, esegui test che rimangono all'interno del tuo ambiente di sviluppo senza coinvolgere componenti remoti. App Engine fornisce utilità di test che utilizzano implementazioni locali di Datastore e di altri servizi App Engine. Ciò significa che puoi verificare l'utilizzo di questi servizi da parte del tuo codice localmente, senza eseguire il deployment del codice in App Engine, utilizzando gli stub di servizio.

Uno stub di servizio è un metodo che simula il comportamento del servizio. Ad esempio, lo stub del servizio Datastore mostrato in Scrivere test di Datastore e Memcache ti consente di testare il codice di Datastore senza effettuare richieste al Datastore reale. Qualsiasi entità archiviata durante un test di unità del datastore viene memorizzata in memoria, non nel datastore, ed eliminata al termine dell'esecuzione del test. Puoi eseguire test piccoli e rapidi senza alcuna dipendenza dal datastore stesso.

Questo documento descrive come scrivere test di unità per diversi servizi App Engine locali, quindi fornisce alcune informazioni sulla configurazione di un framework di test.

Introduzione agli strumenti di test di Python 2

Un modulo Python di App Engine chiamato testbed rende disponibili gli stub di servizio per i test di unità.

Gli stub di servizio sono disponibili per i seguenti servizi:

  • App Identity init_app_identity_stub
  • Blobstore (utilizza init_blobstore_stub)
  • Capacità (utilizza init_capability_stub)
  • Datastore (utilizza init_datastore_v3_stub)
  • File (utilizza init_files_stub)
  • Immagini (solo per dev_appserver; utilizza init_images_stub)
  • LogService (utilizza init_logservice_stub)
  • Posta (usa init_mail_stub)
  • Memcache (utilizza init_memcache_stub)
  • Coda di attività (utilizza init_taskqueue_stub)
  • Recupero URL (utilizza init_urlfetch_stub)
  • Servizio per gli utenti (utilizza init_user_stub)

Per inizializzare tutti gli stub contemporaneamente, puoi utilizzare init_all_stubs.

Scrittura di test di Datastore e memcache

Questa sezione mostra un esempio di come scrivere codice che testa l'utilizzo dei servizi datastore e memcache.

Assicurati che il tuo programma di test abbia le librerie appropriate nel percorso di caricamento di Python, incluse le librerie di App Engine, yaml (incluse nell'SDK di App Engine), la radice dell'applicazione e eventuali altre modifiche al percorso della libreria previste dal codice dell'applicazione (ad esempio una directory ./lib locale, se ne hai una). Ad esempio:

import sys
sys.path.insert(1, 'google-cloud-sdk/platform/google_appengine')
sys.path.insert(1, 'google-cloud-sdk/platform/google_appengine/lib/yaml/lib')
sys.path.insert(1, 'myapp/lib')

Importa il modulo unittest di Python e i moduli App Engine pertinenti ai servizi in fase di test, in questo caso memcache e ndb, che utilizzano sia Datastore sia Memcache. Importa anche il modulo testbed.

import unittest

from google.appengine.api import memcache
from google.appengine.ext import ndb
from google.appengine.ext import testbed

Poi crea un corso TestModel. In questo esempio, una funzione controlla se un'entità è archiviata in memcache. Se non viene trovata alcuna entità, ne viene cercata un'altra nel datastore. Spesso questo può essere ridondante nella vita reale, poiché ndb utilizza memcache stesso dietro le quinte, ma è comunque un pattern accettabile per un test.

class TestModel(ndb.Model):
    """A model class used for testing."""
    number = ndb.IntegerProperty(default=42)
    text = ndb.StringProperty()


class TestEntityGroupRoot(ndb.Model):
    """Entity group root"""
    pass


def GetEntityViaMemcache(entity_key):
    """Get entity from memcache if available, from datastore if not."""
    entity = memcache.get(entity_key)
    if entity is not None:
        return entity
    key = ndb.Key(urlsafe=entity_key)
    entity = key.get()
    if entity is not None:
        memcache.set(entity_key, entity)
    return entity

A questo punto, crea un test case. Indipendentemente dai servizi che stai testando, il caso di test deve creare un'istanza Testbed e attivarla. Il caso di test deve anche inizializzare gli stub di servizio pertinenti, in questo caso utilizzando init_datastore_v3_stub e init_memcache_stub. I metodi per inizializzare altri stub di servizi App Engine sono elencati in Introduzione alle utilità di test di Python.

class DatastoreTestCase(unittest.TestCase):

    def setUp(self):
        # First, create an instance of the Testbed class.
        self.testbed = testbed.Testbed()
        # Then activate the testbed, which prepares the service stubs for use.
        self.testbed.activate()
        # Next, declare which service stubs you want to use.
        self.testbed.init_datastore_v3_stub()
        self.testbed.init_memcache_stub()
        # Clear ndb's in-context cache between tests.
        # This prevents data from leaking between tests.
        # Alternatively, you could disable caching by
        # using ndb.get_context().set_cache_policy(False)
        ndb.get_context().clear_cache()

Il metodo init_datastore_v3_stub() senza argomenti utilizza un datastore in memoria inizialmente vuoto. Se vuoi testare un'entità dell'area dati esistente, includi il relativo percorso come argomento di init_datastore_v3_stub().

Oltre a setUp(), includi un metodo tearDown() che disattivi il laboratorio. In questo modo vengono ripristinati gli stub originali in modo che i test non interferiscano tra loro.

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

Poi implementa i test.

def testInsertEntity(self):
    TestModel().put()
    self.assertEqual(1, len(TestModel.query().fetch(2)))

Ora puoi utilizzare TestModel per scrivere test che utilizzano gli stub dei servizi datastore o memcache anziché i servizi reali.

Ad esempio, il metodo mostrato di seguito crea due entità: la prima utilizza il valore predefinito per l'attributo number (42) e la seconda utilizza un valore non predefinito per number (17). Il metodo genera quindi una query per le entità TestModel, ma solo per quelle con il valore predefinito number.

Dopo aver recuperato tutte le entità corrispondenti, il metodo verifica che sia stata trovata esattamente un'entità e che il valore dell'attributo number di quell'entità sia il valore predefinito.

def testFilterByNumber(self):
    root = TestEntityGroupRoot(id="root")
    TestModel(parent=root.key).put()
    TestModel(number=17, parent=root.key).put()
    query = TestModel.query(ancestor=root.key).filter(
        TestModel.number == 42)
    results = query.fetch(2)
    self.assertEqual(1, len(results))
    self.assertEqual(42, results[0].number)

Come altro esempio, il seguente metodo crea un'entità e la recupera utilizzando la funzione GetEntityViaMemcache() che abbiamo creato sopra. Il metodo quindi verifica che sia stata restituita un'entità e che il suo valore number sia lo stesso dell'entità creata in precedenza.

def testGetEntityViaMemcache(self):
    entity_key = TestModel(number=18).put().urlsafe()
    retrieved_entity = GetEntityViaMemcache(entity_key)
    self.assertNotEqual(None, retrieved_entity)
    self.assertEqual(18, retrieved_entity.number)

Infine, invoca unittest.main().

if __name__ == '__main__':
    unittest.main()

Per eseguire i test, consulta Eseguire i test.

Scrittura di test di Cloud Datastore

Se la tua app utilizza Cloud Datastore, ti consigliamo di scrivere test che verifichino il comportamento dell'applicazione in caso di eventuale coerenza. db.testbed offre opzioni che semplificano questa procedura:

from google.appengine.datastore import datastore_stub_util  # noqa


class HighReplicationTestCaseOne(unittest.TestCase):

    def setUp(self):
        # First, create an instance of the Testbed class.
        self.testbed = testbed.Testbed()
        # Then activate the testbed, which prepares the service stubs for use.
        self.testbed.activate()
        # Create a consistency policy that will simulate the High Replication
        # consistency model.
        self.policy = datastore_stub_util.PseudoRandomHRConsistencyPolicy(
            probability=0)
        # Initialize the datastore stub with this policy.
        self.testbed.init_datastore_v3_stub(consistency_policy=self.policy)
        # Initialize memcache stub too, since ndb also uses memcache
        self.testbed.init_memcache_stub()
        # Clear in-context cache before each test.
        ndb.get_context().clear_cache()

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

    def testEventuallyConsistentGlobalQueryResult(self):
        class TestModel(ndb.Model):
            pass

        user_key = ndb.Key('User', 'ryan')

        # Put two entities
        ndb.put_multi([
            TestModel(parent=user_key),
            TestModel(parent=user_key)
        ])

        # Global query doesn't see the data.
        self.assertEqual(0, TestModel.query().count(3))
        # Ancestor query does see the data.
        self.assertEqual(2, TestModel.query(ancestor=user_key).count(3))

La classe PseudoRandomHRConsistencyPolicy ti consente di controllare la probabilità che un'operazione di scrittura venga applicata prima di ogni query globale (non principale). Impostando la probabilità su 0%, diamo istruzioni allo stub del datastore di operare con la massima coerenza finale. La coerenza finale massima significa che le scritture verranno committate, ma non verranno mai applicate, pertanto le query globali (non antecedenti) non riusciranno mai a vedere le modifiche. Ovviamente, questo non è rappresentativo della quantità di coerenza finale che la tua applicazione vedrà durante l'esecuzione in produzione, ma a scopo di test è molto utile poter configurare il datastore locale in modo che si comporti sempre in questo modo. Se utilizzi una probabilità diversa da zero, PseudoRandomHRConsistencyPolicy genera una sequenza deterministica di decisioni di coerenza in modo che i risultati del test siano coerenti:

def testDeterministicOutcome(self):
    # 50% chance to apply.
    self.policy.SetProbability(.5)
    # Use the pseudo random sequence derived from seed=2.
    self.policy.SetSeed(2)

    class TestModel(ndb.Model):
        pass

    TestModel().put()

    self.assertEqual(0, TestModel.query().count(3))
    self.assertEqual(0, TestModel.query().count(3))
    # Will always be applied before the third query.
    self.assertEqual(1, TestModel.query().count(3))

Le API di test sono utili per verificare che l'applicazione si comporti correttamente in presenza di coerenza finale, ma tieni presente che il modello locale di coerenza di lettura con replica elevata è un'approssimazione del modello di coerenza di lettura con replica elevata di produzione, non una replica esatta. Nell'ambiente locale, l'esecuzione di un get() di un Entity che appartiene a un gruppo di entità con una scrittura non applicata rende sempre visibili i risultati della scrittura non applicata alle query globali successive. In produzione non è così.

Scrivere test di posta

Puoi utilizzare lo stub del servizio di posta per testare il servizio mail. Come per altri servizi supportati dal testbed, all'inizio devi inizializzare lo stub, poi invocare il codice che utilizza l'API Mail e infine verificare se sono stati inviati i messaggi corretti.

import unittest

from google.appengine.api import mail
from google.appengine.ext import testbed


class MailTestCase(unittest.TestCase):

    def setUp(self):
        self.testbed = testbed.Testbed()
        self.testbed.activate()
        self.testbed.init_mail_stub()
        self.mail_stub = self.testbed.get_stub(testbed.MAIL_SERVICE_NAME)

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

    def testMailSent(self):
        mail.send_mail(to='alice@example.com',
                       subject='This is a test',
                       sender='bob@example.com',
                       body='This is a test e-mail')
        messages = self.mail_stub.get_sent_messages(to='alice@example.com')
        self.assertEqual(1, len(messages))
        self.assertEqual('alice@example.com', messages[0].to)

Scrivere test delle coda di attività

Puoi utilizzare lo stub taskqueue per scrivere test che utilizzano il servizio taskqueue. Come per gli altri servizi supportati dal testbed, all'inizio devi inizializzare lo stub, poi invocare il codice che utilizza l'API taskqueue e infine verificare se le attività sono state aggiunte correttamente alla coda.

import operator
import os
import unittest

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


class TaskQueueTestCase(unittest.TestCase):
    def setUp(self):
        self.testbed = testbed.Testbed()
        self.testbed.activate()

        # root_path must be set the the location of queue.yaml.
        # Otherwise, only the 'default' queue will be available.
        self.testbed.init_taskqueue_stub(
            root_path=os.path.join(os.path.dirname(__file__), 'resources'))
        self.taskqueue_stub = self.testbed.get_stub(
            testbed.TASKQUEUE_SERVICE_NAME)

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

    def testTaskAddedToQueue(self):
        taskqueue.Task(name='my_task', url='/url/of/my/task/').add()
        tasks = self.taskqueue_stub.get_filtered_tasks()
        self.assertEqual(len(tasks), 1)
        self.assertEqual(tasks[0].name, 'my_task')

Impostazione del file di configurazione queue.yaml

Se vuoi eseguire test sul codice che interagisce con una coda non predefinita, devi creare e specificare un file queue.yaml da utilizzare per l'applicazione. Di seguito è riportato un esempio di queue.yaml:

Per ulteriori informazioni sulle opzioni disponibili per queue.yaml, consulta la configurazione della coda di attività.

queue:
- name: default
  rate: 5/s
- name: queue-1
  rate: 5/s
- name: queue-2
  rate: 5/s

La posizione di queue.yaml viene specificata durante l'inizializzazione dello stub:

self.testbed.init_taskqueue_stub(root_path='.')

Nel sample, queue.yaml si trova nella stessa directory dei test. Se fosse in un'altra cartella, il percorso dovrebbe essere specificato in root_path.

Filtrare le attività

get_filtered_tasks dello stub taskqueue consente di filtrare le attività in coda. In questo modo è più facile scrivere test che devono verificare il codice che mette in coda più attività.

def testFiltering(self):
    taskqueue.Task(name='task_one', url='/url/of/task/1/').add('queue-1')
    taskqueue.Task(name='task_two', url='/url/of/task/2/').add('queue-2')

    # All tasks
    tasks = self.taskqueue_stub.get_filtered_tasks()
    self.assertEqual(len(tasks), 2)

    # Filter by name
    tasks = self.taskqueue_stub.get_filtered_tasks(name='task_one')
    self.assertEqual(len(tasks), 1)
    self.assertEqual(tasks[0].name, 'task_one')

    # Filter by URL
    tasks = self.taskqueue_stub.get_filtered_tasks(url='/url/of/task/1/')
    self.assertEqual(len(tasks), 1)
    self.assertEqual(tasks[0].name, 'task_one')

    # Filter by queue
    tasks = self.taskqueue_stub.get_filtered_tasks(queue_names='queue-1')
    self.assertEqual(len(tasks), 1)
    self.assertEqual(tasks[0].name, 'task_one')

    # Multiple queues
    tasks = self.taskqueue_stub.get_filtered_tasks(
        queue_names=['queue-1', 'queue-2'])
    self.assertEqual(len(tasks), 2)

Scrivere test delle attività differite

Se il codice dell'applicazione utilizza la libreria differita, puoi utilizzare lo stub taskqueue insieme a deferred per verificare che le funzioni differite vengano messe in coda ed eseguite correttamente.

def testTaskAddedByDeferred(self):
    deferred.defer(operator.add, 1, 2)

    tasks = self.taskqueue_stub.get_filtered_tasks()
    self.assertEqual(len(tasks), 1)

    result = deferred.run(tasks[0].payload)
    self.assertEqual(result, 3)

Modificare le variabili di ambiente predefinite

I servizi App Engine spesso dipendono dalle variabili di ambiente. Il metodo activate() della classe testbed.Testbed utilizza valori predefiniti per questi, ma puoi impostare valori personalizzati in base alle tue esigenze di test con il metodo setup_env della classe testbed.Testbed.

Ad esempio, supponiamo che tu abbia un test che memorizza diverse entità nel datastore, tutte collegate allo stesso ID applicazione. Ora vuoi eseguire nuovamente gli stessi test, ma utilizzando un ID applicazione diverso da quello collegato alle entità memorizzate. Per farlo, passa il nuovo valore a self.setup_env() come app_id.

Ad esempio:

import os
import unittest

from google.appengine.ext import testbed


class EnvVarsTestCase(unittest.TestCase):
    def setUp(self):
        self.testbed = testbed.Testbed()
        self.testbed.activate()
        self.testbed.setup_env(
            app_id='your-app-id',
            my_config_setting='example',
            overwrite=True)

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

    def testEnvVars(self):
        self.assertEqual(os.environ['APPLICATION_ID'], 'your-app-id')
        self.assertEqual(os.environ['MY_CONFIG_SETTING'], 'example')

Simulazione di accesso

Un altro utilizzo frequente di setup_env è simulare l'accesso di un utente, con o senza privilegi amministrativi, per verificare se i gestori funzionano correttamente in ogni caso.

import unittest

from google.appengine.api import users
from google.appengine.ext import testbed


class LoginTestCase(unittest.TestCase):
    def setUp(self):
        self.testbed = testbed.Testbed()
        self.testbed.activate()
        self.testbed.init_user_stub()

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

    def loginUser(self, email='user@example.com', id='123', is_admin=False):
        self.testbed.setup_env(
            user_email=email,
            user_id=id,
            user_is_admin='1' if is_admin else '0',
            overwrite=True)

    def testLogin(self):
        self.assertFalse(users.get_current_user())
        self.loginUser()
        self.assertEquals(users.get_current_user().email(), 'user@example.com')
        self.loginUser(is_admin=True)
        self.assertTrue(users.is_current_user_admin())

Ora i metodi di test possono chiamare, ad esempio, self.loginUser('', '') per simulare l'assenza di un utente che ha eseguito l'accesso, self.loginUser('test@example.com', '123') per simulare l'accesso di un utente non amministratore e self.loginUser('test@example.com', '123', is_admin=True) per simulare l'accesso di un utente amministratore.

Configurazione di un framework di test

Le utilità di test dell'SDK non sono legate a un framework specifico. Puoi eseguire i test di unità con qualsiasi test runner App Engine disponibile, ad esempio nose-gagg o ferrisnose. Puoi anche scrivere un semplice test runner o utilizzare quello mostrato di seguito.

I seguenti script utilizzano il modulo unittest di Python.

Puoi assegnare allo script il nome che preferisci. Quando lo esegui, fornisci il percorso dell'installazione di Google Cloud CLI o Google App Engine SDK e il percorso dei moduli di test. Lo script rileverà tutti i test nel percorso specificato e stamperà i risultati nello stream di errori standard. I file di test seguono la convenzione di avere test come prefisso al nome.

"""App Engine local test runner example.

This program handles properly importing the App Engine SDK so that test modules
can use google.appengine.* APIs and the Google App Engine testbed.

Example invocation:

    $ python runner.py ~/google-cloud-sdk
"""

import argparse
import os
import sys
import unittest


def fixup_paths(path):
    """Adds GAE SDK path to system path and appends it to the google path
    if that already exists."""
    # Not all Google packages are inside namespace packages, which means
    # there might be another non-namespace package named `google` already on
    # the path and simply appending the App Engine SDK to the path will not
    # work since the other package will get discovered and used first.
    # This emulates namespace packages by first searching if a `google` package
    # exists by importing it, and if so appending to its module search path.
    try:
        import google
        google.__path__.append("{0}/google".format(path))
    except ImportError:
        pass

    sys.path.insert(0, path)


def main(sdk_path, test_path, test_pattern):
    # If the SDK path points to a Google Cloud SDK installation
    # then we should alter it to point to the GAE platform location.
    if os.path.exists(os.path.join(sdk_path, 'platform/google_appengine')):
        sdk_path = os.path.join(sdk_path, 'platform/google_appengine')

    # Make sure google.appengine.* modules are importable.
    fixup_paths(sdk_path)

    # Make sure all bundled third-party packages are available.
    import dev_appserver
    dev_appserver.fix_sys_path()

    # Loading appengine_config from the current project ensures that any
    # changes to configuration there are available to all tests (e.g.
    # sys.path modifications, namespaces, etc.)
    try:
        import appengine_config
        (appengine_config)
    except ImportError:
        print('Note: unable to import appengine_config.')

    # Discover and run tests.
    suite = unittest.loader.TestLoader().discover(test_path, test_pattern)
    return unittest.TextTestRunner(verbosity=2).run(suite)


if __name__ == '__main__':
    parser = argparse.ArgumentParser(
        description=__doc__,
        formatter_class=argparse.RawDescriptionHelpFormatter)
    parser.add_argument(
        'sdk_path',
        help='The path to the Google App Engine SDK or the Google Cloud SDK.')
    parser.add_argument(
        '--test-path',
        help='The path to look for tests, defaults to the current directory.',
        default=os.getcwd())
    parser.add_argument(
        '--test-pattern',
        help='The file pattern for test modules, defaults to *_test.py.',
        default='*_test.py')

    args = parser.parse_args()

    result = main(args.sdk_path, args.test_path, args.test_pattern)

    if not result.wasSuccessful():
        sys.exit(1)

Esecuzione dei test

Puoi eseguire questi test semplicemente eseguendo lo script runner.py, descritto in dettaglio nella sezione Configurare un framework di test:

python runner.py <path-to-appengine-or-gcloud-SDK> .