cubicweb: comparison devtools/__init_

equal deleted inserted replaced

-:f4d1d5d9ccbb
+:90f2f20367bc
 # FOR A PARTICULAR PURPOSE.  See the GNU Lesser General Public License for more
 # details.
 #
 # You should have received a copy of the GNU Lesser General Public License along
 # with CubicWeb.  If not, see <http://www.gnu.org/licenses/>.
-"""Test tools for cubicweb
+"""Test tools for cubicweb"""
-"""
 __docformat__ = "restructuredtext en"
 import os
+import sys
 import logging
 from datetime import timedelta
 from os.path import (abspath, join, exists, basename, dirname, normpath, split,
 isfile, isabs)
 class TestServerConfiguration(ServerConfiguration):
 mode = 'test'
 set_language = False
 read_instance_schema = False
 init_repository = True
+db_require_setup = True
 options = cwconfig.merge_options(ServerConfiguration.options + (
 ('anonymous-user',
 {'type' : 'string',
 'default': None,
 'help': 'login of the CubicWeb user account to use for anonymous user (if you want to allow anonymous)',
 'help': 'password of the CubicWeb user account matching login',
 'group': 'main', 'level': 1,
 }),
 ))
-def __init__(self, appid, log_threshold=logging.CRITICAL+10):
+def __init__(self, appid, apphome=None, log_threshold=logging.CRITICAL+10):
+# must be set before calling parent __init__
+if apphome is None:
+if exists(appid):
+apphome = abspath(appid)
+else: # cube test
+apphome = abspath('..')
+self._apphome = apphome
 ServerConfiguration.__init__(self, appid)
 self.init_log(log_threshold, force=True)
 # need this, usually triggered by cubicweb-ctl
 self.load_cwctl_plugins()
 anonymous_user = TwistedConfiguration.anonymous_user.im_func
 @property
 def apphome(self):
-if exists(self.appid):
+return self._apphome
-return abspath(self.appid)
-# cube test
-return abspath('..')
 appdatahome = apphome
 def load_configuration(self):
 super(TestServerConfiguration, self).load_configuration()
 self.global_set_option('anonymous-user', 'anon')
 self.global_set_option('undo-support', '')
 def main_config_file(self):
 """return instance's control configuration file"""
 return join(self.apphome, '%s.conf' % self.name)
-def instance_md5_version(self):
-return ''
 def bootstrap_cubes(self):
 try:
 super(TestServerConfiguration, self).bootstrap_cubes()
 except IOError:
 sources = super(TestServerConfiguration, self).sources()
 if not sources:
 sources = DEFAULT_SOURCES
 return sources
+# web config methods needed here for cases when we use this config as a web
+# config
+def instance_md5_version(self):
+return ''
+def default_base_url(self):
+return BASE_URL
 class BaseApptestConfiguration(TestServerConfiguration, TwistedConfiguration):
 repo_method = 'inmemory'
 options = cwconfig.merge_options(TestServerConfiguration.options
 + TwistedConfiguration.options)
 cube_appobject_path = TestServerConfiguration.cube_appobject_path | TwistedConfiguration.cube_appobject_path
 def available_languages(self, *args):
 return ('en', 'fr', 'de')
-def ext_resources_file(self):
-"""return instance's external resources file"""
-return join(self.apphome, 'data', 'external_resources')
 def pyro_enabled(self):
-# but export PYRO_MULTITHREAD=0 or you get problems with sqlite and threads
+# but export PYRO_MULTITHREAD=0 or you get problems with sqlite and
+# threads
 return True
+# XXX merge with BaseApptestConfiguration ?
 class ApptestConfiguration(BaseApptestConfiguration):
-def __init__(self, appid, log_threshold=logging.CRITICAL, sourcefile=None):
+def __init__(self, appid, apphome=None,
-BaseApptestConfiguration.__init__(self, appid, log_threshold=log_threshold)
+log_threshold=logging.CRITICAL, sourcefile=None):
+BaseApptestConfiguration.__init__(self, appid, apphome,
+log_threshold=log_threshold)
 self.init_repository = sourcefile is None
 self.sourcefile = sourcefile
