[connection] allow simple instantiation of standalone Connection
authorPierre-Yves David <pierre-yves.david@logilab.fr>
Wed, 26 Jun 2013 14:26:06 +0200
changeset 9109 499db4fd03f8
parent 9108 6a4070e2849d
child 9110 ed8b383d94fd
[connection] allow simple instantiation of standalone Connection Such connection will automatically pick a connection id. Note, They are not automatically closed on session close. But they will fails to grab new cnxset once the session is closed.
server/session.py
server/test/unittest_session.py
--- a/server/session.py	Wed Jun 26 14:35:55 2013 +0200
+++ b/server/session.py	Wed Jun 26 14:26:06 2013 +0200
@@ -431,11 +431,16 @@
 
     is_request = False
 
-    def __init__(self, session, cnxid, session_handled=False):
+    def __init__(self, session, cnxid=None, session_handled=False):
         # using super(Connection, self) confuse some test hack
         RequestSessionBase.__init__(self, session.vreg)
+        # only the session provide explicite
+        if cnxid is not None:
+            assert session_handled # only session profive explicite cnxid
         #: connection unique id
         self._open = None
+        if cnxid is None:
+            cnxid = '%s-%s' % (session.id, uuid4().hex)
         self.connectionid = cnxid
         #: self._session_handled
         #: are the life cycle of this Connection automatically controlled by the
--- a/server/test/unittest_session.py	Wed Jun 26 14:35:55 2013 +0200
+++ b/server/test/unittest_session.py	Wed Jun 26 14:26:06 2013 +0200
@@ -17,7 +17,7 @@
 # with CubicWeb.  If not, see <http://www.gnu.org/licenses/>.
 
 from cubicweb.devtools.testlib import CubicWebTC
-from cubicweb.server.session import HOOKS_ALLOW_ALL, HOOKS_DENY_ALL
+from cubicweb.server.session import HOOKS_ALLOW_ALL, HOOKS_DENY_ALL, Connection
 
 class InternalSessionTC(CubicWebTC):
     def test_dbapi_query(self):
@@ -67,6 +67,17 @@
         self.assertEqual(set(), session.disabled_hook_categories)
         self.assertEqual(set(), session.enabled_hook_categories)
 
+    def test_explicite_connection(self):
+        with Connection(self.session) as cnx:
+            rset = cnx.execute('Any X LIMIT 1 WHERE X is CWUser')
+            self.assertEqual(1, len(rset))
+            user = rset.get_entity(0, 0)
+            user.cw_delete()
+            cnx.rollback()
+            new_user = cnx.entity_from_eid(user.eid)
+            self.assertIsNotNone(new_user.login)
+
+
 
 if __name__ == '__main__':
     from logilab.common.testlib import unittest_main