Skip to content

GitLab

Commit 09ad984d authored by Sylvain Thénault's avatar Sylvain Thénault
Browse files

merge

parents 8074dd88e21b b2530e3e0afb
Expand all Hide whitespace changes
Inline Side-by-side
common/test/unittest_mail.py
View file @ a9a2dca5
......@@ -13,7 +13,7 @@ import pwd
from logilab.common.testlib import unittest_main
from logilab.common.umessage import message_from_string
from cubicweb.devtools.apptest import EnvBasedTC
from cubicweb.devtools.testlib import CubicWebTC
from cubicweb.common.mail import format_mail
......@@ -25,7 +25,7 @@ def getlogin():
return pwd.getpwuid(os.getuid())[0]
class EmailTC(EnvBasedTC):
class EmailTC(CubicWebTC):
def test_format_mail(self):
self.set_option('sender-addr', 'bim@boum.fr')
......
common/test/unittest_migration.py
View file @ a9a2dca5
......@@ -10,8 +10,6 @@ from os.path import abspath
from logilab.common.testlib import TestCase, unittest_main
from cubicweb.devtools import TestServerConfiguration
from cubicweb.devtools.apptest import TestEnvironment
from cubicweb.cwconfig import CubicWebConfiguration
from cubicweb.common.migration import MigrationHelper, filter_scripts
from cubicweb.server.migractions import ServerMigrationHelper
......@@ -98,8 +96,8 @@ class BaseCreationTC(TestCase):
config = ApptestConfiguration('data')
source = config.sources()['system']
self.assertEquals(source['db-driver'], 'sqlite')
cleanup_sqlite(source['db-name'], removecube=True)
init_test_database(driver=source['db-driver'], config=config)
cleanup_sqlite(source['db-name'], removetemplate=True)
init_test_database(config=config)
if __name__ == '__main__':
......
common/test/unittest_mixins.py
View file @ a9a2dca5
......@@ -6,9 +6,9 @@
:license: GNU Lesser General Public License, v2.1 - http://www.gnu.org/licenses
"""
from logilab.common.testlib import unittest_main
from cubicweb.devtools.apptest import EnvBasedTC
from cubicweb.devtools.testlib import CubicWebTC
class WorkfloableMixInTC(EnvBasedTC):
class WorkfloableMixInTC(CubicWebTC):
def test_wf_state(self):
s = self.add_entity('State', name=u'activated')
self.execute('SET X state_of ET WHERE ET name "Bookmark", X eid %(x)s',
......
devtools/__init__.py
View file @ a9a2dca5
......@@ -13,20 +13,54 @@ from datetime import timedelta
from os.path import (abspath, join, exists, basename, dirname, normpath, split,
isfile, isabs)
from cubicweb import CW_SOFTWARE_ROOT, ConfigurationError
from cubicweb import CW_SOFTWARE_ROOT, ConfigurationError, schema, cwconfig
from cubicweb.utils import strptime
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
cwconfig.CubicWebConfiguration.cls_adjust_sys_path()
# db auto-population configuration #############################################
SYSTEM_ENTITIES = schema.SCHEMA_TYPES | set((
'CWGroup', 'CWUser', 'CWProperty',
'State', 'Transition', 'TrInfo',
))
SYSTEM_RELATIONS = schema.META_RTYPES | set((
# workflow related
'state_of', 'transition_of', 'initial_state', 'allowed_transition',
'destination_state', 'in_state', 'wf_info_for', 'from_state', 'to_state',
'condition',
# cwproperty
'for_user',
# schema definition
'specializes',
'relation_type', 'from_entity', 'to_entity',
'constrained_by', 'cstrtype', 'widget',
'read_permission', 'update_permission', 'delete_permission', 'add_permission',
# permission
'in_group', 'require_group', 'require_permission',
# deducted from other relations
'primary_email',
))
# content validation configuration #############################################
# 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
# {'vid': validator}
VIEW_VALIDATORS = {}
# cubicweb test configuration ##################################################
BASE_URL = 'http://testing.fr/cubicweb/'
DEFAULT_SOURCES = {'system': {'adapter' : 'native',
'db-encoding' : 'UTF-8', #'ISO-8859-1',
'db-user' : u'admin',
......@@ -40,13 +74,14 @@ DEFAULT_SOURCES = {'system': {'adapter' : 'native',
},
}
class TestServerConfiguration(ServerConfiguration):
mode = 'test'
set_language = False
read_instance_schema = False
bootstrap_schema = False
init_repository = True
options = merge_options(ServerConfiguration.options + (
options = cwconfig.merge_options(ServerConfiguration.options + (
('anonymous-user',
{'type' : 'string',
'default': None,
......@@ -66,7 +101,6 @@ class TestServerConfiguration(ServerConfiguration):
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()
......@@ -118,30 +152,11 @@ class TestServerConfiguration(ServerConfiguration):
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)
options = cwconfig.merge_options(TestServerConfiguration.options
+ TwistedConfiguration.options)
cubicweb_appobject_path = TestServerConfiguration.cubicweb_appobject_path | TwistedConfiguration.cubicweb_appobject_path
cube_appobject_path = TestServerConfiguration.cube_appobject_path | TwistedConfiguration.cube_appobject_path
......@@ -163,98 +178,91 @@ class ApptestConfiguration(BaseApptestConfiguration):
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('.*'))
self.global_set_option('anonymous-user', 'anon')
self.global_set_option('anonymous-password', 'anon')
def load_configuration(self):
super(ApptestConfiguration, self).load_configuration()
self.global_set_option('anonymous-user', 'anon')
self.global_set_option('anonymous-password', 'anon')
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
# 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':
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']),
sources['admin']['password'] or 'xxx')
if driver == 'sqlite':
install_sqlite_patch(repo.querier)
return repo, cnx
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 reset_test_database(config):
"""init a test database for a specific driver"""
driver = config.sources()['system']['db-driver']
if driver == 'sqlite':
reset_test_database_sqlite(config)
else:
raise ValueError('no reset function for driver %r' % driver)
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})
### postgres test database handling ############################################
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
# instance 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:
# cube test
self.apphome = abspath('..')
self.sourcefile = sourcefile
self.global_set_option('realm', '')
self.use_pyro = pyro_name is not None
def init_test_database_postgres(config):
"""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)
def pyro_enabled(self):
if self.use_pyro:
return True
else:
return False
CubicWebConfiguration.cls_adjust_sys_path()
### sqlite test database handling ##############################################
def cleanup_sqlite(dbfile, removetemplate=False):
try:
os.remove(dbfile)
os.remove('%s-journal' % dbfile)
except OSError:
pass
if removetemplate:
try:
os.remove('%s-template' % dbfile)
except OSError:
pass
def reset_test_database_sqlite(config):
import shutil
dbfile = config.sources()['system']['db-name']
cleanup_sqlite(dbfile)
template = '%s-template' % dbfile
if exists(template):
shutil.copy(template, dbfile)
return True
return False
def init_test_database_sqlite(config):
"""initialize a fresh sqlite databse used for testing purpose"""
# remove database file if it exists
dbfile = config.sources()['system']['db-name']
if not reset_test_database_sqlite(config):
# initialize the database
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 :
......@@ -293,57 +301,3 @@ def install_sqlite_patch(querier):
return new_execute
querier.__class__.execute = wrap_execute(querier.__class__.execute)
querier.__class__._devtools_sqlite_patched = True
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_patch(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-template' % 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)
template = '%s-template' % dbfile
if exists(template):
shutil.copy(template, dbfile)
else:
# initialize the database
from cubicweb.server import init_repository
init_repository(config, interactive=False, vreg=vreg)
shutil.copy(dbfile, template)
devtools/_apptest.py deleted 100644 → 0
View file @ 8074dd88
"""Hidden internals for the devtools.apptest module
:organization: Logilab
:copyright: 2001-2009 LOGILAB S.A. (Paris, FRANCE), license is LGPL v2.
:contact: http://www.logilab.fr/ -- mailto:contact@logilab.fr
:license: GNU Lesser General Public License, v2.1 - http://www.gnu.org/licenses
"""
__docformat__ = "restructuredtext en"
import sys, traceback
from logilab.common.pytest import pause_tracing, resume_tracing
import yams.schema
from cubicweb.dbapi import repo_connect, ConnectionProperties, ProgrammingError
from cubicweb.cwvreg import CubicWebVRegistry
from cubicweb.web.application import CubicWebPublisher
from cubicweb.web import Redirect
from cubicweb.devtools import ApptestConfiguration, init_test_database
from cubicweb.devtools.fake import FakeRequest
SYSTEM_ENTITIES = ('CWGroup', 'CWUser',
'CWAttribute', 'CWRelation',
'CWConstraint', 'CWConstraintType', 'CWProperty',
'CWEType', 'CWRType',
'State', 'Transition', 'TrInfo',
'RQLExpression',
)
SYSTEM_RELATIONS = (
# virtual relation
'identity',
# metadata
'is', 'is_instance_of', 'owned_by', 'created_by', 'specializes',
# workflow related
'state_of', 'transition_of', 'initial_state', 'allowed_transition',
'destination_state', 'in_state', 'wf_info_for', 'from_state', 'to_state',
'condition',
# permission
'in_group', 'require_group', 'require_permission',
'read_permission', 'update_permission', 'delete_permission', 'add_permission',
# eproperty
'for_user',
# schema definition
'relation_type', 'from_entity', 'to_entity',
'constrained_by', 'cstrtype', 'widget',
# deducted from other relations
'primary_email',
)
def unprotected_entities(app_schema, strict=False):
"""returned a Set of each non final entity type, excluding CWGroup, and CWUser...
"""
if strict:
protected_entities = yams.schema.BASE_TYPES
else:
protected_entities = yams.schema.BASE_TYPES.union(set(SYSTEM_ENTITIES))
entities = set(app_schema.entities())
return entities - protected_entities
def ignore_relations(*relations):
global SYSTEM_RELATIONS
SYSTEM_RELATIONS += relations
class TestEnvironment(object):
"""TestEnvironment defines a context (e.g. a config + a given connection) in
which the tests are executed
"""
def __init__(self, appid, reporter=None, verbose=False,
configcls=ApptestConfiguration, requestcls=FakeRequest):
config = configcls(appid)
self.requestcls = requestcls
self.cnx = None
config.db_perms = False
source = config.sources()['system']
if verbose:
print "init test database ..."
self.vreg = vreg = CubicWebVRegistry(config)
self.admlogin = source['db-user']
# restore database <=> init database
self.restore_database()
if verbose:
print "init done"
config.repository = lambda x=None: self.repo
self.app = CubicWebPublisher(config, vreg=vreg)
self.verbose = verbose
schema = self.vreg.schema
# else we may run into problems since email address are ususally share in app tests
# XXX should not be necessary anymore
schema.rschema('primary_email').set_rproperty('CWUser', 'EmailAddress', 'composite', False)
self.deletable_entities = unprotected_entities(schema)
def restore_database(self):
"""called by unittests' tearDown to restore the original database
"""
try:
pause_tracing()
if self.cnx:
self.cnx.close()
source = self.vreg.config.sources()['system']
self.repo, self.cnx = init_test_database(driver=source['db-driver'],
vreg=self.vreg)
self._orig_cnx = self.cnx
resume_tracing()
except:
resume_tracing()
traceback.print_exc()
sys.exit(1)
# XXX cnx decoration is usually done by the repository authentication manager,
# necessary in authentication tests
self.cnx.vreg = self.vreg
self.cnx.login = source['db-user']
self.cnx.password = source['db-password']
def create_user(self, login, groups=('users',), req=None):
req = req or self.create_request()
cursor = self._orig_cnx.cursor(req)
rset = cursor.execute('INSERT CWUser X: X login %(login)s, X upassword %(passwd)s,'
'X in_state S WHERE S name "activated"',
{'login': unicode(login), 'passwd': login.encode('utf8')})
user = rset.get_entity(0, 0)
cursor.execute('SET X in_group G WHERE X eid %%(x)s, G name IN(%s)'
% ','.join(repr(g) for g in groups),
{'x': user.eid}, 'x')
user.clear_related_cache('in_group', 'subject')
self._orig_cnx.commit()
return user
def login(self, login, password=None):
if login == self.admlogin:
self.restore_connection()
else:
self.cnx = repo_connect(self.repo, unicode(login),
password or str(login),
ConnectionProperties('inmemory'))
if login == self.vreg.config.anonymous_user()[0]:
self.cnx.anonymous_connection = True
return self.cnx
def restore_connection(self):
if not self.cnx is self._orig_cnx:
try:
self.cnx.close()
except ProgrammingError:
pass # already closed
self.cnx = self._orig_cnx
############################################################################
def execute(self, rql, args=None, eidkey=None, req=None):
"""executes <rql>, builds a resultset, and returns a couple (rset, req)
where req is a FakeRequest
"""
req = req or self.create_request(rql=rql)
return self.cnx.cursor(req).execute(unicode(rql), args, eidkey)
def create_request(self, rql=None, **kwargs):
"""executes <rql>, builds a resultset, and returns a
couple (rset, req) where req is a FakeRequest
"""
if rql:
kwargs['rql'] = rql
req = self.requestcls(self.vreg, form=kwargs)
req.set_connection(self.cnx)
return req
def get_rset_and_req(self, rql, optional_args=None, args=None, eidkey=None):
"""executes <rql>, builds a resultset, and returns a
couple (rset, req) where req is a FakeRequest
"""
return (self.execute(rql, args, eidkey),
self.create_request(rql=rql, **optional_args or {}))
class ExistingTestEnvironment(TestEnvironment):
def __init__(self, appid, sourcefile, verbose=False):
config = ApptestConfiguration(appid, sourcefile=sourcefile)
if verbose:
print "init test database ..."
source = config.sources()['system']
self.vreg = CubicWebVRegistry(config)
self.cnx = init_test_database(driver=source['db-driver'],
vreg=self.vreg)[1]
if verbose:
print "init done"
self.app = CubicWebPublisher(config, vreg=self.vreg)
self.verbose = verbose
# this is done when the publisher is opening a connection
self.cnx.vreg = self.vreg
def setup(self, config=None):
"""config is passed by TestSuite but is ignored in this environment"""
cursor = self.cnx.cursor()
self.last_eid = cursor.execute('Any X WHERE X creation_date D ORDERBY D DESC LIMIT 1').rows[0][0]
def cleanup(self):
"""cancel inserted elements during tests"""
cursor = self.cnx.cursor()
cursor.execute('DELETE Any X WHERE X eid > %(x)s', {'x' : self.last_eid}, eid_key='x')
print "cleaning done"
self.cnx.commit()
devtools/apptest.py deleted 100644 → 0
View file @ 8074dd88
"""This module provides misc utilities to test instances
:organization: Logilab