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 :

• Identité d'application init_app_identity_stub
• Blobstore (utiliser init_blobstore_stub)
• Capacité (utiliser init_capability_stub)
• Datastore (utiliser init_datastore_v3_stub)
• Fichiers (utiliser init_files_stub)
• Images (uniquement pour dev_appserver ; utiliser init_images_stub)
• LogService (utiliser init_logservice_stub)
• Messagerie (utiliser init_mail_stub)
• Memcache (utiliser init_memcache_stub)
• File d'attente de tâches (utiliser init_taskqueue_stub)
• Récupération d'URL (utiliser init_urlfetch_stub)
• Service Users (utiliser init_user_stub)

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

Remarque : Dans certains cas, les simulations de service se comportent différemment des services réels. Même si les tests sont concluants, votre code peut toujours échouer lorsqu'il s'exécute dans un environnement de production.

Écrire des tests pour le datastore et Memcache

Cette section indique comment écrire du code permettant de tester l'utilisation des services 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, yaml (inclus dans le SDK App Engine), la racine de l'application et toute autre modification du chemin de la bibliothèque attendue par le code d'application (comme 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 memcache et ndb, qui utilisent à la fois 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. Cela peut souvent être redondant dans la vie réelle, car ndb utilise Memcache lui-même en arrière-plan, mais cela reste un modèle acceptable pour 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 testés, le scénario de test doit créer une instance Testbed et l'activer. Le scénario de test doit également initialiser les simulations de service pertinentes en utilisant, dans le cas présent, init_datastore_v3_stub et init_memcache_stub. 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 voulez tester une entité existante dans le datastore, ajoutez son chemin d'accès en tant qu'argument à init_datastore_v3_stub().

Remarque : La méthode setUp() utilise un ID de projet par défaut pour les tests. Le test ne peut accéder qu'aux entités du datastore ayant le même ID de projet. Pour modifier l'ID de projet d'un test, utilisez setup_env(), comme décrit dans la section Modifier les variables d'environnement par défaut.

En plus de setUp(), ajoutez une méthode tearDown() qui désactive 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)))

Désormais, vous pouvez utiliser TestModel pour écrire des tests qui font appel aux simulations de service Datastore ou Memcache au lieu de faire appel aux services réels.

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

Une fois toutes les entités correspondantes récupérées, la méthode vérifie qu'une entité a été trouvée et que la valeur d'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() créée précédemment. La méthode vérifie ensuite qu'une entité a été renvoyée et que sa valeur number est identique à celle de l'entité précédemment créée.

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. db.testbed propose des options qui simplifient cette tâche :

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é pour qu'une écriture soit appliquée 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 établit une séquence déterministe de décisions de cohérence de sorte 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='[e-mail]',
                       subject='This is a test',
                       sender='[e-mail]',
                       body='This is a test e-mail')
        messages = self.mail_stub.get_sent_messages(to='[e-mail]')
        self.assertEqual(1, len(messages))
        self.assertEqual('[e-mail]', messages[0].to)

Écrire des tests de file d'attente des 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 que votre application utilisera. Vous trouverez un exemple de fichier queue.yaml ci-dessous.

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 l'exemple, queue.yaml se trouve dans le même répertoire que les tests. S'il se trouve dans un autre dossier, ce chemin d'accès doit être spécifié dans root_path.

Filtrer des tâches

La méthode get_filtered_tasks de la simulation de file d'attente de tâches 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 de test avec la méthode setup_env de la classe testbed.Testbed.

Remarque : Assurez-vous que tout appel à setup_env se produit après l'appel à activate, afin que l'éventuelle instruction deactivate restaure correctement l'environnement d'origine. Il est important de dissocier vos tests unitaires.

Par exemple, prenons le cas d'un test qui stocke plusieurs entités dans un datastore, celles-ci étant toutes liées au même ID d'application. Maintenant vous voulez 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 ce faire, transmettez la nouvelle valeur dans 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='[e-mail]', 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(), '[e-mail]')
        self.loginUser(is_admin=True)
        self.assertTrue(users.is_current_user_admin())

Désormais, vos méthodes de test peuvent, par exemple, appeler self.loginUser('', '') pour simuler l'absence d'utilisateur connecté, self.loginUser('[e-mail]', '123') pour simuler la connexion d'un utilisateur non administrateur, self.loginUser('[e-mail]', '123', is_admin=True) pour simuler la connexion d'un utilisateur administrateur.

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 test 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 de Google Cloud CLI 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> .