[dbapi] retire repo.execute, which was used by the dbapi
We will retire or reform pieces, mostly of the Repository class,
that were related to the old dbapi way of life.
Related to #3933480.
--- a/hooks/test/unittest_hooks.py Thu Jun 05 09:51:57 2014 +0200
+++ b/hooks/test/unittest_hooks.py Wed Jun 04 18:21:24 2014 +0200
@@ -146,20 +146,6 @@
class UserGroupHooksTC(CubicWebTC):
- def test_user_synchronization(self):
- with self.admin_access.repo_cnx() as cnx:
- self.create_user(cnx, 'toto', password='hop', commit=False)
- self.assertRaises(AuthenticationError,
- self.repo.connect, u'toto', password='hop')
- cnx.commit()
- cnxid = self.repo.connect(u'toto', password='hop')
- self.assertNotEqual(cnxid, cnx.sessionid)
- cnx.execute('DELETE CWUser X WHERE X login "toto"')
- self.repo.execute(cnxid, 'State X')
- cnx.commit()
- self.assertRaises(BadConnectionId,
- self.repo.execute, cnxid, 'State X')
-
def test_user_group_synchronization(self):
with self.admin_access.repo_cnx() as cnx:
user = cnx.user
--- a/server/repository.py Thu Jun 05 09:51:57 2014 +0200
+++ b/server/repository.py Wed Jun 04 18:21:24 2014 +0200
@@ -664,30 +664,6 @@
"""open a new session for a given user and return its sessionid """
return self.new_session(login, **kwargs).sessionid
- def execute(self, sessionid, rqlstring, args=None, build_descr=True,
- txid=None):
- """execute a RQL query
-
- * rqlstring should be a unicode string or a plain ascii string
- * args the optional parameters used in the query
- * build_descr is a flag indicating if the description should be
- built on select queries
- """
- session = self._get_session(sessionid, setcnxset=True, txid=txid)
- try:
- try:
- rset = self.querier.execute(session, rqlstring, args,
- build_descr)
- return rset
- except (ValidationError, Unauthorized, RQLSyntaxError):
- raise
- except Exception:
- # FIXME: check error to catch internal errors
- self.exception('unexpected error while executing %s with %s', rqlstring, args)
- raise
- finally:
- session.free_cnxset()
-
@deprecated('[3.19] use .entity_metas(sessionid, eid, txid) instead')
def describe(self, sessionid, eid, txid=None):
"""return a tuple `(type, physical source uri, extid, actual source
--- a/server/test/unittest_repository.py Thu Jun 05 09:51:57 2014 +0200
+++ b/server/test/unittest_repository.py Wed Jun 04 18:21:24 2014 +0200
@@ -92,35 +92,17 @@
self.assertRaises(AuthenticationError,
self.repo.connect, None)
- def test_execute(self):
+ def test_login_upassword_accent(self):
+ with self.admin_access.repo_cnx() as cnx:
+ cnx.execute('INSERT CWUser X: X login %(login)s, X upassword %(passwd)s, '
+ 'X in_group G WHERE G name "users"',
+ {'login': u"barnabé", 'passwd': u"héhéhé".encode('UTF8')})
+ cnx.commit()
repo = self.repo
- cnxid = repo.connect(self.admlogin, password=self.admpassword)
- repo.execute(cnxid, 'Any X')
- repo.execute(cnxid, 'Any X where X is Personne')
- repo.execute(cnxid, 'Any X where X is Personne, X nom ~= "to"')
- repo.execute(cnxid, 'Any X WHERE X has_text %(text)s', {'text': u'\xe7a'})
- repo.close(cnxid)
-
- def test_login_upassword_accent(self):
- repo = self.repo
- cnxid = repo.connect(self.admlogin, password=self.admpassword)
- repo.execute(cnxid, 'INSERT CWUser X: X login %(login)s, X upassword %(passwd)s, X in_group G WHERE G name "users"',
- {'login': u"barnabé", 'passwd': u"héhéhé".encode('UTF8')})
- repo.commit(cnxid)
- repo.close(cnxid)
cnxid = repo.connect(u"barnabé", password=u"héhéhé".encode('UTF8'))
self.assert_(cnxid)
repo.close(cnxid)
- def test_rollback_on_commit_error(self):
- cnxid = self.repo.connect(self.admlogin, password=self.admpassword)
- self.repo.execute(cnxid,
- 'INSERT CWUser X: X login %(login)s, X upassword %(passwd)s',
- {'login': u"tutetute", 'passwd': 'tutetute'})
- self.assertRaises(ValidationError, self.repo.commit, cnxid)
- self.assertFalse(self.repo.execute(cnxid, 'CWUser X WHERE X login "tutetute"'))
- self.repo.close(cnxid)
-
def test_rollback_on_execute_validation_error(self):
class ValidationErrorAfterHook(Hook):
__regid__ = 'valerror-after-hook'
@@ -165,11 +147,6 @@
cnxid = repo.connect(self.admlogin, password=self.admpassword)
self.assert_(cnxid)
repo.close(cnxid)
- self.assertRaises(BadConnectionId, repo.execute, cnxid, 'Any X')
-
- def test_invalid_cnxid(self):
- self.assertRaises(BadConnectionId, self.repo.execute, 0, 'Any X')
- self.assertRaises(BadConnectionId, self.repo.close, None)
def test_shared_data(self):
repo = self.repo
@@ -198,76 +175,6 @@
repo.close(cnxid)
self.assertRaises(BadConnectionId, repo.check_session, cnxid)
- def test_transaction_base(self):
- repo = self.repo
- cnxid = repo.connect(self.admlogin, password=self.admpassword)
- # check db state
- result = repo.execute(cnxid, 'Personne X')
- self.assertEqual(result.rowcount, 0)
- # rollback entity insertion
- repo.execute(cnxid, "INSERT Personne X: X nom 'bidule'")
- result = repo.execute(cnxid, 'Personne X')
- self.assertEqual(result.rowcount, 1)
- repo.rollback(cnxid)
- result = repo.execute(cnxid, 'Personne X')
- self.assertEqual(result.rowcount, 0, result.rows)
- # commit
- repo.execute(cnxid, "INSERT Personne X: X nom 'bidule'")
- repo.commit(cnxid)
- result = repo.execute(cnxid, 'Personne X')
- self.assertEqual(result.rowcount, 1)
- repo.close(cnxid)
-
- def test_transaction_base2(self):
- repo = self.repo
- cnxid = repo.connect(self.admlogin, password=self.admpassword)
- # rollback relation insertion
- repo.execute(cnxid, "SET U in_group G WHERE U login 'admin', G name 'guests'")
- result = repo.execute(cnxid, "Any U WHERE U in_group G, U login 'admin', G name 'guests'")
- self.assertEqual(result.rowcount, 1)
- repo.rollback(cnxid)
- result = repo.execute(cnxid, "Any U WHERE U in_group G, U login 'admin', G name 'guests'")
- self.assertEqual(result.rowcount, 0, result.rows)
- repo.close(cnxid)
-
- def test_transaction_base3(self):
- repo = self.repo
- cnxid = repo.connect(self.admlogin, password=self.admpassword)
- # rollback state change which trigger TrInfo insertion
- session = repo._get_session(cnxid)
- user = session.user
- user.cw_adapt_to('IWorkflowable').fire_transition('deactivate')
- rset = repo.execute(cnxid, 'TrInfo T WHERE T wf_info_for X, X eid %(x)s', {'x': user.eid})
- self.assertEqual(len(rset), 1)
- repo.rollback(cnxid)
- rset = repo.execute(cnxid, 'TrInfo T WHERE T wf_info_for X, X eid %(x)s', {'x': user.eid})
- self.assertEqual(len(rset), 0)
- repo.close(cnxid)
-
- def test_close_kill_processing_request(self):
- repo = self.repo
- cnxid = repo.connect(self.admlogin, password=self.admpassword)
- repo.execute(cnxid, 'INSERT CWUser X: X login "toto", X upassword "tutu", X in_group G WHERE G name "users"')
- repo.commit(cnxid)
- lock = threading.Lock()
- lock.acquire()
- # close has to be in the thread due to sqlite limitations
- def close_in_a_few_moment():
- lock.acquire()
- repo.close(cnxid)
- t = threading.Thread(target=close_in_a_few_moment)
- t.start()
- def run_transaction():
- lock.release()
- repo.execute(cnxid, 'DELETE CWUser X WHERE X login "toto"')
- repo.commit(cnxid)
- try:
- with self.assertRaises(SessionClosedError) as cm:
- run_transaction()
- self.assertEqual(str(cm.exception), 'try to access connections set on a closed session %s' % cnxid)
- finally:
- t.join()
-
def test_initial_schema(self):
schema = self.repo.schema
# check order of attributes is respected