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. Dies ermöglicht es Ihnen, kleine, verwaltbare und wiederverwendbare Codeeinheiten zu entwickeln. Außerdem können Sie Ihren Code sorgfältiger und schneller testen.

Lokale Einheitentests führen Sie in Ihrer eigenen Entwicklungsumgebung und ohne Einbeziehung von Remotekomponenten 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 gestellt werden. Eine Entität, die während eines Datenspeicher-Einheitentests gespeichert wird, verbleibt im Arbeitsspeicher und 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-Testdienstprogramme

Mit dem App Engine-Python-Modul testbed werden Dienst-Stubs für Einheitentests zur Verfügung gestellt.

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

  • App Identity init_app_identity_stub
  • Blobstore (mit init_blobstore_stub)
  • Capability (init_capability_stub)
  • Datenspeicher (mit init_datastore_v3_stub)
  • Dateien (mit init_files_stub)
  • Bilder (nur für dev_appserver; mit init_images_stub)
  • LogService (mit init_logservice_stub)
  • Mail (mit init_mail_stub)
  • Memcache (mit init_memcache_stub)
  • Aufgabenwarteschlange (mit init_taskqueue_stub)
  • URL-Abruf (mit init_urlfetch_stub)
  • Nutzerdienst (mit init_user_stub)

Mit init_all_stubs können Sie alle Stubs gleichzeitig initialisieren.

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.

Prüfen Sie, ob der Test-Runner über die entsprechenden Bibliotheken im Ladepfad von Python verfügt. Hierzu zählen die App Engine-Bibliotheken, yaml (im App Engine-SDK enthalten), das Stammverzeichnis der Anwendung sowie jegliche voraussichtlich durch Anwendungscode am Bibliothekspfad vorgenommenen Änderungen (z. B. ein lokales Verzeichnis ./lib, 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-Modul unittest und die für die zu testenden Dienste relevanten App Engine-Module – in diesem Fall memcache und ndb, die sowohl den Datenspeicher als auch Memcache verwenden. Importieren Sie auch das Modul testbed.

import unittest

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

Erstellen Sie dann 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 davon, welche Dienste Sie testen, muss mit dem Testfall eine Testbed-Instanz erstellt und aktiviert werden. Mit dem Testfall müssen auch die relevanten Dienst-Stubs initialisiert werden, in unserem Beispiel 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()

Die argumentlose Methode init_datastore_v3_stub() verwendet einen Datenspeicher im Arbeitsspeicher, der anfänglich leer ist. Wenn Sie eine vorhandene Datenspeicherentität testen möchten, schließen Sie Ihren Pfadnamen als Argument in init_datastore_v3_stub() ein.

Zum Deaktivieren des testbed-Moduls schließen Sie neben setUp() zusätzlich die Methode tearDown() ein. 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)))

Jetzt können Sie mithilfe von TestModel Tests schreiben, bei denen anstelle der tatsächlichen Dienste die Dienst-Stubs für den Datenspeicher oder Memcache verwendet werden.

Mit der nachfolgend dargestellten Methode werden beispielsweise zwei Entitäten erstellt: Die erste Entität verwendet den Standardwert für das Attribut number (42) und die zweite Entität einen nicht standardmäßigen Wert für number (17). Anschließend wird im Rahmen der Methode eine Abfrage für TestModel-Entitäten erstellt, jedoch nur für jene mit dem Standardwert für number.

Nachdem alle übereinstimmenden Entitäten abgerufen wurden, wird geprüft, ob tatsächlich nur eine Entität gefunden wurde und der Wert des Attributs number der Entität der Standardwert ist.

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)

In einem weiteren Beispiel wird mit der folgenden Methode eine Entität erstellt und mithilfe der oben erstellten Funktion GetEntityViaMemcache() abgerufen. Anschließend wird im Rahmen der Methode geprüft, ob eine Entität zurückgegeben wurde und deren Wert number mit dem Wert für die zuvor erstellte Entität identisch ist.

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. db.testbed enthält Optionen, die dies erleichtern:

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 vor jeder globalen (Nicht-Ancestor-)Abfrage steuern, mit welcher Wahrscheinlichkeit ein Schreibvorgang ausgeführt 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. Bei einer Wahrscheinlichkeit höher als Null wird mit PseudoRandomHRConsistencyPolicy eine deterministische Folge von Konsistenzentscheidungen für konsistente Testergebnisse getroffen:

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 eine Entity, die zu einer Entitätsgruppe 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 Code testen möchten, der mit einer nicht standardmäßigen Warteschlange interagiert, müssen Sie für die zu verwendende Anwendung die Datei queue.yaml erstellen und angeben. Nachfolgend sehen Sie ein Beispiel der Datei 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

Der Speicherort von queue.yaml wird beim Initialisieren des Stubs angegeben:

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

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

Aufgaben filtern

Mit dem Befehl get_filtered_tasks des taskqueue-Stubs 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 im Anwendungscode die deferred-Bibliothek verwendet wird, können Sie den taskqueue-Stub zusammen mit deferred verwenden, um zu verifizieren, dass deferred-Funktionen korrekt in die Warteschlange gestellt und 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. Mit der Methode activate() der Klasse testbed.Testbed werden hierfür Standardwerte verwendet. Sie können jedoch entsprechend Ihren Testanforderungen auch benutzerdefinierte Werte festlegen. Dazu verwenden Sie die Methode setup_env der Klasse testbed.Testbed.

Beispiel: Durch Ihren Test werden mehrere Entitäten im Datenspeicher gespeichert, die alle mit derselben Anwendungs-ID verknüpft sind. Jetzt möchten Sie die gleichen Tests noch einmal ausführen, jedoch mit einer anderen Anwendungs-ID als der, die mit den gespeicherten Entitäten verknüpft ist. Übergeben Sie den neuen Wert in diesem Fall an self.setup_env() als 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

Mithilfe von setup_env wird oft auch ein angemeldeter Nutzer mit oder ohne Administratorrechten simuliert. Sie überprü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())

Sie können mit Ihren Testmethoden jetzt 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 SDK- 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. Den Namen von Testdateien wird standardmäßig das Präfix test vorangestellt.

"""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 die Tests einfach mithilfe des Skripts runner.py ausführen. Detaillierte Informationen hierzu finden Sie unter Test-Framework einrichten:

python runner.py <path-to-appengine-or-gcloud-SDK> .
Hat Ihnen diese Seite weitergeholfen? Teilen Sie uns Ihr Feedback mit:

Feedback geben zu...

App Engine-Standardumgebung für Python 2