"""Hidden internals for the devtools.apptest module
: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 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 CubicWebRegistry
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 = ('EGroup', 'EUser',
'EFRDef', 'ENFRDef',
'EConstraint', 'EConstraintType', 'EProperty',
'EEType', 'ERType',
'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 EGroup, and EUser...
"""
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):
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 = CubicWebRegistry(config)
self.admlogin = source['db-user']
# restore database <=> init database
self.restore_database()
if verbose:
print "init done"
login = source['db-user']
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('EUser', '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 EUser 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):
if login == self.admlogin:
self.restore_connection()
else:
self.cnx = repo_connect(self.repo, unicode(login), 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 {}))
def check_view(self, rql, vid, optional_args, template='main'):
"""checks if vreg.view() raises an exception in this environment
If any exception is raised in this method, it will be considered
as a TestFailure
"""
return self.call_view(vid, rql,
template=template, optional_args=optional_args)
def call_view(self, vid, rql, template='main', optional_args=None):
"""shortcut for self.vreg.view()"""
assert template
if optional_args is None:
optional_args = {}
optional_args['vid'] = vid
req = self.create_request(rql=rql, **optional_args)
return self.vreg.main_template(req, template)
def call_edit(self, req):
"""shortcut for self.app.edit()"""
controller = self.app.select_controller('edit', req)
try:
controller.publish()
except Redirect:
result = 'success'
else:
raise Exception('edit should raise Redirect on success')
req.cnx.commit()
return result
def iter_possible_views(self, req, rset):
"""returns a list of possible vids for <rql>"""
for view in self.vreg.possible_views(req, rset):
if view.category == 'startupview':
continue
yield view.id
if rset.rowcount == 1:
yield 'edition'
def iter_startup_views(self, req):
"""returns the list of startup views"""
for view in self.vreg.possible_views(req, None):
if view.category != 'startupview':
continue
yield view.id
def iter_possible_actions(self, req, rset):
"""returns a list of possible vids for <rql>"""
for action in self.vreg.possible_vobjects('actions', req, rset):
yield action
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 = CubicWebRegistry(config)
repo, self.cnx = init_test_database(driver=source['db-driver'],
vreg=self.vreg)
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
login = source['db-user']
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()