Check for connection closed before getting the isolation level

Closes ticket #74

Also added test to check regressions in isolation_level,
set_isolation_level, set_session, autocommit.
This commit is contained in:
Daniele Varrazzo 2011-11-16 23:51:05 +00:00
parent a76d0525d1
commit dc94a3cb2d
2 changed files with 35 additions and 1 deletions

View File

@ -576,7 +576,12 @@ psyco_conn_autocommit_set(connectionObject *self, PyObject *pyvalue)
static PyObject *
psyco_conn_isolation_level_get(connectionObject *self)
{
int rv = conn_get_isolation_level(self);
int rv;
EXC_IF_CONN_CLOSED(self);
EXC_IF_TPC_PREPARED(self, set_isolation_level);
rv = conn_get_isolation_level(self);
if (-1 == rv) { return NULL; }
return PyInt_FromLong((long)rv);
}

View File

@ -344,6 +344,16 @@ class IsolationLevelsTestCase(unittest.TestCase):
cur2.execute("select count(*) from isolevel;")
self.assertEqual(2, cur2.fetchone()[0])
def test_isolation_level_closed(self):
cnn = self.connect()
cnn.close()
self.assertRaises(psycopg2.InterfaceError, getattr,
cnn, 'isolation_level')
self.assertRaises(psycopg2.InterfaceError,
cnn.set_isolation_level, 0)
self.assertRaises(psycopg2.InterfaceError,
cnn.set_isolation_level, 1)
class ConnectionTwoPhaseTests(unittest.TestCase):
def setUp(self):
@ -725,6 +735,12 @@ class TransactionControlTests(unittest.TestCase):
if not self.conn.closed:
self.conn.close()
def test_closed(self):
self.conn.close()
self.assertRaises(psycopg2.InterfaceError,
self.conn.set_session,
psycopg2.extensions.ISOLATION_LEVEL_SERIALIZABLE)
def test_not_in_transaction(self):
cur = self.conn.cursor()
cur.execute("select 1")
@ -868,6 +884,19 @@ class AutocommitTests(unittest.TestCase):
if not self.conn.closed:
self.conn.close()
def test_closed(self):
self.conn.close()
self.assertRaises(psycopg2.InterfaceError,
setattr, self.conn, 'autocommit', True)
# The getter doesn't have a guard. We may change this in future
# to make it consistent with other methods; meanwhile let's just check
# it doesn't explode.
try:
self.assert_(self.conn.autocommit in (True, False))
except psycopg2.InterfaceError:
pass
def test_default_no_autocommit(self):
self.assert_(not self.conn.autocommit)
self.assertEqual(self.conn.status, psycopg2.extensions.STATUS_READY)