devtools/_apptest.py
branchstable
changeset 4556 43c14e0e8972
parent 4555 8968c50818db
parent 4553 23201259ffeb
child 4560 23e0632df615
--- a/devtools/_apptest.py	Fri Feb 12 12:57:14 2010 +0100
+++ /dev/null	Thu Jan 01 00:00:00 1970 +0000
@@ -1,207 +0,0 @@
-"""Hidden internals for the devtools.apptest module
-
-:organization: Logilab
-:copyright: 2001-2010 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',
-                   'Workflow', 'State', 'BaseTransition', 'Transition', 'WorkflowTransition', 'TrInfo', 'SubWorkflowExitPoint',
-                   'RQLExpression',
-                   )
-SYSTEM_RELATIONS = (
-    # virtual relation
-    'identity',
-    # metadata
-    'is', 'is_instance_of', 'owned_by', 'created_by', 'specializes',
-    # workflow related
-    'workflow_of', 'state_of', 'transition_of', 'initial_state', 'allowed_transition',
-    'destination_state', 'in_state', 'wf_info_for', 'from_state', 'to_state',
-    'condition', 'subworkflow', 'subworkflow_state', 'subworkflow_exit',
-    # 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',
-                              {'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()