__init__.py

import os
import unittest

from google.appengine.api import apiproxy_stub_map
from google.appengine.api import datastore_file_stub
from google.appengine.api import mail_stub
from google.appengine.api import user_service_stub
from google.appengine.api.images import images_stub
from google.appengine.api.labs.taskqueue import taskqueue_stub
from google.appengine.api.memcache import memcache_stub
from google.appengine.api.xmpp import xmpp_service_stub

import test_capabilities


class AppEngineTest(unittest.TestCase):
  def setUp(self, disabled_capabilities=None, disabled_methods=None):
    """Setup routine for App Engine test cases.
    
    Args:
      disabled_capabilities: A set of (package, capability) tuples defining
        capabilities that are disabled.
      disabled_methods: A set of (package, method) tuples defining methods that
        are disabled. An entry of ('package', '*') in disabled_capabilities is
        treated the same as finding the method being tested in this set.
    """
    # Set up a new set of stubs for each test
    self.stub_map = apiproxy_stub_map.APIProxyStubMap()
    apiproxy_stub_map.apiproxy = self.stub_map
    
    if disabled_capabilities:
      self.disabled_capabilities = disabled_capabilities
    else:
      self.disabled_capabilities = set()
    if disabled_methods:
      self.disabled_methods = disabled_methods
    else:
      self.disabled_methods = set()
    capability_stub = test_capabilities.CapabilityServiceStub(
        self.disabled_capabilities, self.disabled_methods)
    self.stub_map.ReigsterStub('capability_service', capability_stub)

  def _RegisterStub(self, service_name, stub):
    wrapped_stub = test_capabilities.CapabilityStubWrapper(stub,
        self.disabled_capabilities, self.disabled_methods)
    self.stub_map.RegisterStub(service_name, wrapped_stub)


class DatastoreTest(AppEngineTest):
  def setUp(self, datastore_file=None, history_file=None, require_indexes=False,
            **kwargs):
    super(DatastoreTest, self).setUp(**kwargs)
    stub = datastore_file_stub.DatastoreFileStub(
        os.environ['APPLICATION_ID'],
        datastore_file,
        history_file,
        require_indexes)
    self._RegisterStub('datastore_v3', stub)


class MemcacheTest(AppEngineTest):
  def setUp(self, **kwargs):
    super(MemcacheTest, self).setUp(**kwargs)
    stub = memcache_stub.MemcacheServiceStub()
    self._RegisterStub('memcache', stub)


class UsersTest(AppEngineTest):
  def setUp(self, user_email=None, user_is_admin=False, **kwargs):
    super(UsersTest, self).setUp(**kwargs)
    stub = user_service_stub.UserServiceStub()
    self._RegisterStub('user', stub)
    self.SetUser(user_email, user_is_admin)

  def SetUser(self, user_email, user_is_admin=False):
    os.environ['USER_EMAIL'] = user_email
    os.environ['USER_IS_ADMIN'] = user_is_admin

    
# TODO: Better test-oriented implementations of Mail, XMPP, URLFetch stubs


class MailTest(AppEngineTest):
  def setUp(self, **kwargs):
    super(MailTest, self).setUp(**kwargs)
    stub = mail_stub.MailServiceStub()
    self._RegisterStub('mail', stub)


class ImagesTest(AppEngineTest):
  def setUp(self, **kwargs):
    super(ImagesTest, self).setUp(**kwargs)
    stub = images_stub.ImagesServiceStub()
    self._RegisterStub('images', stub)


class XmppTest(AppEngineTest):
  def setUp(self, xmpp_log=logging.info, **kwargs):
    super(XmppTest, self).setUp(**kwargs)
    stub = xmpp_service_stub.XmppServiceStub(log=xmpp_log)
    self._RegisterStub('xmpp', stub)


class TaskQueueTest(AppEngineTest):
  def setUp(self, **kwargs):
    super(XmppTest, self).setUp(**kwargs)
    stub = taskqueue_stub.TaskQueueServiceStub()
    self._RegisterStub('taskqueue', self)


def main(app_id, auth_domain='gmail.com',
         server_software='Development/1.0 (AppEngineTest)'):
  os.environ['APPLICATION_ID'] = app_id
  os.environ['AUTH_DOMAIN'] = auth_domain
  os.environ['SERVER_SOFTWARE'] = server_software
  unittest.main()

README

A unit test helper library for App Engine.

