diff -r 000000000000 -r b97547f5f1fa devtools/__init__.py --- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/devtools/__init__.py Wed Nov 05 15:52:50 2008 +0100 @@ -0,0 +1,345 @@ +"""Test tools for cubicweb + +:organization: Logilab +:copyright: 2001-2008 LOGILAB S.A. (Paris, FRANCE), all rights reserved. +:contact: http://www.logilab.fr/ -- mailto:contact@logilab.fr +""" +__docformat__ = "restructuredtext en" + +import os +import logging +from os.path import (abspath, join, exists, basename, dirname, normpath, split, + isfile, isabs) + +from mx.DateTime import strptime, DateTimeDelta + +from cubicweb import CW_SOFTWARE_ROOT, ConfigurationError +from cubicweb.toolsutils import read_config +from cubicweb.cwconfig import CubicWebConfiguration, merge_options +from cubicweb.server.serverconfig import ServerConfiguration +from cubicweb.etwist.twconfig import TwistedConfiguration + +# validators are used to validate (XML, DTD, whatever) view's content +# validators availables are : +# 'dtd' : validates XML + declared DTD +# 'xml' : guarantees XML is well formed +# None : do not try to validate anything +VIEW_VALIDATORS = {} +BASE_URL = 'http://testing.fr/cubicweb/' +DEFAULT_SOURCES = {'system': {'adapter' : 'native', + 'db-encoding' : 'UTF-8', #'ISO-8859-1', + 'db-user' : u'admin', + 'db-password' : 'gingkow', + 'db-name' : 'tmpdb', + 'db-driver' : 'sqlite', + 'db-host' : None, + }, + 'admin' : {'login': u'admin', + 'password': u'gingkow', + }, + } + +class TestServerConfiguration(ServerConfiguration): + mode = 'test' + set_language = False + read_application_schema = False + bootstrap_schema = False + init_repository = True + options = 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)', + 'group': 'main', 'inputlevel': 1, + }), + ('anonymous-password', + {'type' : 'string', + 'default': None, + 'help': 'password of the CubicWeb user account matching login', + 'group': 'main', 'inputlevel': 1, + }), + )) + + if not os.environ.get('APYCOT_ROOT'): + REGISTRY_DIR = normpath(join(CW_SOFTWARE_ROOT, '../cubes')) + + def __init__(self, appid, log_threshold=logging.CRITICAL+10): + ServerConfiguration.__init__(self, appid) + self.global_set_option('log-file', None) + 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 abspath(self.appid) + # application cube test + return abspath('..') + appdatahome = apphome + + def main_config_file(self): + """return application'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: + # no cubes + self.init_cubes( () ) + + sourcefile = None + def sources_file(self): + """define in subclasses self.sourcefile if necessary""" + if self.sourcefile: + print 'Reading sources from', self.sourcefile + sourcefile = self.sourcefile + if not isabs(sourcefile): + sourcefile = join(self.apphome, sourcefile) + else: + sourcefile = super(TestServerConfiguration, self).sources_file() + return sourcefile + + def sources(self): + """By default, we run tests with the sqlite DB backend. One may use its + own configuration by just creating a 'sources' file in the test + directory from wich tests are launched or by specifying an alternative + sources file using self.sourcefile. + """ + sources = super(TestServerConfiguration, self).sources() + if not sources: + sources = DEFAULT_SOURCES + return sources + + def load_defaults(self): + super(TestServerConfiguration, self).load_defaults() + # note: don't call global set option here, OptionManager may not yet be initialized + # add anonymous user + self.set_option('anonymous-user', 'anon') + self.set_option('anonymous-password', 'anon') + # uncomment the line below if you want rql queries to be logged + #self.set_option('query-log-file', '/tmp/test_rql_log.' + `os.getpid()`) + self.set_option('sender-name', 'cubicweb-test') + self.set_option('sender-addr', 'cubicweb-test@logilab.fr') + try: + send_to = '%s@logilab.fr' % os.getlogin() + except OSError: + send_to = '%s@logilab.fr' % (os.environ.get('USER') + or os.environ.get('USERNAME') + or os.environ.get('LOGNAME')) + self.set_option('sender-addr', send_to) + self.set_option('default-dest-addrs', send_to) + self.set_option('base-url', BASE_URL) + + +class BaseApptestConfiguration(TestServerConfiguration, TwistedConfiguration): + repo_method = 'inmemory' + options = merge_options(TestServerConfiguration.options + TwistedConfiguration.options) + cubicweb_vobject_path = TestServerConfiguration.cubicweb_vobject_path | TwistedConfiguration.cubicweb_vobject_path + cube_vobject_path = TestServerConfiguration.cube_vobject_path | TwistedConfiguration.cube_vobject_path + + def available_languages(self, *args): + return ('en', 'fr', 'de') + + def ext_resources_file(self): + """return application'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 + return True + + +class ApptestConfiguration(BaseApptestConfiguration): + + def __init__(self, appid, log_threshold=logging.CRITICAL, sourcefile=None): + BaseApptestConfiguration.__init__(self, appid, log_threshold=log_threshold) + self.init_repository = sourcefile is None + self.sourcefile = sourcefile + import re + self.global_set_option('embed-allowed', re.compile('.*')) + + +class RealDatabaseConfiguration(ApptestConfiguration): + init_repository = False + sourcesdef = {'system': {'adapter' : 'native', + 'db-encoding' : 'UTF-8', #'ISO-8859-1', + 'db-user' : u'admin', + 'db-password' : 'gingkow', + 'db-name' : 'seotest', + 'db-driver' : 'postgres', + 'db-host' : None, + }, + 'admin' : {'login': u'admin', + 'password': u'gingkow', + }, + } + + def __init__(self, appid, log_threshold=logging.CRITICAL, sourcefile=None): + ApptestConfiguration.__init__(self, appid) + self.init_repository = False + + + def sources(self): + """ + By default, we run tests with the sqlite DB backend. + One may use its own configuration by just creating a + 'sources' file in the test directory from wich tests are + launched. + """ + self._sources = self.sourcesdef + return self._sources + + +def buildconfig(dbuser, dbpassword, dbname, adminuser, adminpassword, dbhost=None): + """convenience function that builds a real-db configuration class""" + sourcesdef = {'system': {'adapter' : 'native', + 'db-encoding' : 'UTF-8', #'ISO-8859-1', + 'db-user' : dbuser, + 'db-password' : dbpassword, + 'db-name' : dbname, + 'db-driver' : 'postgres', + 'db-host' : dbhost, + }, + 'admin' : {'login': adminuser, + 'password': adminpassword, + }, + } + return type('MyRealDBConfig', (RealDatabaseConfiguration,), + {'sourcesdef': sourcesdef}) + +def loadconfig(filename): + """convenience function that builds a real-db configuration class + from a file + """ + return type('MyRealDBConfig', (RealDatabaseConfiguration,), + {'sourcesdef': read_config(filename)}) + + +class LivetestConfiguration(BaseApptestConfiguration): + init_repository = False + + def __init__(self, cube=None, sourcefile=None, pyro_name=None, + log_threshold=logging.CRITICAL): + TestServerConfiguration.__init__(self, cube, log_threshold=log_threshold) + self.appid = pyro_name or cube + # don't change this, else some symlink problems may arise in some + # environment (e.g. mine (syt) ;o) + # XXX I'm afraid this test will prevent to run test from a production + # environment + self._sources = None + # application cube test + if cube is not None: + self.apphome = self.cube_dir(cube) + elif 'web' in os.getcwd().split(os.sep): + # web test + self.apphome = join(normpath(join(dirname(__file__), '..')), 'web') + else: + # application cube test + self.apphome = abspath('..') + self.sourcefile = sourcefile + self.global_set_option('realm', '') + self.use_pyro = pyro_name is not None + + def pyro_enabled(self): + if self.use_pyro: + return True + else: + return False + +CubicWebConfiguration.cls_adjust_sys_path() + +def install_sqlite_path(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) + """ + def wrap_execute(base_execute): + def new_execute(*args, **kwargs): + rset = base_execute(*args, **kwargs) + if rset.description: + found_date = False + for row, rowdesc in zip(rset, rset.description): + for cellindex, (value, vtype) in enumerate(zip(row, rowdesc)): + if vtype in ('Date', 'Datetime') and type(value) is unicode: + found_date = True + try: + row[cellindex] = strptime(value, '%Y-%m-%d %H:%M:%S') + except: + row[cellindex] = strptime(value, '%Y-%m-%d') + if vtype == 'Time' and type(value) is unicode: + found_date = True + try: + row[cellindex] = strptime(value, '%H:%M:%S') + except: + # DateTime used as Time? + row[cellindex] = strptime(value, '%Y-%m-%d %H:%M:%S') + if vtype == 'Interval' and type(value) is int: + found_date = True + row[cellindex] = DateTimeDelta(0, 0, 0, value) + if not found_date: + break + return rset + return new_execute + querier.__class__.execute = wrap_execute(querier.__class__.execute) + + +def init_test_database(driver='sqlite', configdir='data', config=None, + vreg=None): + """init a test database for a specific driver""" + from cubicweb.dbapi import in_memory_cnx + if vreg and not config: + config = vreg.config + config = config or TestServerConfiguration(configdir) + source = config.sources() + if driver == 'sqlite': + init_test_database_sqlite(config, source) + elif driver == 'postgres': + init_test_database_postgres(config, source) + else: + raise ValueError('no initialization function for driver %r' % driver) + config._cubes = None # avoid assertion error + repo, cnx = in_memory_cnx(vreg or config, unicode(source['admin']['login']), + source['admin']['password'] or 'xxx') + if driver == 'sqlite': + install_sqlite_path(repo.querier) + return repo, cnx + +def init_test_database_postgres(config, source, vreg=None): + """initialize a fresh sqlite databse used for testing purpose""" + if config.init_repository: + from cubicweb.server import init_repository + init_repository(config, interactive=False, drop=True, vreg=vreg) + +def cleanup_sqlite(dbfile, removecube=False): + try: + os.remove(dbfile) + os.remove('%s-journal' % dbfile) + except OSError: + pass + if removecube: + try: + os.remove('%s-cube' % dbfile) + except OSError: + pass + +def init_test_database_sqlite(config, source, vreg=None): + """initialize a fresh sqlite databse used for testing purpose""" + import shutil + # remove database file if it exists (actually I know driver == 'sqlite' :) + dbfile = source['system']['db-name'] + cleanup_sqlite(dbfile) + cube = '%s-cube' % dbfile + if exists(cube): + shutil.copy(cube, dbfile) + else: + # initialize the database + from cubicweb.server import init_repository + init_repository(config, interactive=False, vreg=vreg) + shutil.copy(dbfile, cube)