# HG changeset patch # User Pierre-Yves David # Date 1372249566 -7200 # Node ID 499db4fd03f827b576e146d5f58cd35a31c0af4e # Parent 6a4070e2849da12a1abc0eb73cd67e3d06d9c812 [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. diff -r 6a4070e2849d -r 499db4fd03f8 server/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 diff -r 6a4070e2849d -r 499db4fd03f8 server/test/unittest_session.py --- 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 . 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