Fixed tpc_recover() with RealDictStuff

Same problem and correction of ticket #114.

NEWS

@@ -7,8 +7,8 @@ What's new in psycopg 2.4.6
   - Dropped GIL release during string adaptation around a function call
     invoking a Python API function, which could cause interpreter crash.
     Thanks to Manu Cupcic for the report (ticket #110).
-  - 'register_hstore()', 'register_composite()' work with RealDictConnection
+  - 'register_hstore()', 'register_composite()', 'tpc_recover()' work with
-    and Cursor. Thanks to Adrian Klaver for the report (ticket #114).
+    RealDictConnection and Cursor (ticket #114).
   - connection.reset() implemented using DISCARD ALL on server versions
     supporting it.
 

psycopg/xid_type.c

@@ -28,6 +28,7 @@
 #include "psycopg/psycopg.h"
 
 #include "psycopg/xid.h"
+#include "psycopg/cursor.h"
 
 
 static const char xid_doc[] =
@@ -660,8 +661,11 @@ xid_recover(PyObject *conn)
     PyObject *tmp;
     Py_ssize_t len, i;
 
-    /* curs = conn.cursor() */
+    /* curs = conn.cursor()
-    if (!(curs = PyObject_CallMethod(conn, "cursor", NULL))) { goto exit; }
+     * (sort of. Use the real cursor in case the connection returns
+     * somenthing non-dbapi -- see ticket #114) */
+    if (!(curs = PyObject_CallFunctionObjArgs(
+        (PyObject *)&cursorType, conn, NULL))) { goto exit; }
 
     /* curs.execute(...) */
     if (!(tmp = PyObject_CallMethod(curs, "execute", "s",

tests/test_connection.py

@@ -458,8 +458,8 @@ class ConnectionTwoPhaseTests(unittest.TestCase):
         cnn.close()
         return rv
 
-    def connect(self):
+    def connect(self, **kwargs):
-        conn = psycopg2.connect(dsn)
+        conn = psycopg2.connect(dsn, **kwargs)
         self._conns.append(conn)
         return conn
 
@@ -760,6 +760,20 @@ class ConnectionTwoPhaseTests(unittest.TestCase):
         cnn.tpc_prepare()
         self.assertRaises(psycopg2.ProgrammingError, cnn.cancel)
 
+    def test_tpc_recover_non_dbapi_connection(self):
+        from psycopg2.extras import RealDictConnection
+        cnn = self.connect(connection_factory=RealDictConnection)
+        cnn.tpc_begin('dict-connection')
+        cnn.tpc_prepare()
+        cnn.reset()
+
+        xids = cnn.tpc_recover()
+        xid = [ xid for xid in xids if xid.database == dbname ][0]
+        self.assertEqual(None, xid.format_id)
+        self.assertEqual('dict-connection', xid.gtrid)
+        self.assertEqual(None, xid.bqual)
+
+
 from testutils import skip_if_tpc_disabled
 decorate_all_tests(ConnectionTwoPhaseTests, skip_if_tpc_disabled)