fromlogilab.common.testlibimportTestCase,mock_objectimportos,os.pathasospimporttimefromshutilimportcopy# additional monkey patches necessary in regular cubicweb environmentfromcubicweb.serverimportrqlannotationfromcubicweb.goa.overridesimportrqlannotationasgoarqlannotationrqlannotation.SQLGenAnnotator=goarqlannotation.SQLGenAnnotatorrqlannotation.set_qdata=goarqlannotation.set_qdatatry:fromgoogle.appengine.apiimportapiproxy_stub_mapfromgoogle.appengine.apiimportdatastore_file_stubfromgoogle.appengine.extimportdbasgdbfromcubicweb.goaimportdb,do_monkey_patchfromcubicweb.goa.dbmyamsimportload_schemaimport_appengine_failed=NoneexceptImportError,exc:raiseclassdb:classModel:passclassDummyProperty:def__init__(self,*args,**kwargs):passTextProperty=DummyPropertyStringProperty=DummyPropertyBlobProperty=DummyPropertyDateProperty=DummyPropertyReferenceProperty=DummyPropertySelfReferenceProperty=DummyPropertyimport_appengine_failed='cannot import appengine: %s'%excfromcubicwebimportCW_SOFTWARE_ROOTfromcubicweb.server.utilsimportcrypt_passwordfromcubicweb.devtools.fakeimportFakeRequestfromcubicweb.goa.goavregimportGAERegistryfromcubicweb.goa.goaconfigimportGAEConfigurationfromcubicweb.goa.dbinitimport(create_user,create_groups,fix_entities,init_persistent_schema,insert_versions)importlogginglogger=logging.getLogger()logger.setLevel(logging.CRITICAL)do_monkey_patch()classGAEBasedTC(TestCase):APP_ID=u'test_app'AUTH_DOMAIN='gmail.com'LOGGED_IN_USER=u't...@example.com'# set to '' for no logged in userMODEL_CLASSES=NoneLOAD_APP_MODULES=Noneconfig=None_DS_TEMPL_FILE='tmpdb-template'defload_schema_hook(self,loader):loader.import_yams_cube_schema('data')@propertydefDS_FILE(self):returnself.DS_TEMPL_FILE.replace('-template','')@propertydefDS_TEMPL_FILE(self):returnself._DS_TEMPL_FILE+'_'.join(sorted(cls.__name__forclsinself.MODEL_CLASSES))def_set_ds_file(self,dsfile):# Start with a fresh api proxy.apiproxy_stub_map.apiproxy=apiproxy_stub_map.APIProxyStubMap()# Use a fresh stub datastore.stub=datastore_file_stub.DatastoreFileStub(self.APP_ID,dsfile,dsfile+'.history')apiproxy_stub_map.apiproxy.RegisterStub('datastore_v3',stub)defsetUp(self):ifimport_appengine_failed:self.skip(import_appengine_failed)# Ensure we're in UTC.os.environ['TZ']='UTC'time.tzset()ifosp.exists(self.DS_TEMPL_FILE):copy(self.DS_TEMPL_FILE,self.DS_FILE)need_ds_init=Falseself._set_ds_file(self.DS_FILE)else:need_ds_init=Trueself._set_ds_file(self.DS_TEMPL_FILE)# from google.appengine.api import mail_stub# from google3.apphosting.api import urlfetch_stub# from google3.apphosting.api import user_service_stub # # Use a fresh stub UserService.# apiproxy_stub_map.apiproxy.RegisterStub(# 'user', user_service_stub.UserServiceStub())os.environ['AUTH_DOMAIN']=self.AUTH_DOMAINos.environ['USER_EMAIL']=self.LOGGED_IN_USER# # Use a fresh urlfetch stub.# apiproxy_stub_map.apiproxy.RegisterStub(# 'urlfetch', urlfetch_stub.URLFetchServiceStub())# # Use a fresh mail stub.# apiproxy_stub_map.apiproxy.RegisterStub(# 'mail', mail_stub.MailServiceStub())ifself.MODEL_CLASSESisNone:raiseException('GAEBasedTC should set MODEL_CLASSES class attribute')gdb._kind_map={}self.config=self.configorGAEConfiguration('toto')self.config.init_log(logging.CRITICAL)self.schema=self.config.load_schema(self.MODEL_CLASSES,self.load_schema_hook)self.vreg=GAERegistry(self.config)self.vreg.schema=self.schemaself.vreg.load_module(db)fromcubicweb.goa.appobjectsimportsessionsself.vreg.load_module(sessions)fromcubicweb.entitiesimportauthobjs,schemaobjsself.vreg.load_module(authobjs)self.vreg.load_module(schemaobjs)ifself.config['use-google-auth']:fromcubicweb.goa.appobjectsimportgauthserviceself.vreg.load_module(gauthservice)ifself.LOAD_APP_MODULESisnotNone:formoduleinself.LOAD_APP_MODULES:self.vreg.load_module(module)forclsinself.MODEL_CLASSES:self.vreg.load_object(cls)self.session_manager=self.vreg.select_component('sessionmanager')ifneed_ds_init:# create default groups and create entities according to the schemacreate_groups()ifnotself.config['use-google-auth']:create_user(self.LOGGED_IN_USER,'toto',('users','managers'))self.session=self.login(self.LOGGED_IN_USER,'toto')else:req=FakeRequest(vreg=self.vreg)self.session=self.session_manager.open_session(req)self.user=self.session.user()ssession=self.config.repo_session(self.session.sessionid)ssession.set_pool()init_persistent_schema(ssession,self.schema)insert_versions(ssession,self.config)ssession.commit()fix_entities(self.schema)copy(self.DS_TEMPL_FILE,self.DS_FILE)self._set_ds_file(self.DS_FILE)else:ifnotself.config['use-google-auth']:self.session=self.login(self.LOGGED_IN_USER,'toto')else:req=FakeRequest(vreg=self.vreg)self.session=self.session_manager.open_session(req)self.user=self.session.user()deftearDown(self):self.session.close()defrequest(self):req=FakeRequest(vreg=self.vreg)req.set_connection(self.session,self.user)returnreqdefadd_entity(self,etype,**kwargs):cu=self.session.cursor()rql='INSERT %s X'%etypeifkwargs:rql+=': %s'%', '.join('X %s%%(%s)s'%(key,key)forkeyinkwargs)rset=cu.execute(rql,kwargs)returnrset.get_entity(0,0)defexecute(self,*args):returnself.session.cursor().execute(*args)defcommit(self):self.session.commit()defrollback(self):self.session.rollback()defcreate_user(self,login,groups=('users',),req=None):assertnotself.config['use-google-auth']user=self.add_entity('EUser',upassword=str(login),login=unicode(login))cu=self.session.cursor()cu.execute('SET X in_group G WHERE X eid %%(x)s, G name IN(%s)'%','.join(repr(g)forgingroups),{'x':user.eid},'x')returnuserdeflogin(self,login,password=None):assertnotself.config['use-google-auth']req=FakeRequest(vreg=self.vreg)req.form['__login']=loginreq.form['__password']=passwordorloginreturnself.session_manager.open_session(req)