cubicweb: comparison devtools/testlib.py

equal deleted inserted replaced

-:aff871b58ba0
+:4a803380f718
 from logilab.common.decorators import cached, classproperty, clear_cache, iclassmethod
 from logilab.common.deprecation import deprecated, class_deprecated
 from logilab.common.shellutils import getlogin
 from cubicweb import ValidationError, NoSelectableObject, AuthenticationError
-from cubicweb import cwconfig, dbapi, devtools, web, server
+from cubicweb import cwconfig, dbapi, devtools, web, server, repoapi
 from cubicweb.utils import json
 from cubicweb.sobjects import notification
 from cubicweb.web import Redirect, application
 from cubicweb.server.hook import SendMailOp
 from cubicweb.devtools import SYSTEM_ENTITIES, SYSTEM_RELATIONS, VIEW_VALIDATORS
 def __getattr__(self, attrname):
 return getattr(self.cnx, attrname)
 def __enter__(self):
-return self.cnx.__enter__()
+# already open
+return self.cnx
 def __exit__(self, exctype, exc, tb):
 try:
 return self.cnx.__exit__(exctype, exc, tb)
 finally:
-self.cnx.close()
 self.testcase.restore_connection()
 # base class for cubicweb tests requiring a full cw environments ###############
 class CubicWebTC(TestCase):
 test_db_id = DEFAULT_EMPTY_DB_ID
 _cnxs = set() # establised connection
 # stay on connection for leak detection purpose
 def __init__(self, *args, **kwargs):
-self._cnx = None  # current connection
+self._admin_session = None
+self._admin_clt_cnx = None
+self._current_session = None
+self._current_clt_cnx = None
 self.repo = None
-self.websession = None
 super(CubicWebTC, self).__init__(*args, **kwargs)
 # repository connection handling ###########################################
-# Too much complicated stuff. the class doesn't need to bear the repo anymore
 def set_cnx(self, cnx):
-self._cnxs.add(cnx)
+# XXX we want to deprecate this
-self._cnx = cnx
+assert getattr(cnx, '_session', None) is not None
+if cnx is self._admin_clt_cnx:
+self._pop_custom_cnx()
+else:
+self._cnxs.add(cnx) # register the cns to make sure it is removed
+self._current_session = cnx._session
+self._current_clt_cnx = cnx
 @property
 def cnx(self):
-return self._cnx
+# XXX we want to deprecate this
+clt_cnx = self._current_clt_cnx
+if clt_cnx is None:
+clt_cnx = self._admin_clt_cnx
+return clt_cnx
 def _close_cnx(self):
+"""ensure that all cnx used by a test have been closed"""
 for cnx in list(self._cnxs):
-if not cnx._closed:
+if cnx._open and not cnx._session.closed:
 cnx.rollback()
 cnx.close()
 self._cnxs.remove(cnx)
 @property
 def session(self):
-"""return current server side session (using default manager account)"""
+"""return current server side session"""
-session = self.repo._sessions[self.cnx.sessionid]
+# XXX We want to use a srv_connection instead and deprecate this
+# property
+session = self._current_session
+if session is None:
+session = self._admin_session
+session.set_cnx(self._admin_clt_cnx._cnxid)
 session.set_cnxset()
 return session
+@property
+def websession(self):
+return self.session
+@property
+def adminsession(self):
+"""return current server side session (using default manager account)"""
+return self._admin_session
 def login(self, login, **kwargs):
 """return a connection for the given login/password"""
+__ = kwargs.pop('autoclose', True) # not used anymore
 if login == self.admlogin:
-self.restore_connection()
 # definitly don't want autoclose when used as a context manager
-return self.cnx
+clt_cnx = repoapi.ClientConnection(self._admin_session)
-autoclose = kwargs.pop('autoclose', True)
+else:
 if not kwargs:
 kwargs['password'] = str(login)
-self.set_cnx(dbapi._repo_connect(self.repo, unicode(login), **kwargs))
+clt_cnx = repoapi.connect(self.repo, login, **kwargs)
-self.websession = dbapi.DBAPISession(self.cnx)
+self.set_cnx(clt_cnx)
-if autoclose:
+clt_cnx.__enter__()
-return TestCaseConnectionProxy(self, self.cnx)
+return TestCaseConnectionProxy(self, clt_cnx)
-return self.cnx
 def restore_connection(self):
