From 283a422b4dda106667b7cf5a6eee91b38b4b718d Mon Sep 17 00:00:00 2001 From: Daniele Varrazzo Date: Thu, 21 Aug 2014 05:01:34 +0100 Subject: [PATCH] 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. --- tests/test_cursor.py | 34 ++++++++++++++++++++++++++++++---- 1 file changed, 30 insertions(+), 4 deletions(-) diff --git a/tests/test_cursor.py b/tests/test_cursor.py index 8470adec..2f92b6e4 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)