Lokale Einheitentests für Python 2

Mit Einheitentests können Sie die Qualität Ihres Codes überprüfen, nachdem Sie ihn geschrieben haben. Sie können mit diesen Tests aber auch Ihren Entwicklungsprozess verbessern. Anstatt die Tests nach der Entwicklung der Anwendung zu schreiben, sollten Sie sie besser währenddessen schreiben. So können Sie kleine, wartbare, wiederverwendbare Codeeinheiten entwickeln. Außerdem können Sie Ihren Code dadurch sorgfältiger und schneller testen.

Lokale Einheitentests führen Sie in Ihrer eigenen Entwicklungsumgebung und ohne Einbeziehung von Remote-Komponenten aus. App Engine stellt Testdienstprogramme bereit, die lokale Implementierungen von Datastore und anderen App Engine-Diensten verwenden. Mithilfe von Dienst-Stubs können Sie lokal testen, wie Ihr Code diese Dienste nutzt, ohne den Code in App Engine bereitzustellen.

Ein Dienst-Stub ist eine Methode, die das Verhalten des Diensts simuliert. Mit dem Datenspeicherdienst-Stub, der unter Datenspeicher- und Memcache-Tests schreiben dargestellt wird, können Sie beispielsweise Ihren Datenspeichercode testen, ohne dass Anfragen an den realen Datenspeicher gesendet werden. Eine Entität, die während eines Datenspeicher-Einheitentests gespeichert wird, verbleibt im Arbeitsspeicher, nicht im Datenspeicher, und wird im Anschluss an den Test gelöscht. Sie können kurze, schnelle Tests unabhängig vom Datenspeicher ausführen.

In diesem Dokument wird erläutert, wie Sie Einheitentests für verschiedene lokale App Engine-Dienste schreiben. Außerdem erfahren Sie, wie Sie ein Test-Framework einrichten.

Einführung in die Python 2-Testdienstprogramme

Ein App Engine-Python-Modul namens testbed stellt Dienst-Stubs zum Testen von Komponenten zur Verfügung.

Dienst-Stubs stehen für folgende Dienste zur Verfügung:

  • Anwendungsidentität init_app_identity_stub
  • Blobstore (init_blobstore_stub verwenden)
  • Funktion (init_capability_stub verwenden)
  • Datastore (init_datastore_v3_stub verwenden)
  • Dateien (init_files_stub verwenden)
  • Images (nur für dev_appserver; init_images_stub verwenden)
  • LogService (init_logservice_stub verwenden)
  • Mail (init_mail_stub verwenden)
  • Memcache (init_memcache_stub verwenden)
  • Aufgabenwarteschlange (init_taskqueue_stub verwenden)
  • URL Fetch (init_urlfetch_stub verwenden)
  • Nutzerdienst (init_user_stub verwenden)

Initialisieren Sie alle Stubs gleichzeitig mit init_all_stubs.

Datenspeicher- und Memcache-Tests schreiben

In diesem Abschnitt wird anhand eines Beispiels erläutert, wie Sie Code schreiben, mit dem Sie die Verwendung von Datenspeicher- und Memcache-Diensten testen.

Achten Sie darauf, dass Ihr Test-Runner die entsprechenden Bibliotheken im Ladepfad von Python hat, einschließlich der App Engine-Bibliotheken, yaml im App Engine-SDK, des Anwendungsstamms und anderer vom Anwendungscode erwarteter Änderungen am Bibliothekspfad (beispielsweise ein lokales ./lib-Verzeichnis, falls vorhanden). Beispiel:

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

Importieren Sie das Python-unittest-Modul und die App Engine-Module, die für die getesteten Dienste relevant sind, in diesem Fall memcache und ndb, die sowohl den Datenspeicher als auch Memcache verwenden. Importieren Sie auch das testbed-Modul.

import unittest

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

Erstellen Sie anschließend eine TestModel-Klasse. In diesem Beispiel wird von einer Funktion überprüft, ob in Memcache eine Entität gespeichert ist. Wenn keine Entität gefunden wird, prüft die Funktion den Datenspeicher hinsichtlich einer vorhandenen Entität. In der Praxis ist dies häufig überflüssig, da ndb Memcache im Hintergrund verwendet. Für einen Test ist es jedoch ein geeignetes Muster.

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

Erstellen Sie als Nächstes einen Testfall. Unabhängig von den zu testenden Diensten muss der Testfall eine Testbed-Instanz erstellen und aktivieren. Der Testfall muss außerdem die relevanten Dienst-Stubs initialisieren, in diesem Fall mit init_datastore_v3_stub und init_memcache_stub. Die Methoden zum Initialisieren anderer App Engine-Dienst-Stubs finden Sie unter Einführung in die Python-Testdienstprogramme.

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

