# Local Unit Testing for Python - Google App Engine - Google Code

English ▼
Skip to page content Skip to main navigation
e.g. "templates" or "datastore"

# Local Unit Testing for Python

Unit testing allows you to check the quality of your code after you've written it, but you can also use unit testing to improve your development process as you go along. Instead of writing tests after you finish developing your application, consider writing the tests as you go. This helps you design small, maintainable, reusable units of code. It also makes it easier for you to test your code thoroughly and quickly.

When you do local unit testing, you run tests that stay inside your own development environment without involving remote components. App Engine provides testing utilities that use local implementations of datastore and other App Engine services. This means you can exercise your code's use of these services locally, without deploying your code to App Engine, by using service stubs.

A service stub is a method that simulates the behavior of the service. For example, the datastore service stub shown in Writing Datastore and Memcache Tests allows you to test your datastore code without making any requests to the real datastore. Any entity stored during a datastore unit test is held in memory, not in the datastore, and is deleted after the test run. You can run small, fast tests without any dependency on datastore itself.

This document describes how to write unit tests against several local App Engine services, then gives some information about setting up a testing framework.

- Introducing the Python Testing Utilities
- Writing Datastore and Memcache Tests
- Writing High Replication Datastore Tests
- Writing Mail Tests
- Changing the Default Environment Variables
- Setting Up a Testing Framework

## Introducing the Python Testing Utilities

An App Engine Python module called `testbed` makes service stubs available for unit testing. The `testbed` module was inspired by GAE Testbed, which was designed by JJ Geewax.

To write a test that uses `testbed`, you need to create and activate a `Testbed` instance, then declare the service stubs that you want to use. For an example, see Writing Datastore and Memcache Tests.

Service stubs are available for the following services:

- Blobstore (use `init_blobstore_stub`)
- Datastore (use `init_datastore_v3_stub`)
- Memcache (use `init_memcache_stub`)
- Images (only for dev_appserver; use `init_images_stub`)
- Mail (use `init_mail_stub`)
- Task Queue (use `init_taskqueue_stub`)
- URL fetch (use `init_urlfetch_stub`)
- User service (use `init_user_stub`)
- XMPP (use `init_xmpp_stub`)

Note: In some cases, service stubs behave differently than the actual services. Even if tests pass, your code could still fail when it runs in a production environment.

## Writing Datastore and Memcache Tests

This section shows an example of how to write code that tests the use of the datastore and memcache services.

First, import the Python `unittest` module and the App Engine modules that are relevant to the services being tested—in this case `memcache` and `db`, which is for the datastore. Also import the App Engine `testbed` module.

    import unittest
    from google.appengine.api import memcache
    from google.appengine.ext import db
    from google.appengine.ext import testbed