+class RealDatabaseConfiguration(ApptestConfiguration):
+"""configuration class for tests to run on a real database.
+The intialization is done by specifying a source file path.
+Important note: init_test_database / reset_test_database steps are
+skipped. It's thus up to the test developer to implement setUp/tearDown
+accordingly.
+Example usage::
+class MyTests(CubicWebTC):
+_config = RealDatabseConfiguration('myapp',
+sourcefile='/path/to/sources')
+def test_something(self):
+rset = self.execute('Any X WHERE X is CWUser')
+self.view('foaf', rset)
+"""
+db_require_setup = False    # skip init_db / reset_db steps
+read_instance_schema = True # read schema from database
 # test database handling #######################################################
 def init_test_database(config=None, configdir='data'):
 """init a test database for a specific driver"""
 from cubicweb.dbapi import in_memory_cnx
 config = config or TestServerConfiguration(configdir)
 sources = config.sources()
 driver = sources['system']['db-driver']
-if driver == 'sqlite':
+if config.db_require_setup:
-init_test_database_sqlite(config)
+if driver == 'sqlite':
-elif driver == 'postgres':
+init_test_database_sqlite(config)
-init_test_database_postgres(config)
+elif driver == 'postgres':
-elif driver == 'sqlserver2005':
+init_test_database_postgres(config)
-init_test_database_sqlserver2005(config)
+else:
-else:
+raise ValueError('no initialization function for driver %r' % driver)
-raise ValueError('no initialization function for driver %r' % driver)
 config._cubes = None # avoid assertion error
 repo, cnx = in_memory_cnx(config, unicode(sources['admin']['login']),
 password=sources['admin']['password'] or 'xxx')
 if driver == 'sqlite':
 install_sqlite_patch(repo.querier)
 return repo, cnx
 def reset_test_database(config):
 """init a test database for a specific driver"""
+if not config.db_require_setup:
+return
 driver = config.sources()['system']['db-driver']
 if driver == 'sqlite':
 reset_test_database_sqlite(config)
-elif driver in ('sqlserver2005', 'postgres'):
+elif driver == 'postgres':
-# XXX do something with dump/restore ?
+init_test_database_postgres(config)
-print 'resetting the database is not done for', driver
-print 'you should handle it manually'
 else:
 raise ValueError('no reset function for driver %r' % driver)
 ### postgres test database handling ############################################
 def init_test_database_postgres(config):
 """initialize a fresh postgresql databse used for testing purpose"""
-if config.init_repository:
+from logilab.database import get_db_helper
 from cubicweb.server import init_repository
-init_repository(config, interactive=False, drop=True)
+from cubicweb.server.serverctl import (createdb, system_source_cnx,
+_db_sys_cnx)
-### sqlserver2005 test database handling ############################################
+source = config.sources()['system']
+dbname = source['db-name']
+templdbname = dbname + '_template'
+helper = get_db_helper('postgres')
+# connect on the dbms system base to create our base
+dbcnx = _db_sys_cnx(source, 'CREATE DATABASE and / or USER', verbose=0)
+cursor = dbcnx.cursor()
+try:
+if dbname in helper.list_databases(cursor):
+cursor.execute('DROP DATABASE %s' % dbname)
+if not templdbname in helper.list_databases(cursor):
+source['db-name'] = templdbname
+createdb(helper, source, dbcnx, cursor)
+dbcnx.commit()
+cnx = system_source_cnx(source, special_privs='LANGUAGE C', verbose=0)
+templcursor = cnx.cursor()
+# XXX factorize with db-create code
+helper.init_fti_extensions(templcursor)
+# install plpythonu/plpgsql language if not installed by the cube
+langs = sys.platform == 'win32' and ('plpgsql',) or ('plpythonu', 'plpgsql')
+for extlang in langs:
+helper.create_language(templcursor, extlang)
+cnx.commit()
+templcursor.close()
+cnx.close()
+init_repository(config, interactive=False)
+source['db-name'] = dbname
+except:
+dbcnx.rollback()
+# XXX drop template
+raise
+createdb(helper, source, dbcnx, cursor, template=templdbname)
+dbcnx.commit()
+dbcnx.close()
+### sqlserver2005 test database handling #######################################
 def init_test_database_sqlserver2005(config):
 """initialize a fresh sqlserver databse used for testing purpose"""
 if config.init_repository:
 from cubicweb.server import init_repository
 import shutil
 from cubicweb.server import init_repository
 init_repository(config, interactive=False)
 dbfile = config.sources()['system']['db-name']
 shutil.copy(dbfile, '%s-template' % dbfile)
 def install_sqlite_patch(querier):
 """This patch hotfixes the following sqlite bug :
 - http://www.sqlite.org/cvstrac/tktview?tn=1327,33
 (some dates are returned as strings rather thant date objects)

branch	oldstable
changeset 6665	90f2f20367bc
parent 6593	0fd8792c9c8a
child 6594	e10468a23291
child 6729	1a423eaee782