Bei der init_datastore_v3_stub()-Methode ohne Argument wird ein arbeitsspeicherinterner Datenspeicher verwendet, der anfangs leer ist. Wenn Sie eine bestehende Datenspeicher-Entität testen möchten, müssen Sie ihren Pfadnamen als Argument für init_datastore_v3_stub() aufnehmen.

Nehmen Sie zusätzlich zu setUp() auch die Methode tearDown() auf, die die Prüfumgebung deaktiviert. Dadurch werden die ursprünglichen Stubs wiederhergestellt, sodass die Tests sich nicht gegenseitig behindern.

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

Implementieren Sie anschließend die Tests.

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

Nun können Sie mit TestModel Tests schreiben, die anstelle der eigentlichen Dienste die Dienst-Stubs für den Datenspeicher oder für Memcache verwenden.

Bei der unten angezeigten Methode werden beispielsweise zwei Entitäten erstellt: Die erste Entität verwendet den Standardwert für das number-Attribut (42), das zweite verwendet einen Nicht-Standardwert für number (17). Anschließend erstellt die Methode eine Anfrage für TestModel-Entitäten, jedoch nur für diejenigen mit dem Standardwert für number.

Nach dem Abrufen aller übereinstimmenden Entitäten testet die Methode, ob genau eine Entität gefunden wurde und ob es sich bei dem Wert des Attributs number dieser Entität um den Standardwert handelt.

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)

Hier ein weiteres Beispiel: Die folgende Methode erstellt eine Entität und ruft sie mit der oben erstellten GetEntityViaMemcache()-Funktion ab. Die Methode testet dann, ob eine Entität zurückgegeben wurde und der Wert number derselbe ist wie für die zuvor erstellte Entität.

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)

Rufen Sie abschließend unittest.main() auf.

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

Wie Sie Tests ausführen, erfahren Sie unter Tests ausführen.

Cloud Datastore-Tests schreiben

Wenn Ihre Anwendung Cloud Datastore verwendet, können Sie Tests schreiben, mit denen das Verhalten Ihrer Anwendung im Hinblick auf Eventual Consistency überprüft wird. Von db.testbed werden Optionen bereitgestellt, die dies vereinfachen:

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

Mit der Klasse PseudoRandomHRConsistencyPolicy können Sie die Wahrscheinlichkeit steuern, mit der vor jeder globalen Anfrage (ohne Ancestors) ein Schreibvorgang übernommen wird. Wenn Sie die Wahrscheinlichkeit auf 0 % setzen, weisen Sie den Datenspeicher-Stub an, maximale Eventual Consistency anzuwenden. Maximale Eventual Consistency bedeutet, dass für Schreibvorgänge ein Commit durchgeführt wird, aber bei ihrer Anwendung immer ein Fehler auftritt. Aus der Sicht globaler (Nicht-Ancestor-)Anfragen treten somit konsistent keinerlei Änderungen auf. Dies ist nicht für das Maß an Eventual Consistency repräsentativ, das während der Ausführung der Anwendung in der Produktion auftritt. Für Testzwecke ist es jedoch sehr hilfreich, den lokalen Datenspeicher so konfigurieren zu können, dass er sich jedes Mal gleich verhält. Wenn Sie eine Wahrscheinlichkeit ungleich 0 verwenden, nimmt PseudoRandomHRConsistencyPolicy eine deterministische Abfolge von Konsistenzentscheidungen vor, sodass die Testergebnisse konsistent sind:

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

Mithilfe der Test-APIs können Sie überprüfen, ob sich Ihre Anwendung in Bezug auf Eventual Consistency ordnungsgemäß verhält. Beachten Sie jedoch, dass das lokale High Replication-Lesekonsistenzmodell eine Annäherung an das in der Produktion verwendete High Replication-Lesekonsistenzmodell und kein genaues Replikat ist. In der lokalen Umgebung führt die Ausführung von get() für Entity, die zu einer Entitätengruppe mit einem nicht angewendeten Schreibvorgang gehört, immer dazu, dass die Ergebnisse des nicht angewendeten Schreibvorgangs für nachfolgende globale Abfragen sichtbar werden. In der Produktion ist dies nicht der Fall.

Mail-Tests schreiben

Mit dem E-Mail-Dienst-Stub können Sie den E-Mail-Dienst testen. Ähnlich wie bei anderen von testbed unterstützten Diensten initialisieren Sie zuerst den Stub. Dann rufen Sie den Code auf, der die Mail API verwendet, und testen schließlich, ob die richtigen Nachrichten gesendet wurden.

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)

