--- a/devtools/testlib.py Wed Jun 11 14:52:27 2014 +0200
+++ b/devtools/testlib.py Wed Jun 11 15:58:01 2014 +0200
@@ -156,30 +156,6 @@
cwconfig.SMTP = MockSMTP
-class TestCaseConnectionProxy(object):
- """thin wrapper around `cubicweb.repoapi.Connection` context-manager
- used in CubicWebTC (cf. `cubicweb.devtools.testlib.CubicWebTC.login` method)
-
- It just proxies to the default connection context manager but
- restores the original connection on exit.
- """
- def __init__(self, testcase, cnx):
- self.testcase = testcase
- self.cnx = cnx
-
- def __getattr__(self, attrname):
- return getattr(self.cnx, attrname)
-
- def __enter__(self):
- # already open
- return self.cnx
-
- def __exit__(self, exctype, exc, tb):
- try:
- return self.cnx.__exit__(exctype, exc, tb)
- finally:
- self.testcase.restore_connection()
-
# Repoaccess utility ###############################################3###########
class RepoAccess(object):
@@ -291,17 +267,12 @@
requestcls = fake.FakeRequest
tags = TestCase.tags | Tags('cubicweb', 'cw_repo')
test_db_id = DEFAULT_EMPTY_DB_ID
- _cnxs = set() # establised connection
- # stay on connection for leak detection purpose
# anonymous is logged by default in cubicweb test cases
anonymous_allowed = True
def __init__(self, *args, **kwargs):
self._admin_session = None
- self._admin_cnx = None
- self._current_session = None
- self._current_cnx = None
self.repo = None
self._open_access = set()
super(CubicWebTC, self).__init__(*args, **kwargs)
@@ -324,80 +295,11 @@
except BadConnectionId:
continue # already closed
- @deprecated('[3.19] explicitly use RepoAccess object in test instead')
- def set_cnx(self, cnx):
- assert getattr(cnx, '_session', None) is not None
- if cnx is self._admin_cnx:
- self._pop_custom_cnx()
- else:
- self._cnxs.add(cnx) # register the cnx to make sure it is removed
- self._current_session = cnx.session
- self._current_cnx = cnx
-
@property
- @deprecated('[3.19] explicitly use RepoAccess object in test instead')
- def cnx(self):
- # XXX we want to deprecate this
- cnx = self._current_cnx
- if cnx is None:
- cnx = self._admin_cnx
- return cnx
-
- def _close_cnx(self):
- """ensure that all cnx used by a test have been closed"""
- for cnx in list(self._cnxs):
- cnx.rollback()
- self._cnxs.remove(cnx)
-
- @property
- @deprecated('[3.19] explicitly use RepoAccess object in test instead')
def session(self):
- """return current server side session"""
- session = self._current_session
- if session is None:
- session = self._admin_session
- return session
-
- @property
- @deprecated('[3.19] explicitly use RepoAccess object in test instead')
- def websession(self):
- return self.session
-
- @property
- @deprecated('[3.19] explicitly use RepoAccess object in test instead')
- def adminsession(self):
- """return current server side session (using default manager account)"""
+ """return admin session"""
return self._admin_session
- @deprecated('[3.19] explicitly use RepoAccess object in test instead')
- def login(self, login, **kwargs):
- """return a connection for the given login/password"""
- __ = kwargs.pop('autoclose', True) # not used anymore
- if login == self.admlogin:
- # undo any previous login, if we're not used as a context manager
- self.restore_connection()
- return self.cnx
- else:
- if not kwargs:
- kwargs['password'] = str(login)
- cnx = repoapi.connect(self.repo, login, **kwargs)
- self.set_cnx(cnx)
- cnx.__enter__()
- return TestCaseConnectionProxy(self, cnx)
-
- @deprecated('[3.19] explicitly use RepoAccess object in test instead')
- def restore_connection(self):
- self._pop_custom_cnx()
-
- def _pop_custom_cnx(self):
- if self._current_cnx is not None:
- if self._current_cnx._open:
- self._current_cnx.close()
- if not self._current_session.closed:
- self.repo.close(self._current_session.sessionid)
- self._current_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.
@@ -411,9 +313,6 @@
login = unicode(db_handler.config.default_admin_config['login'])
self.admin_access = self.new_access(login)
self._admin_session = self.admin_access._session
- self._admin_cnx = repoapi.Connection(self._admin_session)
- self._cnxs.add(self._admin_cnx)
- self._admin_cnx.__enter__()
self.config.repository = lambda x=None: self.repo
@@ -482,15 +381,6 @@
"""return the application schema"""
return self.vreg.schema
- @deprecated('[3.19] explicitly use RepoAccess object in test instead')
- def shell(self):
- """return a shell session object"""
- from cubicweb.server.migractions import ServerMigrationHelper
- return ServerMigrationHelper(None, repo=self.repo, cnx=self.cnx,
- interactive=False,
- # hack so it don't try to load fs schema
- schema=1)
-
def set_option(self, optname, value):
self.config.global_set_option(optname, value)
@@ -511,19 +401,15 @@
self.skipTest('repository is not initialised: %r' % previous_failure)
try:
self._init_repo()
- self.addCleanup(self._close_cnx)
except Exception as ex:
self.__class__._repo_init_failed = ex
raise
self.addCleanup(self._close_access)
self.setup_database()
- self._admin_cnx.commit()
MAILBOX[:] = [] # reset mailbox
def tearDown(self):
# XXX hack until logilab.common.testlib is fixed
- if self._admin_cnx is not None:
- self._admin_cnx = None
if self._admin_session is not None:
self.repo.close(self._admin_session.sessionid)
self._admin_session = None
@@ -565,19 +451,10 @@
def create_user(self, req, login=None, groups=('users',), password=None,
email=None, commit=True, **kwargs):
"""create and return a new user entity"""
- if isinstance(req, basestring):
- warn('[3.12] create_user arguments are now (req, login[, groups, password, commit, **kwargs])',
- DeprecationWarning, stacklevel=2)
- if not isinstance(groups, (tuple, list)):
- password = groups
- groups = login
- elif isinstance(login, tuple):
- groups = login
- req = self._admin_cnx
+ if password is None:
+ password = login
if login is not None:
login = unicode(login)
- if password is None:
- password = login
user = req.create_entity('CWUser', login=login,
upassword=password, **kwargs)
req.execute('SET X in_group G WHERE X eid %%(x)s, G name IN(%s)'