psycopg/psycopg2
1
1
Fork 0
mirror of https://github.com/psycopg/psycopg2.git synced 2025-01-31 17:34:08 +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 23:01:49 +01:00
parent 98a9203827
commit a5970a21a8
3 changed files with 26 additions and 23 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
tests/test_connection.py
View File

@ -71,16 +71,8 @@ class ConnectionTests(ConnectingTestCase):
# ticket #148
conn = self.conn
cur = conn.cursor()
try:
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.assertRaises(psycopg2.OperationalError,
cur.execute, "select pg_terminate_backend(pg_backend_pid())")
self.assertEqual(conn.closed, 2)
conn.close()

28
tests/test_cursor.py
View File

@ -32,6 +32,8 @@ import psycopg2.extras
from psycopg2.extensions import b
from testutils import unittest, ConnectingTestCase, skip_before_postgres
from testutils import skip_if_no_namedtuple, skip_if_no_getrefcount
from testutils import skip_if_no_superuser, skip_if_windows
class CursorTests(ConnectingTestCase):
@ -51,8 +53,10 @@ class CursorTests(ConnectingTestCase):
conn = self.conn
cur = conn.cursor()
cur.execute("create temp table test_exc (data int);")
def buggygen():
yield 1//0
yield 1 // 0
self.assertRaises(ZeroDivisionError,
cur.executemany, "insert into test_exc values (%s)", buggygen())
cur.close()
@ -197,16 +201,16 @@ class CursorTests(ConnectingTestCase):
self._create_withhold_table()
curs = self.conn.cursor("W")
self.assertEqual(curs.withhold, False);
self.assertEqual(curs.withhold, False)
curs.withhold = True
self.assertEqual(curs.withhold, True);
self.assertEqual(curs.withhold, True)
curs.execute("select data from withhold order by data")
self.conn.commit()
self.assertEqual(curs.fetchall(), [(10,), (20,), (30,)])
curs.close()
curs = self.conn.cursor("W", withhold=True)
self.assertEqual(curs.withhold, True);
self.assertEqual(curs.withhold, True)
curs.execute("select data from withhold order by data")
self.conn.commit()
self.assertEqual(curs.fetchall(), [(10,), (20,), (30,)])
@ -268,18 +272,18 @@ class CursorTests(ConnectingTestCase):
curs = self.conn.cursor()
curs.execute("create table scrollable (data int)")
curs.executemany("insert into scrollable values (%s)",
[ (i,) for i in range(100) ])
[(i,) for i in range(100)])
curs.close()
for t in range(2):
if not t:
curs = self.conn.cursor("S")
self.assertEqual(curs.scrollable, None);
self.assertEqual(curs.scrollable, None)
curs.scrollable = True
else:
curs = self.conn.cursor("S", scrollable=True)
self.assertEqual(curs.scrollable, True);
self.assertEqual(curs.scrollable, True)
curs.itersize = 10
# complex enough to make postgres cursors declare without
@ -307,7 +311,7 @@ class CursorTests(ConnectingTestCase):
curs = self.conn.cursor()
curs.execute("create table scrollable (data int)")
curs.executemany("insert into scrollable values (%s)",
[ (i,) for i in range(100) ])
[(i,) for i in range(100)])
curs.close()
curs = self.conn.cursor("S") # default scrollability
@ -344,7 +348,7 @@ class CursorTests(ConnectingTestCase):
def test_iter_named_cursor_default_itersize(self):
curs = self.conn.cursor('tmp')
curs.execute('select generate_series(1,50)')
rv = [ (r[0], curs.rownumber) for r in curs ]
rv = [(r[0], curs.rownumber) for r in curs]
# everything swallowed in one gulp
self.assertEqual(rv, [(i,i) for i in range(1,51)])
@ -353,7 +357,7 @@ class CursorTests(ConnectingTestCase):
curs = self.conn.cursor('tmp')
curs.itersize = 30
curs.execute('select generate_series(1,50)')
rv = [ (r[0], curs.rownumber) for r in curs ]
rv = [(r[0], curs.rownumber) for r in curs]
# everything swallowed in two gulps
self.assertEqual(rv, [(i,((i - 1) % 30) + 1) for i in range(1,51)])
@ -493,6 +497,8 @@ class CursorTests(ConnectingTestCase):
cur = self.conn.cursor()
self.assertRaises(TypeError, cur.callproc, 'lower', 42)
@skip_if_no_superuser
@skip_if_windows
@skip_before_postgres(8, 4)
def test_external_close_sync(self):
# If a "victim" connection is closed by a "control" connection
@ -505,6 +511,8 @@ class CursorTests(ConnectingTestCase):
wait_func = lambda conn: None
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):
# Issue #443 is in the async code too. Since the fix is duplicated,

9
tests/test_quote.py
View File

@ -193,7 +193,8 @@ class TestQuotedString(ConnectingTestCase):
a = adapt(snowman)
a.encoding = 'utf8'
self.assertEqual(a.encoding, 'utf8')
self.assertEqual(a.getquoted(), b("'\xe2\x98\x83'"))
q = a.getquoted()
self.assert_(q in (b("'\xe2\x98\x83'"), b("E'\xe2\x98\x83'")), q)
def test_connection_wins_anyway(self):
from psycopg2.extensions import adapt
@ -205,7 +206,8 @@ class TestQuotedString(ConnectingTestCase):
a.prepare(self.conn)
self.assertEqual(a.encoding, 'utf_8')
self.assertEqual(a.getquoted(), b("'\xe2\x98\x83'"))
q = a.getquoted()
self.assert_(q in (b("'\xe2\x98\x83'"), b("E'\xe2\x98\x83'")), q)
@testutils.skip_before_python(3)
def test_adapt_bytes(self):
@ -213,7 +215,8 @@ class TestQuotedString(ConnectingTestCase):
self.conn.set_client_encoding('utf8')
a = psycopg2.extensions.QuotedString(snowman.encode('utf8'))
a.prepare(self.conn)
self.assertEqual(a.getquoted(), b("'\xe2\x98\x83'"))
q = a.getquoted()
self.assert_(q in (b("'\xe2\x98\x83'"), b("E'\xe2\x98\x83'")), q)
def test_suite():