Note that this is currently COMPLETELY UNTESTED. Consider it demo code only.

This library aims to make it easier to unit-test app engine apps and libraries
by handling the creation and registration of service stubs and so forth for you.
It also provides a custom implementation of the Capability service that allows
you to specify what capabilities you want it to report as disabled, and it wraps
all stubs in a wrapper that will throw a CapabilityDisabledError if you attempt
to use a disabled service or method.

Example usage:

class MyUnitTest(appenginetest.DatastoreTest, appenginetest.MemcacheTest):
  def setUp(self):
    # Pass arguments to the test classes here to control how stubs
    # are initialized
    super(MyUnitTest, self).setUp(
        require_indexes=True,
        disabled_capabilities=set([('datastore_v3', '*')]))

  def testMemcache(self):
    memcache.put('testkey', 'testval')
    self.assertEqual(memcache.get('testkey'), 'testval')

  def testCapabilities(self):
    cset = capabilities.CapabilitySet('datastore_v3', methods=['get'])
    self.assertEqual(cset.is_enabled(), False)

  def testDatastore(self):
    self.assertRaises(apiproxy_errors.CapabilityDisabledError,
                      db.get, db.Key.from_path('Model', 1))


if __name__ == "__main__":
  appenginetest.main()

test_capabilities.py

from google.appengine.api import apiproxy_rpc
from google.appengine.api import apiproxy_stub
from google.appengine.api import capabilities
from google.appengine.runtime import apiproxy_errors

IsEnabledRequest = capabilities.IsEnabledRequest
IsEnabledResponse = capabilities.IsEnabledResponse
CapabilityConfig = capabilities.CapabilityConfig

class CapabilityServiceStub(apiproxy_stub.APIProxyStub):
  """Test-oriented capability service stub."""

  def __init__(self, disabled_capabilities, disabled_methods,
               service_name='capability_service'):
    """Constructor.

    Args:
      disabled_capabilities: A set of (package, capability) tuples defining
        capabilities that are disabled.
      disabled_methods: A set of (package, method) tuples defining methods that
        are disabled. An entry of ('package', '*') in disabled_capabilities
        is treated the same as finding the method being tested in this set.
      service_name: Service name expected for all calls.
    """
    super(CapabilityServiceStub, self).__init__(service_name)
    self.disabled_capabilities = disabled_capabilities
    self.disabled_methods = disabled_methods

  def _Dynamic_IsEnabled(self, request, response):
    """Implementation of CapabilityService::IsEnabled().

    Args:
      request: An IsEnabledRequest.
      response: An IsEnabledResponse.
    """
    package = request.package()
    if (package, '*') in self.disabled_capabilities:
      response.set_summary_status(IsEnabledRequest.DISABLED)
      config = response.add_config()
      config.set_package(package)
      config.set_capability('*')
      config.set_status(CapabilityConfig.DISABLED)
    else:
      any_disabled = False
      for method in request.call_list():
        config = response.add_config()
        config.set_package(package)
        config.set_capability(method)
        if (package, method) in self.disabled_methods:
          config.set_status(IsEnabledResponse.DISABLED)
          any_disabled = True
        else:
          config.set_status(IsEnabledResponse.ENABLED)
      for capability in request.capability_list():
        config = response.add_config()
        config.set_package(package)
        config.set_capability(capability)
        if (package, capability) in self.disabled_capabilities:
          any_disabled = True
          config.set_status(IsEnabledResponse.DISABLED)
        else:
          config.set_status(IsEnabledResponse.ENABLED)
    response.set_summary_status(IsEnabledResponse.DISABLED if any_disabled
                                else IsEnabledResponse.ENABLED)


class CapabilityStubWrapper(object):
  """A wrapper for stubs that raises CapabilityDisabledError when needed."""

  def __init__(self, wrapped_stub, disabled_capabilities, disabled_methods):
    self.wrapped_stub = wrapped_stub
    self.disabled_capabilities = disabled_capabilities
    self.disabled_methods = disabled_methods

  def CreateRPC(self):
    """Creates a (dummy) RPC object instance."""
    return apiproxy_rpc.RPC(stub=self)

  def MakeSyncCall(self, service, call, request, response):
    if ((service, '*') in self.disabled_capabilities
        or (service, call) in self.disabled_methods):
      raise apiproxy_errors.CapabilityDisabledError()
    self.wrapped_stub.MakeSyncCall(service, call, request, response)