psycopg/psycopg2
1
1
Fork 0
mirror of https://github.com/psycopg/psycopg2.git synced 2025-08-01 02:50:08 +03:00
Code Issues Packages Projects Releases Wiki Activity

Always close connection objects in tests

Browse Source
This commit is contained in:
Jon Dufresne 2020-02-01 08:11:09 -08:00
parent e9ae67ff07
commit 6b63fae20a
6 changed files with 33 additions and 20 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
tests/dbapi20.py
View File

@ -232,6 +232,7 @@ class DatabaseAPI20Test(unittest.TestCase):
self.failUnless(con.InternalError is drv.InternalError) self.failUnless(con.InternalError is drv.InternalError)
self.failUnless(con.ProgrammingError is drv.ProgrammingError) self.failUnless(con.ProgrammingError is drv.ProgrammingError)
self.failUnless(con.NotSupportedError is drv.NotSupportedError) self.failUnless(con.NotSupportedError is drv.NotSupportedError)
con.close()
def test_commit(self): def test_commit(self):
@ -251,6 +252,7 @@ class DatabaseAPI20Test(unittest.TestCase):
con.rollback() con.rollback()
except self.driver.NotSupportedError: except self.driver.NotSupportedError:
pass pass
con.close()
def test_cursor(self): def test_cursor(self):
con = self._connect() con = self._connect()

25
tests/dbapi20_tpc.py
View File

@ -24,19 +24,22 @@ class TwoPhaseCommitTests(unittest.TestCase):
def test_xid(self): def test_xid(self):
con = self.connect() con = self.connect()
try: try:
xid = con.xid(42, "global", "bqual") try:
except self.driver.NotSupportedError: xid = con.xid(42, "global", "bqual")
self.fail("Driver does not support transaction IDs.") except self.driver.NotSupportedError:
self.fail("Driver does not support transaction IDs.")
self.assertEquals(xid[0], 42) self.assertEquals(xid[0], 42)
self.assertEquals(xid[1], "global") self.assertEquals(xid[1], "global")
self.assertEquals(xid[2], "bqual") self.assertEquals(xid[2], "bqual")
# Try some extremes for the transaction ID: # Try some extremes for the transaction ID:
xid = con.xid(0, "", "") xid = con.xid(0, "", "")
self.assertEquals(tuple(xid), (0, "", "")) self.assertEquals(tuple(xid), (0, "", ""))
xid = con.xid(0x7fffffff, "a" * 64, "b" * 64) xid = con.xid(0x7fffffff, "a" * 64, "b" * 64)
self.assertEquals(tuple(xid), (0x7fffffff, "a" * 64, "b" * 64)) self.assertEquals(tuple(xid), (0x7fffffff, "a" * 64, "b" * 64))
finally:
con.close()
def test_tpc_begin(self): def test_tpc_begin(self):
con = self.connect() con = self.connect()

4
tests/test_async.py
View File

@ -451,14 +451,16 @@ class AsyncTests(ConnectingTestCase):
self.assertEqual(cur.fetchone(), (42,)) self.assertEqual(cur.fetchone(), (42,))
def test_async_connection_error_message(self): def test_async_connection_error_message(self):
cnn = psycopg2.connect('dbname=thisdatabasedoesntexist', async_=True)
try: try:
cnn = psycopg2.connect('dbname=thisdatabasedoesntexist', async_=True)
self.wait(cnn) self.wait(cnn)
except psycopg2.Error as e: except psycopg2.Error as e:
self.assertNotEqual(str(e), "asynchronous connection failed", self.assertNotEqual(str(e), "asynchronous connection failed",
"connection error reason lost") "connection error reason lost")
else: else:
self.fail("no exception raised") self.fail("no exception raised")
finally:
cnn.close()
@skip_before_postgres(8, 2) @skip_before_postgres(8, 2)
def test_copy_no_hang(self): def test_copy_no_hang(self):

3
tests/test_async_keyword.py
View File

@ -89,6 +89,8 @@ class AsyncTests(ConnectingTestCase):
"connection error reason lost") "connection error reason lost")
else: else:
self.fail("no exception raised") self.fail("no exception raised")
finally:
cnn.close()
class CancelTests(ConnectingTestCase): class CancelTests(ConnectingTestCase):
@ -118,6 +120,7 @@ class CancelTests(ConnectingTestCase):
cur.execute("select 1") cur.execute("select 1")
extras.wait_select(async_conn) extras.wait_select(async_conn)
self.assertEqual(cur.fetchall(), [(1, )]) self.assertEqual(cur.fetchall(), [(1, )])
async_conn.close()
def test_async_connection_cancel(self): def test_async_connection_cancel(self):
async_conn = psycopg2.connect(dsn, async=True) async_conn = psycopg2.connect(dsn, async=True)

1
tests/test_cancel.py
View File

@ -105,6 +105,7 @@ class CancelTests(ConnectingTestCase):
cur.execute("select 1") cur.execute("select 1")
extras.wait_select(async_conn) extras.wait_select(async_conn)
self.assertEqual(cur.fetchall(), [(1, )]) self.assertEqual(cur.fetchall(), [(1, )])
async_conn.close()
def test_async_connection_cancel(self): def test_async_connection_cancel(self):
async_conn = psycopg2.connect(dsn, async_=True) async_conn = psycopg2.connect(dsn, async_=True)

18
tests/testutils.py
View File

@ -248,15 +248,17 @@ def skip_if_tpc_disabled(f):
@wraps(f) @wraps(f)
def skip_if_tpc_disabled_(self): def skip_if_tpc_disabled_(self):
cnn = self.connect() cnn = self.connect()
cur = cnn.cursor()
try: try:
cur.execute("SHOW max_prepared_transactions;") cur = cnn.cursor()
except psycopg2.ProgrammingError: try:
return self.skipTest( cur.execute("SHOW max_prepared_transactions;")
"server too old: two phase transactions not supported.") except psycopg2.ProgrammingError:
else: return self.skipTest(
mtp = int(cur.fetchone()[0]) "server too old: two phase transactions not supported.")
cnn.close() else:
mtp = int(cur.fetchone()[0])
finally:
cnn.close()
if not mtp: if not mtp:
return self.skipTest( return self.skipTest(