cubicweb: comparison devtools/

equal deleted inserted replaced

-:672acc730ce5
+:d628defebc17
 """Hidden internals for the devtools.apptest module
 :organization: Logilab
-:copyright: 2001-2008 LOGILAB S.A. (Paris, FRANCE), all rights reserved.
+:copyright: 2001-2009 LOGILAB S.A. (Paris, FRANCE), all rights reserved.
 :contact: http://www.logilab.fr/ -- mailto:contact@logilab.fr
 """
 __docformat__ = "restructuredtext en"
 import sys, traceback
 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',
 '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...
 """
 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
 # 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,'
 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:
 """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 = {}
 """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)
 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]
 """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()

branch	tls-sprint
changeset 1802	d628defebc17
parent 1398	5fe84a5f7035
child 1977	606923dff11b