[testlib] introduce a RepoAccess class to easily create connection and request
Each RepoAccess hold a session for a user and three helper function to help
create Connection, ClientConnection of WebRequest related to this session.
related to #2920299
--- a/devtools/__init__.py Thu Jun 27 15:13:47 2013 +0200
+++ b/devtools/__init__.py Thu Jun 27 18:15:57 2013 +0200
@@ -267,7 +267,6 @@
skip_db_create_and_restore = True
read_instance_schema = True # read schema from database
-
# test database handling #######################################################
DEFAULT_EMPTY_DB_ID = '__default_empty_db__'
--- a/devtools/test/unittest_testlib.py Thu Jun 27 15:13:47 2013 +0200
+++ b/devtools/test/unittest_testlib.py Thu Jun 27 18:15:57 2013 +0200
@@ -172,5 +172,31 @@
self.assertTrue(rdef.permissions['add'])
self.assertTrue(rdef.permissions['read'], ())
+class RepoAccessTC(CubicWebTC):
+
+ def test_repo_connection(self):
+ acc = self.new_access('admin')
+ with acc.repo_cnx() as cnx:
+ rset = cnx.execute('Any X WHERE X is CWUser')
+ self.assertTrue(rset)
+
+ def test_client_connection(self):
+ acc = self.new_access('admin')
+ with acc.client_cnx() as cnx:
+ rset = cnx.execute('Any X WHERE X is CWUser')
+ self.assertTrue(rset)
+
+ def test_web_request(self):
+ acc = self.new_access('admin')
+ with acc.web_request(elephant='babar') as req:
+ rset = req.execute('Any X WHERE X is CWUser')
+ self.assertTrue(rset)
+ self.assertEqual('babar', req.form['elephant'])
+
+ def test_close(self):
+ acc = self.new_access('admin')
+ acc.close()
+
+
if __name__ == '__main__':
unittest_main()
--- a/devtools/testlib.py Thu Jun 27 15:13:47 2013 +0200
+++ b/devtools/testlib.py Thu Jun 27 18:15:57 2013 +0200
@@ -45,6 +45,7 @@
from cubicweb.sobjects import notification
from cubicweb.web import Redirect, application
from cubicweb.server.hook import SendMailOp
+from cubicweb.server.session import Session
from cubicweb.devtools import SYSTEM_ENTITIES, SYSTEM_RELATIONS, VIEW_VALIDATORS
from cubicweb.devtools import fake, htmlparser, DEFAULT_EMPTY_DB_ID
from cubicweb.utils import json
@@ -155,6 +156,76 @@
finally:
self.testcase.restore_connection()
+# Repoaccess utility ###############################################3###########
+
+class RepoAccess(object):
+ """An helper to easily create object to access the repo as a specific user
+
+ Each RepoAccess have it own session.
+
+ A repo access can create three type of object:
+
+ .. automethod:: cubicweb.testlib.RepoAccess.repo_cnx
+ .. automethod:: cubicweb.testlib.RepoAccess.client_cnx
+ .. automethod:: cubicweb.testlib.RepoAccess.web_request
+
+ The RepoAccess need to be closed to destroy the associated Session.
+ TestCase usually take care of this aspect for the user.
+
+ .. automethod:: cubicweb.testlib.RepoAccess.close
+ """
+
+ def __init__(self, repo, login, requestcls):
+ self._repo = repo
+ self._login = login
+ self.requestcls = requestcls
+ # opening session
+ #
+ # XXX this very hackish code should be cleaned and move on repo.
+ with repo.internal_cnx() as cnx:
+ rset = cnx.execute('CWUser U WHERE U login %(u)s', {'u': login})
+ user = rset.get_entity(0, 0)
+ user.groups
+ user.properties
+ self._session = Session(user, repo)
+ repo._sessions[self._session.id] = self._session
+ self._session.user._cw = self._session
+
+ @ contextmanager
+ def repo_cnx(self):
+ """Context manager returning a server side connection for the user"""
+ with self._session.new_cnx() as cnx:
+ yield cnx
+
+ @ contextmanager
+ def client_cnx(self):
+ """Context manager returning a client side connection for the user"""
+ with repoapi.ClientConnection(self._session) as cnx:
+ yield cnx
+
+ @ contextmanager
+ def web_request(self, url=None, headers={}, **kwargs):
+ """Context manager returning a web request pre-linked to a client cnx
+
+ To commit and rollback use::
+
+ req.cnx.commit()
+ req.cnx.rolback()
+ """
+ req = self.requestcls(self._repo.vreg, url=url, headers=headers, form=kwargs)
+ clt_cnx = repoapi.ClientConnection(self._session)
+ req.set_cnx(clt_cnx)
+ with clt_cnx:
+ yield req
+
+ def close(self):
+ """Close the session associated to the RepoAccess"""
+ if self._session is not None:
+ self._repo.close(self._session.id)
+ self._session = None
+
+
+
# base class for cubicweb tests requiring a full cw environments ###############
class CubicWebTC(TestCase):
@@ -186,9 +257,21 @@
self._current_session = None
self._current_clt_cnx = None
self.repo = None
+ self._open_access = set()
super(CubicWebTC, self).__init__(*args, **kwargs)
# repository connection handling ###########################################
+ def new_access(self, login):
+ """provide a new RepoAccess object for a given user
+
+ The access is automatically closed at the end of the test."""
+ access = RepoAccess(self.repo, login, self.requestcls)
+ self._open_access.add(access)
+ return access
+
+ def _close_access(self):
+ while self._open_access:
+ self._open_access.pop().close()
def set_cnx(self, cnx):
# XXX we want to deprecate this