Aufgabenwarteschlangen-Tests schreiben

Mit dem taskqueue-Stub können Sie Tests schreiben, die den taskqueue-Dienst verwenden. Ähnlich wie bei anderen von testbed unterstützten Diensten initialisieren Sie zuerst den Stub. Dann rufen Sie den Code auf, der die Taskqueue API verwendet, und testen schließlich, ob die Aufgaben der Warteschlange ordnungsgemäß hinzugefügt wurden.

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

Konfigurationsdatei queue.yaml festlegen

Wenn Sie Tests für Code ausführen möchten, der mit einer nicht standardmäßigen Warteschlange interagiert, müssen Sie eine queue.yaml-Datei für Ihre Anwendung erstellen und angeben. Ein Beispiel für queue.yaml:

Weitere Informationen zu den für queue.yaml verfügbaren Optionen finden Sie unter Konfiguration der Aufgabenwarteschlange.

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

Die Position von queue.yaml wird beim Initialisieren des Stubs angegeben:

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

Im Beispiel befindet sich queue.yaml in demselben Verzeichnis wie die Tests. Wenn es in einem anderen Ordner wäre, müsste dieser Pfad in root_path angegeben werden.

Aufgaben filtern

Mit den taskqueue-Stub-get_filtered_tasks können Sie Aufgaben in der Warteschlange filtern. Dies erleichtert das Schreiben von Tests zum Überprüfen von Code, mit dem mehrere Aufgaben in die Warteschlange gestellt werden.

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)

Tests für zurückgestellte Aufgaben schreiben

Wenn Ihr Anwendungscode die zurückgestellte Bibliothek verwendet, können Sie den taskqueue-Stub zusammen mit deferred verwenden, um zu kontrollieren, ob zurückgestellte Funktionen in die Warteschlange gestellt und korrekt ausgeführt werden.

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)

Standardumgebungsvariablen ändern

App Engine-Dienste sind oft von Umgebungsvariablen abhängig. Die Methode activate() der Klasse testbed.Testbed verwendet dafür Standardwerte. Sie können jedoch mit den Werten benutzerdefinierte Werte entsprechend Ihren Testanforderungen mit der Methode setup_env der Klasse testbed.Testbed festlegen.

Nehmen wir beispielsweise an, Sie verwenden einen Test, bei dem mehrere Entitäten im Datenspeicher gespeichert werden, die alle mit derselben Anwendungs-ID verknüpft sind. Nun möchten Sie dieselben Tests noch einmal ausführen, diesmal jedoch mit einer anderen Anwendungs-ID als der, die mit den gespeicherten Entitäten verknüpft ist. Übergeben Sie dazu den neuen Wert als self.setup_env() an app_id.

Beispiel:

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

Anmeldung simulieren

Mit setup_env wird oft auch ein angemeldeter Nutzer mit oder ohne Administratorrechten simuliert. Sie prüfen damit, ob Ihre Handler in jedem Fall ordnungsgemäß funktionieren.

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

Nun können Ihre Testmethoden beispielsweise self.loginUser('', '') aufrufen, um zu simulieren, dass kein Nutzer angemeldet ist. Mit self.loginUser('test@example.com', '123') simulieren Sie, dass ein Nutzer ohne Administratorrechte angemeldet ist, und mit self.loginUser('test@example.com', '123', is_admin=True), dass ein Administrator angemeldet ist.

Test-Framework einrichten

Die Testdienstprogramme des SDK sind nicht an ein bestimmtes Framework gebunden. Sie können Ihre Einheitentests mit jedem verfügbaren App Engine-Test-Runner wie etwa nose-gae oder ferrisnose ausführen. Sie können auch selbst einen einfachen Test-Runner schreiben oder den unten angezeigten Test-Runner verwenden.

Für die folgenden Skripts wird das Einheitentestmodul von Python verwendet.

Sie können dem Skript jeden beliebigen Namen geben. Geben Sie bei der Ausführung den Pfad zu Ihrer Google Cloud CLI- oder Google App Engine SDK-Installation sowie den Pfad zu Ihren Testmodulen an. Das Skript erkennt alle Tests im angegebenen Pfad und gibt die Ergebnisse an den Standardfehlerstream aus. Testdateien folgen der Konvention, dass test ihrem Namen vorangestellt ist.

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

Tests ausführen

Sie können diese Tests einfach mit dem Skript runner.py ausführen, das ausführlich unter Test-Framework einrichten beschrieben wird:

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