Skip to content

Instantly share code, notes, and snippets.

@Arachnid
Created September 13, 2009 17:30
Show Gist options
  • Save Arachnid/186251 to your computer and use it in GitHub Desktop.
Save Arachnid/186251 to your computer and use it in GitHub Desktop.
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()
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()
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)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment