psycopg/psycopg2
1
1
Fork 0
mirror of https://github.com/psycopg/psycopg2.git synced 2025-10-31 07:47:30 +03:00
Code Issues Packages Projects Releases Wiki Activity

Bunch of test tweaks to make the test grid green

Browse Source
This commit is contained in:
Daniele Varrazzo 2016-07-04 21:49:45 +01:00
parent 7c2333dd81
commit 8e28444897
2 changed files with 14 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
tests/test_connection.py
View File

@ -72,16 +72,8 @@ class ConnectionTests(ConnectingTestCase):
# ticket #148 # ticket #148
conn = self.conn conn = self.conn
cur = conn.cursor() cur = conn.cursor()
try: self.assertRaises(psycopg2.OperationalError,
cur.execute("select pg_terminate_backend(pg_backend_pid())") cur.execute, "select pg_terminate_backend(pg_backend_pid())")
except psycopg2.OperationalError, e:
if e.pgcode != psycopg2.errorcodes.ADMIN_SHUTDOWN:
raise
except psycopg2.DatabaseError, e:
# curiously when disconnected in green mode we get a DatabaseError
# without pgcode.
if e.pgcode is not None:
raise
self.assertEqual(conn.closed, 2) self.assertEqual(conn.closed, 2)
conn.close() conn.close()

14
tests/test_cursor.py
View File

@ -27,7 +27,8 @@ import pickle
import psycopg2 import psycopg2
import psycopg2.extensions import psycopg2.extensions
from testutils import (unittest, ConnectingTestCase, skip_before_postgres, from testutils import (unittest, ConnectingTestCase, skip_before_postgres,
skip_if_no_namedtuple, skip_if_no_getrefcount, slow) skip_if_no_namedtuple, skip_if_no_getrefcount, slow, skip_if_no_superuser,
skip_if_windows)
import psycopg2.extras import psycopg2.extras
@ -539,6 +540,9 @@ class CursorTests(ConnectingTestCase):
self.assertRaises(exception, cur.callproc, procname, parameter_sequence) self.assertRaises(exception, cur.callproc, procname, parameter_sequence)
self.conn.rollback() self.conn.rollback()
@skip_if_no_superuser
@skip_if_windows
@skip_before_postgres(8, 4)
def test_external_close_sync(self): def test_external_close_sync(self):
# If a "victim" connection is closed by a "control" connection # If a "victim" connection is closed by a "control" connection
# behind psycopg2's back, psycopg2 always handles it correctly: # behind psycopg2's back, psycopg2 always handles it correctly:
@ -550,6 +554,9 @@ class CursorTests(ConnectingTestCase):
wait_func = lambda conn: None wait_func = lambda conn: None
self._test_external_close(control_conn, connect_func, wait_func) self._test_external_close(control_conn, connect_func, wait_func)
@skip_if_no_superuser
@skip_if_windows
@skip_before_postgres(8, 4)
def test_external_close_async(self): def test_external_close_async(self):
# Issue #443 is in the async code too. Since the fix is duplicated, # Issue #443 is in the async code too. Since the fix is duplicated,
# so is the test. # so is the test.
@ -575,11 +582,14 @@ class CursorTests(ConnectingTestCase):
cur.execute('select pg_terminate_backend(%s)', (pid1,)) cur.execute('select pg_terminate_backend(%s)', (pid1,))
time.sleep(0.001) time.sleep(0.001)
with self.assertRaises(psycopg2.OperationalError):
def f():
with victim_conn.cursor() as cur: with victim_conn.cursor() as cur:
cur.execute('select 1') cur.execute('select 1')
wait_func(victim_conn) wait_func(victim_conn)
self.assertRaises(psycopg2.OperationalError, f)
self.assertEqual(victim_conn.closed, 2) self.assertEqual(victim_conn.closed, 2)