psycopg/psycopg2
1
1
Fork 0
mirror of https://github.com/psycopg/psycopg2.git synced 2024-11-22 17:06:33 +03:00
Code Issues Packages Projects Releases Wiki Activity

Test suite converted into a proper package.

Browse Source 
Dropped cyclic import from modules to tests: they were only working
because a second copy of the package was found in the project dir.
Use relative import so that 2to3 can do a good conversion.
This commit is contained in:
Daniele Varrazzo 2010-12-21 04:58:38 +00:00
parent f697410ab4
commit ade1b2cc7b
18 changed files with 96 additions and 106 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

32
tests/__init__.py
View File

@ -1,37 +1,9 @@
#!/usr/bin/env python
import os
import sys
from testconfig import dsn
from testutils import unittest
dbname = os.environ.get('PSYCOPG2_TESTDB', 'psycopg2_test')
dbhost = os.environ.get('PSYCOPG2_TESTDB_HOST', None)
dbport = os.environ.get('PSYCOPG2_TESTDB_PORT', None)
dbuser = os.environ.get('PSYCOPG2_TESTDB_USER', None)
# Check if we want to test psycopg's green path.
green = os.environ.get('PSYCOPG2_TEST_GREEN', None)
if green:
if green == '1':
from psycopg2.extras import wait_select as wait_callback
elif green == 'eventlet':
from eventlet.support.psycopg2_patcher import eventlet_wait_callback \
as wait_callback
else:
raise ValueError("please set 'PSYCOPG2_TEST_GREEN' to a valid value")
import psycopg2.extensions
psycopg2.extensions.set_wait_callback(wait_callback)
# Construct a DSN to connect to the test database:
dsn = 'dbname=%s' % dbname
if dbhost is not None:
dsn += ' host=%s' % dbhost
if dbport is not None:
dsn += ' port=%s' % dbport
if dbuser is not None:
dsn += ' user=%s' % dbuser
# If connection to test db fails, bail out early.
import psycopg2
try:
@ -81,4 +53,4 @@ def test_suite():
return suite
if __name__ == '__main__':
unittest.main(defaultTest='test_suite')
unittest.main(defaultTest=test_suite)

8
tests/bug_gc.py
View File

@ -6,18 +6,14 @@ import time
import unittest
import gc
import sys
if sys.version_info < (3,):
import tests
else:
import py3tests as tests
from testconfig import dsn
class StolenReferenceTestCase(unittest.TestCase):
def test_stolen_reference_bug(self):
def fish(val, cur):
gc.collect()
return 42
conn = psycopg2.connect(tests.dsn)
conn = psycopg2.connect(dsn)
UUID = psycopg2.extensions.new_type((2950,), "UUID", fish)
psycopg2.extensions.register_type(UUID, conn)
curs = conn.cursor()

9
tests/extras_dictcursor.py
View File

