Tests unitaires en local pour Python 2

Les tests unitaires permettent de vérifier la qualité de votre code après l'avoir écrit, mais également d'améliorer progressivement votre processus de développement. Au lieu d'écrire les tests une fois que vous avez fini de développer votre application, écrivez-les en même temps. Vous pouvez ainsi concevoir de petites unités de code gérables et réutilisables. Il est également plus simple pour vous de tester complètement et rapidement votre code.

Lorsque vous réalisez un test unitaire local, celui-ci s'exécute dans votre propre environnement de développement sans impliquer de composants distants. App Engine propose des utilitaires de test qui s'appuient sur des mises en œuvre locales du datastore et d'autres services App Engine. De cette façon, vous pouvez exécuter le code en local, sans le déployer sur App Engine, grâce aux simulations de service.

Une simulation de service est une méthode qui simule le comportement du service. Par exemple, la simulation de service du datastore indiquée dans la section Écrire des tests pour le datastore et le cache mémoire vous permet de tester le code de votre datastore, sans effectuer de requête auprès du véritable datastore. Toute entité stockée lors d'un test unitaire du datastore est conservée en mémoire (pas dans le datastore), puis en est supprimée après le test. Vous pouvez donc effectuer de petits tests rapides, sans aucune dépendance sur le datastore.

Ce document décrit la procédure d'écriture de tests unitaires pour plusieurs services App Engine locaux et donne des informations sur la configuration d'un framework de test.

Présentation des utilitaires de test Python 2

Un module App Engine Python nommé testbed assure la disponibilité des simulations de service dans le cadre des tests unitaires.

Voici les services pour lesquels des simulations de service sont disponibles :

  • App Identity init_app_identity_stub
  • Blobstore (utilisez init_blobstore_stub)
  • Capabilities (utilisez init_capability_stub)
  • Datastore (utilisez init_datastore_v3_stub)
  • Files (utilisez init_files_stub)
  • Images (seulement pour dev_appserver, utilisez init_images_stub)
  • LogService (utilisez init_logservice_stub)
  • Messagerie (utilisez init_mail_stub)
  • Memcache (utilisez init_memcache_stub)
  • Task Queue (utilisez init_taskqueue_stub)
  • Récupération d'URL (utilisez init_urlfetch_stub)
  • Users (utilisez init_user_stub)

Pour initialiser toutes les simulations en même temps, vous pouvez utiliser init_all_stubs.

Écrire des tests pour le datastore et Memcache

Cette section indique comment écrire un code permettant de tester l'utilisation des services de Datastore et Memcache.

Assurez-vous que votre lanceur de test dispose des bibliothèques appropriées sur le chemin de chargement Python, y compris les bibliothèques App Engine, le fichier yaml (inclus dans le SDK App Engine), la racine de l'application et toute autre modification du chemin de la bibliothèque attendu par le code de l'application (telle qu'un répertoire ./lib local, si vous en avez un). Exemple :

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

Importez le module Python unittest et les modules App Engine pertinents pour les services testés. Dans ce cas, il s'agit de memcache et ndb, qui utilisent à la fois le datastore et Memcache. Importez également le module testbed.

import unittest

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

Créez ensuite une classe TestModel. Dans cet exemple, une fonction s'assure qu'une entité est stockée dans Memcache. Si elle ne trouve aucune entité, elle en recherche une dans le datastore. Cette procédure peut souvent s'avérer redondante en configuration réelle, puisque ndb utilise lui-même Memcache en arrière-plan, mais reste acceptable dans le contexte d'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

Ensuite, créez un scénario de test. Quels que soient les services que vous testez, le scénario de test doit créer une instance Testbed et l'activer. Il doit également initialiser les simulations de service appropriées, en utilisant init_datastore_v3_stub et init_memcache_stub dans ce cas précis. Les méthodes d'initialisation d'autres simulations de service App Engine sont répertoriées dans la section Présentation des utilitaires de test 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()

La méthode init_datastore_v3_stub() sans argument utilise un datastore en mémoire initialement vide. Si vous souhaitez tester une entité de datastore existante, ajoutez son chemin d'accès en tant qu'argument à init_datastore_v3_stub().

En plus de setUp(), ajoutez une méthode tearDown() qui va désactiver le module testbed. Ceci permet de rétablir les simulations d'origine, de sorte que les tests n'interfèrent pas entre eux.

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

Puis mettez en œuvre les tests.

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

Vous pouvez maintenant vous servir de TestModel pour écrire des tests qui utilisent les simulations de service de datastore ou Memcache au lieu des services réels.

Par exemple, la méthode présentée ci-dessous crée deux entités : la première utilise la valeur par défaut pour l'attribut number (42) et la seconde utilise une valeur autre que la valeur par défaut pour number (17). La méthode construit ensuite une requête pour les entités TestModel, mais seulement pour celles dont l'attribut number utilise la valeur par défaut.

Après avoir récupéré toutes les entités correspondantes, la méthode vérifie qu'une seule entité a été trouvée et que la valeur de l'attribut number de cette entité est la valeur par défaut.

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)

