diff -r f4d1d5d9ccbb -r 90f2f20367bc devtools/__init__.py --- a/devtools/__init__.py Tue Jul 27 12:36:03 2010 +0200 +++ b/devtools/__init__.py Wed Nov 03 16:38:28 2010 +0100 @@ -15,12 +15,12 @@ # # You should have received a copy of the GNU Lesser General Public License along # with CubicWeb. If not, see . -"""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, @@ -95,6 +95,7 @@ set_language = False read_instance_schema = False init_repository = True + db_require_setup = True options = cwconfig.merge_options(ServerConfiguration.options + ( ('anonymous-user', {'type' : 'string', @@ -110,7 +111,14 @@ }), )) - 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 @@ -120,10 +128,7 @@ @property def apphome(self): - if exists(self.appid): - return abspath(self.appid) - # cube test - return abspath('..') + return self._apphome appdatahome = apphome def load_configuration(self): @@ -137,9 +142,6 @@ """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() @@ -170,6 +172,15 @@ 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' @@ -181,23 +192,45 @@ 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): - BaseApptestConfiguration.__init__(self, appid, log_threshold=log_threshold) + def __init__(self, appid, apphome=None, + 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'): @@ -206,14 +239,13 @@ config = config or TestServerConfiguration(configdir) sources = config.sources() driver = sources['system']['db-driver'] - if driver == 'sqlite': - init_test_database_sqlite(config) - elif driver == 'postgres': - init_test_database_postgres(config) - elif driver == 'sqlserver2005': - init_test_database_sqlserver2005(config) - else: - raise ValueError('no initialization function for driver %r' % driver) + if config.db_require_setup: + if driver == 'sqlite': + init_test_database_sqlite(config) + elif driver == 'postgres': + init_test_database_postgres(config) + else: + 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') @@ -221,16 +253,15 @@ 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'): - # XXX do something with dump/restore ? - print 'resetting the database is not done for', driver - print 'you should handle it manually' + elif driver == 'postgres': + init_test_database_postgres(config) else: raise ValueError('no reset function for driver %r' % driver) @@ -239,11 +270,46 @@ def init_test_database_postgres(config): """initialize a fresh postgresql databse used for testing purpose""" - if config.init_repository: - from cubicweb.server import init_repository - init_repository(config, interactive=False, drop=True) + from logilab.database import get_db_helper + from cubicweb.server import init_repository + from cubicweb.server.serverctl import (createdb, system_source_cnx, + _db_sys_cnx) + 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 ############################################ +### sqlserver2005 test database handling ####################################### def init_test_database_sqlserver2005(config): """initialize a fresh sqlserver databse used for testing purpose""" @@ -286,7 +352,6 @@ 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