<s> """<STR_LIT>""" <EOL> import os <EOL> import unittest <EOL> from google . appengine . api import apiproxy_stub_map <EOL> from google . appengine . api import datastore_file_stub <EOL> try : <EOL> from google . appengine . api import mail_stub <EOL> except AttributeError : <EOL> mail_stub = None <EOL> from google . appengine . api import urlfetch_stub <EOL> from google . appengine . api import user_service_stub <EOL> from google . appengine . api . app_identity import app_identity_stub <EOL> from google . appengine . api . blobstore import blobstore_stub <EOL> from google . appengine . api . blobstore import dict_blob_storage <EOL> from google . appengine . api . capabilities import capability_stub <EOL> from google . appengine . api . channel import channel_service_stub <EOL> from google . appengine . api . files import file_service_stub <EOL> try : <EOL> from google . appengine . api . images import images_stub <EOL> except ImportError : <EOL> images_stub = None <EOL> try : <EOL> from google . appengine . api . logservice import logservice_stub <EOL> except ImportError : <EOL> logservice_stub = None <EOL> from google . appengine . api . memcache import memcache_stub <EOL> from google . appengine . api . taskqueue import taskqueue_stub <EOL> from google . appengine . api . xmpp import xmpp_service_stub <EOL> try : <EOL> from google . appengine . datastore import datastore_sqlite_stub <EOL> except ImportError : <EOL> datastore_sqlite_stub = None <EOL> from google . appengine . datastore import datastore_stub_util <EOL> from google . appengine . ext . cloudstorage import stub_dispatcher as gcs_dispatcher <EOL> DEFAULT_ENVIRONMENT = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } <EOL> DEFAULT_APP_ID = DEFAULT_ENVIRONMENT [ '<STR_LIT>' ] <EOL> DEFAULT_AUTH_DOMAIN = DEFAULT_ENVIRONMENT [ '<STR_LIT>' ] <EOL> DEFAULT_SERVER_NAME = DEFAULT_ENVIRONMENT [ '<STR_LIT>' ] <EOL> DEFAULT_SERVER_SOFTWARE = DEFAULT_ENVIRONMENT [ '<STR_LIT>' ] <EOL> DEFAULT_SERVER_PORT = DEFAULT_ENVIRONMENT [ '<STR_LIT>' ] <EOL> APP_IDENTITY_SERVICE_NAME = '<STR_LIT>' <EOL> BLOBSTORE_SERVICE_NAME = '<STR_LIT>' <EOL> CAPABILITY_SERVICE_NAME = '<STR_LIT>' <EOL> CHANNEL_SERVICE_NAME = '<STR_LIT>' <EOL> DATASTORE_SERVICE_NAME = '<STR_LIT>' <EOL> FILES_SERVICE_NAME = '<STR_LIT:file>' <EOL> IMAGES_SERVICE_NAME = '<STR_LIT>' <EOL> LOG_SERVICE_NAME = '<STR_LIT>' <EOL> MAIL_SERVICE_NAME = '<STR_LIT>' <EOL> MEMCACHE_SERVICE_NAME = '<STR_LIT>' <EOL> TASKQUEUE_SERVICE_NAME = '<STR_LIT>' <EOL> URLFETCH_SERVICE_NAME = '<STR_LIT>' <EOL> USER_SERVICE_NAME = '<STR_LIT:user>' <EOL> XMPP_SERVICE_NAME = '<STR_LIT>' <EOL> INIT_STUB_METHOD_NAMES = { <EOL> APP_IDENTITY_SERVICE_NAME : '<STR_LIT>' , <EOL> BLOBSTORE_SERVICE_NAME : '<STR_LIT>' , <EOL> CAPABILITY_SERVICE_NAME : '<STR_LIT>' , <EOL> CHANNEL_SERVICE_NAME : '<STR_LIT>' , <EOL> DATASTORE_SERVICE_NAME : '<STR_LIT>' , <EOL> FILES_SERVICE_NAME : '<STR_LIT>' , <EOL> IMAGES_SERVICE_NAME : '<STR_LIT>' , <EOL> LOG_SERVICE_NAME : '<STR_LIT>' , <EOL> MAIL_SERVICE_NAME : '<STR_LIT>' , <EOL> MEMCACHE_SERVICE_NAME : '<STR_LIT>' , <EOL> TASKQUEUE_SERVICE_NAME : '<STR_LIT>' , <EOL> URLFETCH_SERVICE_NAME : '<STR_LIT>' , <EOL> USER_SERVICE_NAME : '<STR_LIT>' , <EOL> XMPP_SERVICE_NAME : '<STR_LIT>' , <EOL> } <EOL> SUPPORTED_SERVICES = sorted ( INIT_STUB_METHOD_NAMES ) <EOL> AUTO_ID_POLICY_SEQUENTIAL = datastore_stub_util . SEQUENTIAL <EOL> AUTO_ID_POLICY_SCATTERED = datastore_stub_util . SCATTERED <EOL> class Error ( Exception ) : <EOL> """<STR_LIT>""" <EOL> class NotActivatedError ( Error ) : <EOL> """<STR_LIT>""" <EOL> class StubNotSupportedError ( Error ) : <EOL> """<STR_LIT>""" <EOL> class Testbed ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self ) : <EOL> self . _activated = False <EOL> self . _enabled_stubs = { } <EOL> self . _blob_storage = None <EOL> def activate ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _orig_env = dict ( os . environ ) <EOL> self . setup_env ( ) <EOL> self . _original_stub_map = apiproxy_stub_map . apiproxy <EOL> self . _test_stub_map = apiproxy_stub_map . APIProxyStubMap ( ) <EOL> internal_map = self . _original_stub_map . _APIProxyStubMap__stub_map <EOL> self . _test_stub_map . _APIProxyStubMap__stub_map = dict ( internal_map ) <EOL> apiproxy_stub_map . apiproxy = self . _test_stub_map <EOL> self . _activated = True <EOL> def deactivate ( self ) : <EOL> """<STR_LIT>""" <EOL> if not self . _activated : <EOL> raise NotActivatedError ( '<STR_LIT>' ) <EOL> for service_name , deactivate_callback in self . _enabled_stubs . iteritems ( ) : <EOL> if deactivate_callback : <EOL> deactivate_callback ( self . _test_stub_map . GetStub ( service_name ) ) <EOL> apiproxy_stub_map . apiproxy = self . _original_stub_map <EOL> self . _enabled_stubs = { } <EOL> os . environ . clear ( ) <EOL> os . environ . update ( self . _orig_env ) <EOL> self . _blob_storage = None <EOL> self . _activated = False <EOL> def setup_env ( self , overwrite = False , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> merged_kwargs = { } <EOL> for key , value in kwargs . iteritems ( ) : <EOL> if key == '<STR_LIT>' : <EOL> key = '<STR_LIT>' <EOL> merged_kwargs [ key . upper ( ) ] = value <EOL> if not overwrite : <EOL> for key , value in DEFAULT_ENVIRONMENT . iteritems ( ) : <EOL> if key not in merged_kwargs : <EOL> merged_kwargs [ key ] = value <EOL> for key , value in merged_kwargs . iteritems ( ) : <EOL> if overwrite or key not in os . environ : <EOL> os . environ [ key ] = value <EOL> def _register_stub ( self , service_name , stub , deactivate_callback = None ) : <EOL> """<STR_LIT>""" <EOL> self . _disable_stub ( service_name ) <EOL> self . _test_stub_map . RegisterStub ( service_name , stub ) <EOL> self . _enabled_stubs [ service_name ] = deactivate_callback <EOL> def _disable_stub ( self , service_name ) : <EOL> """<STR_LIT>""" <EOL> if not self . _activated : <EOL> raise NotActivatedError ( '<STR_LIT>' ) <EOL> deactivate_callback = self . _enabled_stubs . pop ( service_name , None ) <EOL> if deactivate_callback : <EOL> deactivate_callback ( self . _test_stub_map . GetStub ( service_name ) ) <EOL> if service_name in self . _test_stub_map . _APIProxyStubMap__stub_map : <EOL> del self . _test_stub_map . _APIProxyStubMap__stub_map [ service_name ] <EOL> def get_stub ( self , service_name ) : <EOL> """<STR_LIT>""" <EOL> if not self . _activated : <EOL> raise NotActivatedError ( '<STR_LIT>' ) <EOL> if service_name not in SUPPORTED_SERVICES : <EOL> msg = '<STR_LIT>' % service_name <EOL> raise StubNotSupportedError ( msg ) <EOL> if service_name not in self . _enabled_stubs : <EOL> return None <EOL> return self . _test_stub_map . GetStub ( service_name ) <EOL> def init_app_identity_stub ( self , enable = True ) : <EOL> """<STR_LIT>""" <EOL> if not enable : <EOL> self . _disable_stub ( APP_IDENTITY_SERVICE_NAME ) <EOL> return <EOL> stub = app_identity_stub . AppIdentityServiceStub ( ) <EOL> self . _register_stub ( APP_IDENTITY_SERVICE_NAME , stub ) <EOL> def _get_blob_storage ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . _blob_storage is None : <EOL> self . _blob_storage = dict_blob_storage . DictBlobStorage ( ) <EOL> return self . _blob_storage <EOL> def init_blobstore_stub ( self , enable = True ) : <EOL> """<STR_LIT>""" <EOL> if not enable : <EOL> self . _disable_stub ( BLOBSTORE_SERVICE_NAME ) <EOL> return <EOL> stub = blobstore_stub . BlobstoreServiceStub ( self . _get_blob_storage ( ) ) <EOL> self . _register_stub ( BLOBSTORE_SERVICE_NAME , stub ) <EOL> def init_capability_stub ( self , enable = True ) : <EOL> """<STR_LIT>""" <EOL> if not enable : <EOL> self . _disable_stub ( CAPABILITY_SERVICE_NAME ) <EOL> return <EOL> stub = capability_stub . CapabilityServiceStub ( ) <EOL> self . _register_stub ( CAPABILITY_SERVICE_NAME , stub ) <EOL> def init_channel_stub ( self , enable = True ) : <EOL> """<STR_LIT>""" <EOL> if not enable : <EOL> self . _disable_stub ( CHANNEL_SERVICE_NAME ) <EOL> return <EOL> stub = channel_service_stub . ChannelServiceStub ( ) <EOL> self . _register_stub ( CHANNEL_SERVICE_NAME , stub ) <EOL> def init_datastore_v3_stub ( self , enable = True , datastore_file = None , <EOL> use_sqlite = False , <EOL> auto_id_policy = AUTO_ID_POLICY_SEQUENTIAL , <EOL> ** stub_kw_args ) : <EOL> """<STR_LIT>""" <EOL> if not enable : <EOL> self . _disable_stub ( DATASTORE_SERVICE_NAME ) <EOL> return <EOL> if use_sqlite : <EOL> if datastore_sqlite_stub is None : <EOL> raise StubNotSupportedError ( <EOL> '<STR_LIT>' ) <EOL> stub = datastore_sqlite_stub . DatastoreSqliteStub ( <EOL> os . environ [ '<STR_LIT>' ] , <EOL> datastore_file , <EOL> use_atexit = False , <EOL> auto_id_policy = auto_id_policy , <EOL> ** stub_kw_args ) <EOL> else : <EOL> stub_kw_args . setdefault ( '<STR_LIT>' , False ) <EOL> stub = datastore_file_stub . DatastoreFileStub ( <EOL> os . environ [ '<STR_LIT>' ] , <EOL> datastore_file , <EOL> use_atexit = False , <EOL> auto_id_policy = auto_id_policy , <EOL> ** stub_kw_args ) <EOL> self . _register_stub ( DATASTORE_SERVICE_NAME , stub , <EOL> self . _deactivate_datastore_v3_stub ) <EOL> def _deactivate_datastore_v3_stub ( self , stub ) : <EOL> stub . Write ( ) <EOL> def init_files_stub ( self , enable = True ) : <EOL> """<STR_LIT>""" <EOL> if not enable : <EOL> self . _disable_stub ( FILES_SERVICE_NAME ) <EOL> return <EOL> stub = file_service_stub . FileServiceStub ( self . _get_blob_storage ( ) ) <EOL> self . _register_stub ( FILES_SERVICE_NAME , stub ) <EOL> def init_images_stub ( self , enable = True ) : <EOL> """<STR_LIT>""" <EOL> if not enable : <EOL> self . _disable_stub ( IMAGES_SERVICE_NAME ) <EOL> return <EOL> if images_stub is None : <EOL> msg = ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> raise StubNotSupportedError ( msg ) <EOL> stub = images_stub . ImagesServiceStub ( ) <EOL> self . _register_stub ( IMAGES_SERVICE_NAME , stub ) <EOL> def init_logservice_stub ( self , enable = True ) : <EOL> """<STR_LIT>""" <EOL> if not enable : <EOL> self . _disable_stub ( LOG_SERVICE_NAME ) <EOL> return <EOL> if logservice_stub is None : <EOL> raise StubNotSupportedError ( <EOL> '<STR_LIT>' ) <EOL> stub = logservice_stub . LogServiceStub ( ) <EOL> self . _register_stub ( LOG_SERVICE_NAME , stub ) <EOL> def init_mail_stub ( self , enable = True , ** stub_kw_args ) : <EOL> """<STR_LIT>""" <EOL> if not enable : <EOL> self . _disable_stub ( MAIL_SERVICE_NAME ) <EOL> return <EOL> stub = mail_stub . MailServiceStub ( ** stub_kw_args ) <EOL> self . _register_stub ( MAIL_SERVICE_NAME , stub ) <EOL> def init_memcache_stub ( self , enable = True ) : <EOL> """<STR_LIT>""" <EOL> if not enable : <EOL> self . _disable_stub ( MEMCACHE_SERVICE_NAME ) <EOL> return <EOL> stub = memcache_stub . MemcacheServiceStub ( ) <EOL> self . _register_stub ( MEMCACHE_SERVICE_NAME , stub ) <EOL> def init_taskqueue_stub ( self , enable = True , ** stub_kw_args ) : <EOL> """<STR_LIT>""" <EOL> if not enable : <EOL> self . _disable_stub ( TASKQUEUE_SERVICE_NAME ) <EOL> return <EOL> stub = taskqueue_stub . TaskQueueServiceStub ( ** stub_kw_args ) <EOL> self . _register_stub ( TASKQUEUE_SERVICE_NAME , stub ) <EOL> def init_urlfetch_stub ( self , enable = True ) : <EOL> """<STR_LIT>""" <EOL> if not enable : <EOL> self . _disable_stub ( URLFETCH_SERVICE_NAME ) <EOL> return <EOL> urlmatchers_to_fetch_functions = [ ] <EOL> urlmatchers_to_fetch_functions . extend ( <EOL> gcs_dispatcher . URLMATCHERS_TO_FETCH_FUNCTIONS ) <EOL> stub = urlfetch_stub . URLFetchServiceStub ( <EOL> urlmatchers_to_fetch_functions = urlmatchers_to_fetch_functions ) <EOL> self . _register_stub ( URLFETCH_SERVICE_NAME , stub ) <EOL> def init_user_stub ( self , enable = True , ** stub_kw_args ) : <EOL> """<STR_LIT>""" <EOL> if not enable : <EOL> self . _disable_stub ( USER_SERVICE_NAME ) <EOL> return <EOL> stub = user_service_stub . UserServiceStub ( ** stub_kw_args ) <EOL> self . _register_stub ( USER_SERVICE_NAME , stub ) <EOL> def init_xmpp_stub ( self , enable = True ) : <EOL> """<STR_LIT>""" <EOL> if not enable : <EOL> self . _disable_stub ( XMPP_SERVICE_NAME ) <EOL> return <EOL> stub = xmpp_service_stub . XmppServiceStub ( ) <EOL> self . _register_stub ( XMPP_SERVICE_NAME , stub ) <EOL> def _init_stub ( self , service_name , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> if not self . _activated : <EOL> raise NotActivatedError ( '<STR_LIT>' ) <EOL> method_name = INIT_STUB_METHOD_NAMES . get ( service_name , None ) <EOL> if method_name is None : <EOL> msg = '<STR_LIT>' % service_name <EOL> raise StubNotSupportedError ( msg ) <EOL> method = getattr ( self , method_name ) <EOL> method ( * args , ** kwargs ) <EOL> def init_all_stubs ( self , enable = True ) : <EOL> """<STR_LIT>""" <EOL> for service_name in SUPPORTED_SERVICES : <EOL> self . _init_stub ( service_name , enable ) </s>