Then create a `TestModel` class. In this example, a function checks to see whether an entity is stored in memcache. If no entity is found, it checks for an entity in the datastore.

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

    class TestEntityGroupRoot(db.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
      entity = TestModel.get(entity_key)
      if entity is not None:
        memcache.set(entity_key, entity)
      return entity

Next, create a test case. No matter what services you are testing, the test case must create a `Testbed` instance and activate it. The test case must also initialize the relevant service stubs, in this case using `init_datastore_v3_stub` and `init_memcache_stub`. (The methods for initializing other App Engine service stubs are listed in Introducing the Python Testing Utilities.)

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

The `init_datastore_v3_stub()` method with no argument uses an in-memory datastore that is initially empty. If you want to test an existing datastore entity, include its pathname as an argument to `init_datastore_v3_stub()`.

Note: The `setUp()` method uses a default application ID for test runs. The test can only access entities in the datastore that have that same application ID. To change the application ID of a test, use `setup_env()`, as described in Changing the Default Environment Variables.

In addition to `setUp()`, include a `tearDown()` method that deactivates the testbed. This restores the original stubs so that tests do not interfere with each other.

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

Then implement the tests.

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

Now you can use `TestModel` to write tests that use the datastore or memcache service stubs instead of using the real services.

For example, the method shown below creates two entities: the first entity uses the default value for the `number` attribute (42), and the second uses a nondefault value for `number` (17). The method then builds a query for `TestModel` entities, but only for those with the default value of `number`.

After retrieving all matching entities, the method tests that exactly one entity was found, and that the `number` attribute value of that entity is the default value.

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

As another example, the following method creates an entity and retrieves it using the `GetEntityViaMemcache()` function that we created above. The method then tests that an entity was returned, and that its `number` value is the same as for the the previously created entity.

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

And finally, invoke `unittest.main()`.

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

## Writing High Replication Datastore Tests

If your app uses the High Replication Datastore (HRD), you may want to write tests that verify your application's behavior in the face of eventual consistency. `db.testbed` exposes options that make this easy:

    import unittest
    from google.appengine.ext import db
    from google.appengine.ext import testbed
    from google.appengine.datastore import datastore_stub_util

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

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

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

        user_key = db.Key.from_path('User', 'ryan')
        # Put two entities
        db.put([TestModel(parent=user_key), TestModel(parent=user_key)])

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

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

The `PseudoRandomHRConsistencyPolicy` class lets you control the likelihood of a write applying before each global (non-ancestor) query. By setting the probability to 0%, we are instructing the datastore stub to operate with the maximum amount of eventual consistency. Maximum eventual consistency means writes will commit but always fail to apply, so global (non-ancestor) queries will consistently fail to see changes. This is of course not representative of the amount of eventual consistency your application will see when running in production, but for testing purposes, it's very useful to be able to configure the local datastore to behave this way every time. If you use a non-zero probability, `PseudoRandomHRConsistencyPolicy` makes a deterministic sequence of consistency decisions so test outcomes are consistent:

    class DemoTestCase(unittest.TestCase):

      # ...

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

        class TestModel(db.Model):
          pass

        TestModel().put()

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

The testing APIs are useful for verifying that your application behaves properly in the face of eventual consistency, but please keep in mind that the local High Replication read consistency model is an approximation of the production High Replication read consistency model, not an exact replica. In the local environment, performing a get() of an `Entity` that belongs to an entity group with an unapplied write will always make the results of the unapplied write visible to subsequent global queries. In production this is not the case.

## Writing Mail Tests

You can use the mail service stub to test the mail service. Similar to other services supported by testbed, at first you initialize the stub, then invoke the code which uses the mail API, and finally test whether the correct messages were sent.

    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)

    unittest.main()

## Changing the Default Environment Variables

App Engine services often depend on environment variables. The `setUp()` method within `testbed` uses default values for these, but you can set custom values based on your testing needs.

For example, let's say you have a test that stores several entities in datastore, all of them linked to the same application ID. Now you want to run the same tests again, but using an application ID different from the one that is linked to the stored entities. To do this, pass the new value into `self.setup_env()` as `app_id`.

For example:

      def setUp(self):
        self.testbed = testbed.Testbed()
        self.testbed.activate()
        self.testbed.setup_env(app_id=your_app_id)
        self.testbed.init_datastore_v3_stub()

## Setting Up a Testing Framework

The SDK's testing utilities are not tied to a specific framework. You can run your unit tests with any available App Engine testrunner, for example nose-gae or gaeunit. You can also write a simple testrunner of your own, or use the one shown below.

The following script is a basic testrunner that you can copy and paste into a file. Before you use it, install the Python unittest2 module.

You can name the script anything you want. When you run it, provide the path to your App Engine SDK installation and the path to your test modules. The script runs the tests and prints results to the standard error stream.

    #!/usr/bin/python
    import optparse
    import sys
    # Install the Python unittest2 package before you run this script.
    import unittest2

    USAGE = """%prog SDK_PATH TEST_PATH
    Run unit tests for App Engine apps.

    SDK_PATH    Path to the SDK installation
    TEST_PATH   Path to package containing test modules"""

    def main(sdk_path, test_path):
        sys.path.insert(0, sdk_path)
        import dev_appserver
        dev_appserver.fix_sys_path()
        suite = unittest2.loader.TestLoader().discover(test_path)
        unittest2.TextTestRunner(verbosity=2).run(suite)

    if __name__ == '__main__':
        parser = optparse.OptionParser(USAGE)
        options, args = parser.parse_args()
        if len(args) != 2:
            print 'Error: Exactly 2 arguments required.'
            parser.print_help()
            sys.exit(1)
        SDK_PATH = args[0]
        TEST_PATH = args[1]
        main(SDK_PATH, TEST_PATH)