-if not self.cnx is self._orig_cnx[0]:
+self._pop_custom_cnx()
-if not self.cnx._closed:
-self.cnx.close()
+def _pop_custom_cnx(self):
-cnx, self.websession = self._orig_cnx
+if self._current_clt_cnx is not None:
-self.set_cnx(cnx)
+if self._current_clt_cnx._open:
+self._current_clt_cnx.close()
+if not  self._current_session.closed:
+self.repo.close(self._current_session.id)
+self._current_clt_cnx = None
+self._current_session = None
 #XXX this doesn't need to a be classmethod anymore
 def _init_repo(self):
 """init the repository and connection to it.
 """
 self.init_config(self.config)
 # get or restore and working db.
 db_handler = devtools.get_test_db_handler(self.config)
 db_handler.build_db_cache(self.test_db_id, self.pre_setup_database)
-self.repo, cnx = db_handler.get_repo_and_cnx(self.test_db_id)
+db_handler.restore_database(self.test_db_id)
+self.repo = db_handler.get_repo(startup=True)
+# get an admin session (without actual login)
+sources = db_handler.config.sources()
+login = unicode(sources['admin']['login'])
+with self.repo.internal_session() as session:
+rset = session.execute('CWUser U WHERE U login %(u)s', {'u': login})
+user = rset.get_entity(0, 0)
+user.groups
+user.properties
+from cubicweb.server.session import Session
+self._admin_session = Session(user, self.repo)
+self.repo._sessions[self._admin_session.id] =  self._admin_session
+self._admin_session.user._cw = self._admin_session
+self._admin_clt_cnx = repoapi.ClientConnection(self._admin_session)
 # no direct assignation to cls.cnx anymore.
 # cnx is now an instance property that use a class protected attributes.
-self.set_cnx(cnx)
+self._cnxs.add(self._admin_clt_cnx)
-self.websession = dbapi.DBAPISession(cnx, self.admlogin)
+self._admin_clt_cnx.__enter__()
-self._orig_cnx = (cnx, self.websession)
 self.config.repository = lambda x=None: self.repo
 # db api ##################################################################
 @nocoverage
 def cursor(self, req=None):
-return self.cnx.cursor(req or self.request())
+if req is not None:
+return req.cnx
+else:
+return self.cnx
 @nocoverage
 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
 self.session.set_cnxset() # ensure cnxset still set after commit
 requestcls = fake.FakeRequest
 def request(self, rollbackfirst=False, url=None, headers={}, **kwargs):
 """return a web ui request"""
+if rollbackfirst:
+self.cnx.rollback()
 req = self.requestcls(self.vreg, url=url, headers=headers, form=kwargs)
-if rollbackfirst:
+req.set_cnx(self.cnx)
-self.websession.cnx.rollback()
-req.set_session(self.websession)
 return req
-@property
-def adminsession(self):
-"""return current server side session (using default manager account)"""
-return self.repo._sessions[self._orig_cnx[0].sessionid]
 # server side db api #######################################################
 def sexecute(self, rql, args=None, eid_key=None):
 if eid_key is not None:
 self.commit()
 MAILBOX[:] = [] # reset mailbox
 def tearDown(self):
 # XXX hack until logilab.common.testlib is fixed
+if self._admin_clt_cnx is not None:
+if self._admin_clt_cnx._open:
+self._admin_clt_cnx.close()
+self._admin_clt_cnx = None
+if self._admin_session is not None:
+if not self._admin_session.closed:
+self.repo.close(self._admin_session.id)
+self._admin_session = None
 while self._cleanups:
 cleanup, args, kwargs = self._cleanups.pop(-1)
 cleanup(*args, **kwargs)
 def _patch_SendMailOp(self):
 # user / session management ###############################################
 def user(self, req=None):
 """return the application schema"""
 if req is None:
-req = self.request()
+return self.request().user
-return self.cnx.user(req)
 else:
 return req.user
 @iclassmethod # XXX turn into a class method
 def create_user(self, req, login=None, groups=('users',), password=None,

changeset 9070	4a803380f718
parent 9044	cfec5cc46008
child 9071	46885bfa4150