Autre exemple, la méthode suivante crée une entité et la récupère à l'aide de la fonction GetEntityViaMemcache(), que nous avons créée ci-dessus. La méthode vérifie ensuite qu'une entité a été affichée et que la valeur de son attribut number est identique à celle de l'entité créée précédemment.

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)

Enfin, appelez unittest.main().

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

Pour exécuter les tests, consultez la section Exécuter des tests.

Écrire des tests pour Cloud Datastore

Si votre application utilise Cloud Datastore, vous souhaitez peut-être écrire des tests permettant de vérifier le comportement de votre application en cas de cohérence à terme. Les options proposées par db.testbed vont alors s'avérer très utiles :

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 vous permet de contrôler la probabilité d'une application en écriture avant chaque requête globale (non ascendante). En définissant la probabilité sur 0 %, nous indiquons à la simulation du datastore de fonctionner avec la quantité maximale de cohérence à terme. Une cohérence à terme maximale signifie que les écritures seront validées, mais que leur application échouera systématiquement, si bien que les requêtes globales (non ascendantes) ne parviendront jamais à voir les modifications. Cela n'est évidemment pas représentatif de la quantité de cohérence à terme que votre application verra lors de son exécution dans l'environnement de production mais, à des fins de test, pouvoir configurer le datastore local pour qu'il se comporte à chaque fois de cette manière s'avère particulièrement utile. Si vous utilisez une probabilité non nulle, PseudoRandomHRConsistencyPolicy réalise une séquence déterministe de décisions de cohérence, afin que les résultats des tests soient cohérents :

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

Les API de test permettent de vérifier que votre application se comporte correctement face à la cohérence à terme. Toutefois, n'oubliez pas que le modèle local de cohérence de lecture avec réplication avancée est une approximation du modèle dans l'environnement de production, et non une réplique exacte. Dans l'environnement local, l'exécution d'une opération get() d'une entité Entity, appartenant à un groupe d'entités avec une écriture non appliquée, rendra toujours les résultats de l'écriture non appliquée visibles pour les requêtes globales suivantes, ce qui n'est pas le cas dans l'environnement de production.

Écrire des tests de messagerie

Vous pouvez utiliser la simulation de service de messagerie pour tester le service de messagerie. Comme pour les autres services compatibles avec testbed, vous devez d'abord initialiser la simulation, puis appeler le code utilisant l'API de messagerie et enfin vérifier si les bons messages ont été envoyés.

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)

Écrire des tests de la file d'attente de tâches

Vous pouvez utiliser la simulation de file d'attente de tâches pour écrire des tests qui utilisent le service taskqueue. Comme pour les autres services compatibles avec testbed, vous devez d'abord initialiser la simulation, puis appeler le code qui utilise l'API Task Queue et enfin vérifier si les tâches ont été correctement ajoutées à la file d'attente.

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