@ -17,15 +17,14 @@
import psycopg2
import psycopg2.extras
from testutils import unittest
import tests
from testconfig import dsn
class ExtrasDictCursorTests(unittest.TestCase):
"""Test if DictCursor extension class works."""
def setUp(self):
self.conn = psycopg2.connect(tests.dsn)
self.conn = psycopg2.connect(dsn)
curs = self.conn.cursor()
curs.execute("CREATE TEMPORARY TABLE ExtrasDictCursorTests (foo text)")
curs.execute("INSERT INTO ExtrasDictCursorTests VALUES ('bar')")
@ -135,7 +134,7 @@ class NamedTupleCursorTest(unittest.TestCase):
self.conn = None
return
self.conn = psycopg2.connect(tests.dsn,
self.conn = psycopg2.connect(dsn,
connection_factory=NamedTupleConnection)
curs = self.conn.cursor()
curs.execute("CREATE TEMPORARY TABLE nttest (i int, s text)")
@ -207,7 +206,7 @@ class NamedTupleCursorTest(unittest.TestCase):
try:
if self.conn is not None:
self.conn.close()
self.conn = psycopg2.connect(tests.dsn,
self.conn = psycopg2.connect(dsn,
connection_factory=NamedTupleConnection)
curs = self.conn.cursor()
curs.execute("select 1")

13
tests/test_async.py
View File

@ -8,12 +8,7 @@ import time
import select
import StringIO
import sys
if sys.version_info < (3,):
import tests
else:
import py3tests as tests
from testconfig import dsn
class PollableStub(object):
"""A 'pollable' wrapper allowing analysis of the `poll()` calls."""
@ -33,8 +28,8 @@ class PollableStub(object):
class AsyncTests(unittest.TestCase):
def setUp(self):
self.sync_conn = psycopg2.connect(tests.dsn)
self.conn = psycopg2.connect(tests.dsn, async=True)
self.sync_conn = psycopg2.connect(dsn)
self.conn = psycopg2.connect(dsn, async=True)
self.wait(self.conn)
@ -309,7 +304,7 @@ class AsyncTests(unittest.TestCase):
def __init__(self, dsn, async=0):
psycopg2.extensions.connection.__init__(self, dsn, async=async)
conn = psycopg2.connect(tests.dsn, connection_factory=MyConn, async=True)
conn = psycopg2.connect(dsn, connection_factory=MyConn, async=True)
self.assert_(isinstance(conn, MyConn))
self.assert_(conn.async)
conn.close()

10
tests/test_cancel.py
View File

@ -2,18 +2,18 @@
import time
import threading
from testutils import unittest, skip_if_no_pg_sleep
import tests
import psycopg2
import psycopg2.extensions
from psycopg2 import extras
from testconfig import dsn
from testutils import unittest, skip_if_no_pg_sleep
class CancelTests(unittest.TestCase):
def setUp(self):
self.conn = psycopg2.connect(tests.dsn)
self.conn = psycopg2.connect(dsn)
cur = self.conn.cursor()
cur.execute('''
CREATE TEMPORARY TABLE table1 (
@ -65,7 +65,7 @@ class CancelTests(unittest.TestCase):
@skip_if_no_pg_sleep('conn')
def test_async_cancel(self):
async_conn = psycopg2.connect(tests.dsn, async=True)
async_conn = psycopg2.connect(dsn, async=True)
self.assertRaises(psycopg2.OperationalError, async_conn.cancel)
extras.wait_select(async_conn)
cur = async_conn.cursor()
@ -79,7 +79,7 @@ class CancelTests(unittest.TestCase):
self.assertEqual(cur.fetchall(), [(1, )])
def test_async_connection_cancel(self):
async_conn = psycopg2.connect(tests.dsn, async=True)
async_conn = psycopg2.connect(dsn, async=True)
async_conn.close()
self.assertTrue(async_conn.closed)

28
tests/test_connection.py
View File

@ -7,12 +7,12 @@ from operator import attrgetter
import psycopg2
import psycopg2.extensions
import tests
from testconfig import dsn, dbname
class ConnectionTests(unittest.TestCase):
def setUp(self):
self.conn = psycopg2.connect(tests.dsn)
self.conn = psycopg2.connect(dsn)
def tearDown(self):
if not self.conn.closed:
@ -95,7 +95,7 @@ class ConnectionTests(unittest.TestCase):
@skip_if_no_pg_sleep('conn')
def test_concurrent_execution(self):
def slave():
cnn = psycopg2.connect(tests.dsn)
cnn = psycopg2.connect(dsn)
cur = cnn.cursor()
cur.execute("select pg_sleep(2)")
cur.close()
@ -141,7 +141,7 @@ class IsolationLevelsTestCase(unittest.TestCase):
conn.close()
def connect(self):
conn = psycopg2.connect(tests.dsn)
conn = psycopg2.connect(dsn)
self._conns.append(conn)
return conn
@ -333,7 +333,7 @@ class ConnectionTwoPhaseTests(unittest.TestCase):
try:
cur.execute(
"select gid from pg_prepared_xacts where database = %s",
(tests.dbname,))
(dbname,))
except psycopg2.ProgrammingError:
cnn.rollback()
cnn.close()
@ -362,7 +362,7 @@ class ConnectionTwoPhaseTests(unittest.TestCase):
cur.execute("""
select count(*) from pg_prepared_xacts
where database = %s;""",
(tests.dbname,))
(dbname,))
rv = cur.fetchone()[0]
cnn.close()
return rv
@ -377,7 +377,7 @@ class ConnectionTwoPhaseTests(unittest.TestCase):
return rv
def connect(self):
conn = psycopg2.connect(tests.dsn)
conn = psycopg2.connect(dsn)
self._conns.append(conn)
return conn
@ -540,13 +540,13 @@ class ConnectionTwoPhaseTests(unittest.TestCase):
select gid, prepared, owner, database
from pg_prepared_xacts
where database = %s;""",
(tests.dbname,))
(dbname,))
okvals = cur.fetchall()
okvals.sort()
cnn = self.connect()
xids = cnn.tpc_recover()
xids = [ xid for xid in xids if xid.database == tests.dbname ]
xids = [ xid for xid in xids if xid.database == dbname ]
xids.sort(key=attrgetter('gtrid'))
# check the values returned
@ -566,7 +566,7 @@ class ConnectionTwoPhaseTests(unittest.TestCase):
cnn = self.connect()
cur = cnn.cursor()
cur.execute("select gid from pg_prepared_xacts where database = %s;",
(tests.dbname,))
(dbname,))
self.assertEqual('42_Z3RyaWQ=_YnF1YWw=', cur.fetchone()[0])
def test_xid_roundtrip(self):
@ -583,7 +583,7 @@ class ConnectionTwoPhaseTests(unittest.TestCase):
cnn = self.connect()
xids = [ xid for xid in cnn.tpc_recover()
if xid.database == tests.dbname ]
if xid.database == dbname ]
self.assertEqual(1, len(xids))
xid = xids[0]
self.assertEqual(xid.format_id, fid)
@ -605,7 +605,7 @@ class ConnectionTwoPhaseTests(unittest.TestCase):
cnn = self.connect()
xids = [ xid for xid in cnn.tpc_recover()
if xid.database == tests.dbname ]
if xid.database == dbname ]
self.assertEqual(1, len(xids))
xid = xids[0]
self.assertEqual(xid.format_id, None)
@ -651,7 +651,7 @@ class ConnectionTwoPhaseTests(unittest.TestCase):
cnn.tpc_prepare()
cnn.reset()
xid = [ xid for xid in cnn.tpc_recover()
if xid.database == tests.dbname ][0]
if xid.database == dbname ][0]
self.assertEqual(10, xid.format_id)
self.assertEqual('uni', xid.gtrid)
self.assertEqual('code', xid.bqual)
@ -667,7 +667,7 @@ class ConnectionTwoPhaseTests(unittest.TestCase):
cnn.reset()
xid = [ xid for xid in cnn.tpc_recover()
if xid.database == tests.dbname ][0]
if xid.database == dbname ][0]
self.assertEqual(None, xid.format_id)
self.assertEqual('transaction-id', xid.gtrid)
self.assertEqual(None, xid.bqual)

6
tests/test_copy.py
View File

@ -7,11 +7,11 @@ from itertools import cycle, izip
import psycopg2
import psycopg2.extensions
import tests
from testconfig import dsn, green
def skip_if_green(f):
def skip_if_green_(self):
if tests.green:
if green:
return self.skipTest("copy in async mode currently not supported")
else:
return f(self)
@ -42,7 +42,7 @@ class MinimalWrite(object):
class CopyTests(unittest.TestCase):
def setUp(self):
self.conn = psycopg2.connect(tests.dsn)
self.conn = psycopg2.connect(dsn)
curs = self.conn.cursor()
curs.execute('''
CREATE TEMPORARY TABLE tcopy (

4
tests/test_cursor.py
View File

@ -3,12 +3,12 @@
import unittest
import psycopg2
import psycopg2.extensions
import tests
from testconfig import dsn
class CursorTests(unittest.TestCase):
def setUp(self):
self.conn = psycopg2.connect(tests.dsn)
self.conn = psycopg2.connect(dsn)
def tearDown(self):
self.conn.close()

6
tests/test_dates.py
View File

@ -2,9 +2,9 @@
import math
import unittest
import tests
import psycopg2
from psycopg2.tz import FixedOffsetTimezone
from testconfig import dsn
class CommonDatetimeTestsMixin:
@ -75,7 +75,7 @@ class DatetimeTests(unittest.TestCase, CommonDatetimeTestsMixin):
"""Tests for the datetime based date handling in psycopg2."""
def setUp(self):
self.conn = psycopg2.connect(tests.dsn)
self.conn = psycopg2.connect(dsn)
self.curs = self.conn.cursor()
self.DATE = psycopg2._psycopg.PYDATE
self.TIME = psycopg2._psycopg.PYTIME
@ -293,7 +293,7 @@ class mxDateTimeTests(unittest.TestCase, CommonDatetimeTestsMixin):
"""Tests for the mx.DateTime based date handling in psycopg2."""
def setUp(self):
self.conn = psycopg2.connect(tests.dsn)
self.conn = psycopg2.connect(dsn)
self.curs = self.conn.cursor()
self.DATE = psycopg2._psycopg.MXDATE
self.TIME = psycopg2._psycopg.MXTIME

4
tests/test_green.py
View File

@ -4,7 +4,7 @@ import unittest
import psycopg2
import psycopg2.extensions
import psycopg2.extras
import tests
from testconfig import dsn
class ConnectionStub(object):
"""A `connection` wrapper allowing analysis of the `poll()` calls."""
@ -24,7 +24,7 @@ class GreenTests(unittest.TestCase):
def setUp(self):
self._cb = psycopg2.extensions.get_wait_callback()
psycopg2.extensions.set_wait_callback(psycopg2.extras.wait_select)
self.conn = psycopg2.connect(tests.dsn)
self.conn = psycopg2.connect(dsn)
def tearDown(self):
self.conn.close()

8
tests/test_lobject.py
View File

@ -2,11 +2,11 @@
import os
import shutil
import tempfile
from testutils import unittest, decorate_all_tests
import psycopg2
import psycopg2.extensions
import tests
from testconfig import dsn, green
from testutils import unittest, decorate_all_tests
def skip_if_no_lo(f):
def skip_if_no_lo_(self):
@ -19,7 +19,7 @@ def skip_if_no_lo(f):
def skip_if_green(f):
def skip_if_green_(self):
if tests.green:
if green:
return self.skipTest("libpq doesn't support LO in async mode")
else:
return f(self)
@ -30,7 +30,7 @@ def skip_if_green(f):
class LargeObjectMixin(object):
# doesn't derive from TestCase to avoid repeating tests twice.
def setUp(self):
self.conn = psycopg2.connect(tests.dsn)
self.conn = psycopg2.connect(dsn)
self.lo_oid = None
self.tmpdir = None

12
tests/test_notify.py
View File

@ -3,23 +3,19 @@ from testutils import unittest
import psycopg2
from psycopg2 import extensions
from testconfig import dsn
import sys
import time
import select
import signal
from subprocess import Popen, PIPE
import sys
if sys.version_info < (3,):
import tests
else:
import py3tests as tests
class NotifiesTests(unittest.TestCase):
def setUp(self):
self.conn = psycopg2.connect(tests.dsn)
self.conn = psycopg2.connect(dsn)
def tearDown(self):
self.conn.close()
@ -54,7 +50,7 @@ curs.execute("NOTIFY " %(name)r %(payload)r)
curs.close()
conn.close()
"""
% { 'dsn': tests.dsn, 'sec': sec, 'name': name, 'payload': payload})
% { 'dsn': dsn, 'sec': sec, 'name': name, 'payload': payload})
return Popen([sys.executable, '-c', script], stdout=PIPE)

6
tests/test_psycopg2_dbapi20.py
View File

@ -5,12 +5,12 @@ from test_connection import skip_if_tpc_disabled
from testutils import unittest, decorate_all_tests
import psycopg2
import tests
from testconfig import dsn
class Psycopg2Tests(dbapi20.DatabaseAPI20Test):
driver = psycopg2
connect_args = ()
connect_kw_args = {'dsn': tests.dsn}
connect_kw_args = {'dsn': dsn}
lower_func = 'lower' # For stored procedure test
@ -27,7 +27,7 @@ class Psycopg2TPCTests(dbapi20_tpc.TwoPhaseCommitTests, unittest.TestCase):
driver = psycopg2
def connect(self):
return psycopg2.connect(dsn=tests.dsn)
return psycopg2.connect(dsn=dsn)
decorate_all_tests(Psycopg2TPCTests, skip_if_tpc_disabled)

4
tests/test_quote.py
View File

@ -3,7 +3,7 @@ from testutils import unittest
import psycopg2
import psycopg2.extensions
import tests
from testconfig import dsn
class QuotingTestCase(unittest.TestCase):
r"""Checks the correct quoting of strings and binary objects.
@ -24,7 +24,7 @@ class QuotingTestCase(unittest.TestCase):
http://www.postgresql.org/docs/8.1/static/runtime-config-compatible.html
"""
def setUp(self):
self.conn = psycopg2.connect(tests.dsn)
self.conn = psycopg2.connect(dsn)
def tearDown(self):
self.conn.close()

9
tests/test_transaction.py
View File

@ -5,13 +5,12 @@ from testutils import unittest, skip_if_no_pg_sleep
import psycopg2
from psycopg2.extensions import (
ISOLATION_LEVEL_SERIALIZABLE, STATUS_BEGIN, STATUS_READY)
import tests
from testconfig import dsn
class TransactionTests(unittest.TestCase):
def setUp(self):
self.conn = psycopg2.connect(tests.dsn)
self.conn = psycopg2.connect(dsn)
self.conn.set_isolation_level(ISOLATION_LEVEL_SERIALIZABLE)
curs = self.conn.cursor()
curs.execute('''
@ -75,7 +74,7 @@ class DeadlockSerializationTests(unittest.TestCase):
"""Test deadlock and serialization failure errors."""
def connect(self):
conn = psycopg2.connect(tests.dsn)
conn = psycopg2.connect(dsn)
conn.set_isolation_level(ISOLATION_LEVEL_SERIALIZABLE)
return conn
@ -208,7 +207,7 @@ class QueryCancellationTests(unittest.TestCase):
"""Tests for query cancellation."""
def setUp(self):
self.conn = psycopg2.connect(tests.dsn)
self.conn = psycopg2.connect(dsn)
self.conn.set_isolation_level(ISOLATION_LEVEL_SERIALIZABLE)
def tearDown(self):

33
tests/testconfig.py Normal file
View File

@ -0,0 +1,33 @@
# Configure the test suite from the env variables.
import os
dbname = os.environ.get('PSYCOPG2_TESTDB', 'psycopg2_test')
dbhost = os.environ.get('PSYCOPG2_TESTDB_HOST', None)
dbport = os.environ.get('PSYCOPG2_TESTDB_PORT', None)
dbuser = os.environ.get('PSYCOPG2_TESTDB_USER', None)
# Check if we want to test psycopg's green path.
green = os.environ.get('PSYCOPG2_TEST_GREEN', None)
if green:
if green == '1':
from psycopg2.extras import wait_select as wait_callback
elif green == 'eventlet':
from eventlet.support.psycopg2_patcher import eventlet_wait_callback \
as wait_callback
else:
raise ValueError("please set 'PSYCOPG2_TEST_GREEN' to a valid value")
import psycopg2.extensions
psycopg2.extensions.set_wait_callback(wait_callback)
# Construct a DSN to connect to the test database:
dsn = 'dbname=%s' % dbname
if dbhost is not None:
dsn += ' host=%s' % dbhost
if dbport is not None:
dsn += ' port=%s' % dbport
if dbuser is not None:
dsn += ' user=%s' % dbuser

4
tests/types_basic.py
View File

@ -30,14 +30,14 @@ import sys
from testutils import unittest
import psycopg2
import tests
from testconfig import dsn
class TypesBasicTests(unittest.TestCase):
"""Test that all type conversions are working."""
def setUp(self):
self.conn = psycopg2.connect(tests.dsn)
self.conn = psycopg2.connect(dsn)
def tearDown(self):
self.conn.close()

6
tests/types_extras.py
View File

@ -24,7 +24,7 @@ from testutils import unittest
import psycopg2
import psycopg2.extras
import tests
from testconfig import dsn
def skip_if_no_uuid(f):
@ -58,7 +58,7 @@ class TypesExtrasTests(unittest.TestCase):
"""Test that all type conversions are working."""
def setUp(self):
self.conn = psycopg2.connect(tests.dsn)
self.conn = psycopg2.connect(dsn)
def tearDown(self):
self.conn.close()
@ -145,7 +145,7 @@ def skip_if_no_hstore(f):
class HstoreTestCase(unittest.TestCase):
def setUp(self):
self.conn = psycopg2.connect(tests.dsn)
self.conn = psycopg2.connect(dsn)
def tearDown(self):
self.conn.close()