Added test to verify withhold transaction behaviour

A withhold cursor can read its data when the transaction is closed, so
it shouldn't start a new one upon movement/close.

1 files changed, 30 insertions, 4 deletions

diff --git a/tests/test_cursor.py b/tests/test_cursor.py
index 8470ade..2f92b6e 100755
--- a/tests/test_cursor.py
+++ b/tests/test_cursor.py
@@ -176,10 +176,7 @@ class CursorTests(ConnectingTestCase):
         curs.execute("select data from invname order by data")
         self.assertEqual(curs.fetchall(), [(10,), (20,), (30,)])
 
-    def test_withhold(self):
-        self.assertRaises(psycopg2.ProgrammingError, self.conn.cursor,
-                          withhold=True)
-
+    def _create_withhold_table(self):
         curs = self.conn.cursor()
         try:
             curs.execute("drop table withhold")
@@ -190,6 +187,11 @@ class CursorTests(ConnectingTestCase):
             curs.execute("insert into withhold values (%s)", (i,))
         curs.close()
 
+    def test_withhold(self):
+        self.assertRaises(psycopg2.ProgrammingError, self.conn.cursor,
+                          withhold=True)
+
+        self._create_withhold_table()
         curs = self.conn.cursor("W")
         self.assertEqual(curs.withhold, False);
         curs.withhold = True
@@ -209,6 +211,30 @@ class CursorTests(ConnectingTestCase):
         curs.execute("drop table withhold")
         self.conn.commit()
 
+    def test_withhold_no_begin(self):
+        self._create_withhold_table()
+        curs = self.conn.cursor("w", withhold=True)
+        curs.execute("select data from withhold order by data")
+        self.assertEqual(curs.fetchone(), (10,))
+        self.assertEqual(self.conn.status, psycopg2.extensions.STATUS_BEGIN)
+        self.assertEqual(self.conn.get_transaction_status(),
+                         psycopg2.extensions.TRANSACTION_STATUS_INTRANS)
+
+        self.conn.commit()
+        self.assertEqual(self.conn.status, psycopg2.extensions.STATUS_READY)
+        self.assertEqual(self.conn.get_transaction_status(),
+                         psycopg2.extensions.TRANSACTION_STATUS_IDLE)
+
+        self.assertEqual(curs.fetchone(), (20,))
+        self.assertEqual(self.conn.status, psycopg2.extensions.STATUS_READY)
+        self.assertEqual(self.conn.get_transaction_status(),
+                         psycopg2.extensions.TRANSACTION_STATUS_IDLE)
+
+        curs.close()
+        self.assertEqual(self.conn.status, psycopg2.extensions.STATUS_READY)
+        self.assertEqual(self.conn.get_transaction_status(),
+                         psycopg2.extensions.TRANSACTION_STATUS_IDLE)
+
     def test_scrollable(self):
         self.assertRaises(psycopg2.ProgrammingError, self.conn.cursor,
                           scrollable=True)