Définir le fichier de configuration queue.yaml

Si vous souhaitez exécuter des tests sur du code qui interagit avec une file d'attente autre que celle par défaut, vous devez créer et spécifier un fichier queue.yaml qui sera utilisé par votre application. Vous trouverez ci-dessous un exemple de fichier queue.yaml.

Pour plus d'informations sur les options queue.yaml disponibles, consultez la page Configuration de la file d'attente de tâches.

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

L'emplacement du fichier queue.yaml est spécifié lors de l'initialisation de la simulation :

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

Dans cet exemple, queue.yaml se trouve dans le même répertoire que les tests. S'il se trouvait dans un autre dossier, ce chemin devrait être spécifié dans root_path.

Filtrer des tâches

La simulation de la file d'attente de tâches get_filtered_tasks vous permet de filtrer les tâches en file d'attente. Cela facilite l'écriture des tests dont l'objectif est de vérifier le code qui place plusieurs tâches en file d'attente.

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)

Écrire des tests de tâches différées

Si le code d'application utilise la bibliothèque différée, vous pouvez avoir recours à la simulation de file d'attente de tâches avec deferred pour vérifier que les fonctions différées sont mises en file d'attente et exécutées correctement.

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)

Modifier les variables d'environnement par défaut

Les services App Engine dépendent souvent de variables d'environnement. La méthode activate() de la classe testbed..Testbed utilise des valeurs par défaut pour celles-ci, mais vous pouvez définir des valeurs personnalisées en fonction de vos besoins en matière de test avec la méthode setup_env de la classe testbed.Testbed.

Par exemple, supposons qu'un test stocke plusieurs entités dans le datastore, toutes liées au même ID d'application. Vous souhaitez ré-exécuter les mêmes tests, mais en utilisant un ID d'application différent de celui lié aux entités stockées. Pour cela, transmettez la nouvelle valeur à self.setup_env() en tant que app_id.

Exemple :

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

Simuler la connexion

La méthode setup_env est également fréquemment utilisée pour simuler un utilisateur connecté, avec ou sans privilèges d'administrateur, pour vérifier si vos gestionnaires fonctionnent correctement dans chaque cas.

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

Désormais, vos méthodes de test peuvent appeler, par exemple, self.loginUser('', '') pour simuler l'absence d'utilisateur connecté, self.loginUser('test@example.com', '123') pour simuler un utilisateur non administrateur connecté, ou bien self.loginUser('test@example.com', '123', is_admin=True) pour simuler un utilisateur administrateur connecté.

Configurer un framework de test

Les utilitaires de test du SDK ne sont associés à aucun framework donné. Vous pouvez exécuter vos tests unitaires avec n'importe quel lanceur de tests App Engine disponible, par exemple nose-gae ou ferrisnose. Vous pouvez également écrire un lanceur de test simple ou utiliser celui illustré ci-dessous.

Les scripts suivants utilisent le module Python unittest.

Vous pouvez donner le nom que vous voulez au script. Lorsque vous l'exécutez, indiquez le chemin d'accès à votre installation du SDK Google Cloud ou du SDK Google App Engine, ainsi que le chemin d'accès à vos modules de test. Le script va détecter tous les tests dans le chemin fourni et va imprimer les résultats dans le flux d'erreur standard. Les fichiers de test suivent la convention voulant que leur nom contienne le préfixe test.

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

Exécuter les tests

Vous pouvez exécuter ces tests simplement en exécutant le script runner.py, qui est décrit en détail dans la section Configurer un framework de test :

python runner.py <path-to-appengine-or-gcloud-SDK> .
Cette page vous a-t-elle été utile ? Évaluez-la :

Envoyer des commentaires concernant…

Environnement standard App Engine pour Python 2