psycopg/psycopg2
1
1
Fork 0
mirror of https://github.com/psycopg/psycopg2.git synced 2024-09-21 03:08:57 +03:00
Code Issues Packages Projects Releases Wiki Activity

Added reason for skipping on CockroachDB

Browse Source
This commit is contained in:
Daniele Varrazzo 2020-07-27 22:58:43 +01:00
parent 5e957daa82
commit 6eb4fab1db
19 changed files with 97 additions and 89 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
tests/test_async.py
View File

@ -328,6 +328,7 @@ class AsyncTests(ConnectingTestCase):
conn.close()
@slow
@skip_if_crdb("flush on write flakey")
def test_flush_on_write(self):
# a very large query requires a flush loop to be sent to the backend
curs = self.conn.cursor()

7
tests/test_cancel.py
View File

@ -34,16 +34,15 @@ from psycopg2 import extras
from .testconfig import dsn
import unittest
from .testutils import ConnectingTestCase, skip_before_postgres, slow
from .testutils import crdb_version
from .testutils import skip_if_crdb
class CancelTests(ConnectingTestCase):
def setUp(self):
ConnectingTestCase.setUp(self)
# here, instead of a decorator, to avoid creating the temp table
if crdb_version(self.conn) is not None:
self.skipTest("cancel not supported on CockroachDB")
skip_if_crdb("cancel", self.conn)
cur = self.conn.cursor()
cur.execute('''

36
tests/test_connection.py
View File

@ -75,7 +75,7 @@ class ConnectionTests(ConnectingTestCase):
conn.close()
self.assertEqual(curs.closed, True)
@skip_if_crdb
@skip_if_crdb("backend pid")
@skip_before_postgres(8, 4)
@skip_if_no_superuser
@skip_if_windows
@ -90,7 +90,7 @@ class ConnectionTests(ConnectingTestCase):
conn.close()
self.assertEqual(conn.closed, 1)
@skip_if_crdb
@skip_if_crdb("isolation level")
def test_reset(self):
conn = self.conn
# switch session characteristics
@ -114,7 +114,7 @@ class ConnectionTests(ConnectingTestCase):
if self.conn.info.server_version >= 90100:
self.assert_(conn.deferrable is None)
@skip_if_crdb
@skip_if_crdb("notice")
def test_notices(self):
conn = self.conn
cur = conn.cursor()
@ -124,7 +124,7 @@ class ConnectionTests(ConnectingTestCase):
self.assertEqual("CREATE TABLE", cur.statusmessage)
self.assert_(conn.notices)
@skip_if_crdb
@skip_if_crdb("notice")
def test_notices_consistent_order(self):
conn = self.conn
cur = conn.cursor()
@ -145,7 +145,7 @@ class ConnectionTests(ConnectingTestCase):
self.assert_('table4' in conn.notices[3])
@slow
@skip_if_crdb
@skip_if_crdb("notice")
def test_notices_limited(self):
conn = self.conn
cur = conn.cursor()
@ -160,7 +160,7 @@ class ConnectionTests(ConnectingTestCase):
self.assert_('table99' in conn.notices[-1], conn.notices[-1])
@slow
@skip_if_crdb
@skip_if_crdb("notice")
def test_notices_deque(self):
conn = self.conn
self.conn.notices = deque()
@ -191,7 +191,7 @@ class ConnectionTests(ConnectingTestCase):
self.assertEqual(len([n for n in conn.notices if 'CREATE TABLE' in n]),
100)
@skip_if_crdb
@skip_if_crdb("notice")
def test_notices_noappend(self):
conn = self.conn
self.conn.notices = None # will make an error swallowes ok
@ -238,7 +238,7 @@ class ConnectionTests(ConnectingTestCase):
self.assert_(time.time() - t0 < 7,
"something broken in concurrency")
@skip_if_crdb
@skip_if_crdb("encoding")
def test_encoding_name(self):
self.conn.set_client_encoding("EUC_JP")
# conn.encoding is 'EUCJP' now.
@ -338,7 +338,7 @@ class ConnectionTests(ConnectingTestCase):
cur = conn.cursor(cursor_factory=None)
self.assertEqual(type(cur), psycopg2.extras.DictCursor)
@skip_if_crdb
@skip_if_crdb("connect any db")
def test_failed_init_status(self):
class SubConnection(ext.connection):
def __init__(self, dsn):
@ -599,7 +599,7 @@ class IsolationLevelsTestCase(ConnectingTestCase):
conn = self.connect()
self.assert_(conn.encoding in ext.encodings)
@skip_if_crdb
@skip_if_crdb("isolation level")
def test_set_isolation_level(self):
conn = self.connect()
curs = conn.cursor()
@ -647,7 +647,7 @@ class IsolationLevelsTestCase(ConnectingTestCase):
curs.execute('show transaction_isolation;')
self.assertEqual(curs.fetchone()[0], 'serializable')
@skip_if_crdb
@skip_if_crdb("isolation level")
def test_set_isolation_level_default(self):
conn = self.connect()
curs = conn.cursor()
@ -722,7 +722,7 @@ class IsolationLevelsTestCase(ConnectingTestCase):
cur1.execute("select count(*) from isolevel;")
self.assertEqual(1, cur1.fetchone()[0])
@skip_if_crdb
@skip_if_crdb("isolation level")
def test_isolation_level_read_committed(self):
cnn1 = self.connect()
cnn2 = self.connect()
@ -749,7 +749,7 @@ class IsolationLevelsTestCase(ConnectingTestCase):
cur1.execute("select count(*) from isolevel;")
self.assertEqual(2, cur1.fetchone()[0])
@skip_if_crdb
@skip_if_crdb("isolation level")
def test_isolation_level_serializable(self):
cnn1 = self.connect()
cnn2 = self.connect()
@ -787,7 +787,7 @@ class IsolationLevelsTestCase(ConnectingTestCase):
self.assertRaises(psycopg2.InterfaceError,
cnn.set_isolation_level, 1)
@skip_if_crdb
@skip_if_crdb("isolation level")
def test_setattr_isolation_level_int(self):
cur = self.conn.cursor()
self.conn.isolation_level = ext.ISOLATION_LEVEL_SERIALIZABLE
@ -836,7 +836,7 @@ class IsolationLevelsTestCase(ConnectingTestCase):
cur.execute("SHOW default_transaction_isolation;")
self.assertEqual(cur.fetchone()[0], isol)
@skip_if_crdb
@skip_if_crdb("isolation level")
def test_setattr_isolation_level_str(self):
cur = self.conn.cursor()
self.conn.isolation_level = "serializable"
@ -1273,7 +1273,7 @@ class ConnectionTwoPhaseTests(ConnectingTestCase):
self.assertEqual(None, xid.bqual)
@skip_if_crdb
@skip_if_crdb("isolation level")
class TransactionControlTests(ConnectingTestCase):
def test_closed(self):
self.conn.close()
@ -1702,7 +1702,7 @@ class PasswordLeakTestCase(ConnectingTestCase):
# the password away
PasswordLeakTestCase.dsn = self.dsn
@skip_if_crdb
@skip_if_crdb("connect any db")
def test_leak(self):
self.assertRaises(psycopg2.DatabaseError,
self.GrassingConnection, "dbname=nosuch password=whateva")
@ -1888,7 +1888,7 @@ class TestConnectionInfo(ConnectingTestCase):
self.assert_(self.conn.info.socket >= 0)
self.assert_(self.bconn.info.socket < 0)
@skip_if_crdb
@skip_if_crdb("backend pid")
def test_backend_pid(self):
cur = self.conn.cursor()
try:

6
tests/test_copy.py
View File

@ -28,7 +28,7 @@ import sys
import string
import unittest
from .testutils import ConnectingTestCase, skip_before_postgres, slow, StringIO
from .testutils import crdb_version, skip_if_crdb
from .testutils import skip_if_crdb
from itertools import cycle
from subprocess import Popen, PIPE
@ -59,7 +59,6 @@ class MinimalWrite(TextIOBase):
return self.f.write(data)
@skip_if_crdb
@skip_copy_if_green
class CopyTests(ConnectingTestCase):
@ -68,9 +67,8 @@ class CopyTests(ConnectingTestCase):
self._create_temp_table()
def _create_temp_table(self):
skip_if_crdb("copy", self.conn)
curs = self.conn.cursor()
if crdb_version(self.conn) is not None:
curs.execute("SET experimental_enable_temp_tables = 'on'")
curs.execute('''
CREATE TEMPORARY TABLE tcopy (
id serial PRIMARY KEY,

10
tests/test_cursor.py
View File

@ -213,7 +213,7 @@ class CursorTests(ConnectingTestCase):
self.assertEqual(c.precision, None)
self.assertEqual(c.scale, None)
@skip_if_crdb
@skip_if_crdb("table oid")
def test_description_extra_attribs(self):
curs = self.conn.cursor()
curs.execute("""
@ -271,7 +271,7 @@ class CursorTests(ConnectingTestCase):
# It would be inappropriate to test callproc's named parameters in the
# DBAPI2.0 test section because they are a psycopg2 extension.
@skip_before_postgres(9, 0)
@skip_if_crdb
@skip_if_crdb("stored procedure")
def test_callproc_dict(self):
# This parameter name tests for injection and quote escaping
paramname = '''
@ -309,7 +309,7 @@ class CursorTests(ConnectingTestCase):
@skip_if_no_superuser
@skip_if_windows
@skip_if_crdb
@skip_if_crdb("backend pid")
@skip_before_postgres(8, 4)
def test_external_close_sync(self):
# If a "victim" connection is closed by a "control" connection
@ -327,7 +327,7 @@ class CursorTests(ConnectingTestCase):
@skip_if_no_superuser
@skip_if_windows
@skip_if_crdb
@skip_if_crdb("backend pid")
@skip_before_postgres(8, 4)
def test_external_close_async(self):
# Issue #443 is in the async code too. Since the fix is duplicated,
@ -403,7 +403,7 @@ class CursorTests(ConnectingTestCase):
self.assert_(curs.pgresult_ptr is None)
@skip_if_crdb
@skip_if_crdb("named cursor")
class NamedCursorTests(ConnectingTestCase):
def test_invalid_name(self):
curs = self.conn.cursor()

6
tests/test_dates.py
View File

@ -246,7 +246,7 @@ class DatetimeTests(ConnectingTestCase, CommonDatetimeTestsMixin):
[time(13, 30, 29)])
self.assertEqual(value, '13:30:29')
@skip_if_crdb
@skip_if_crdb("cast adds tz")
def test_adapt_datetime(self):
value = self.execute('select (%s)::timestamp::text',
[datetime(2007, 1, 1, 13, 30, 29)])
@ -387,7 +387,7 @@ class DatetimeTests(ConnectingTestCase, CommonDatetimeTestsMixin):
self.assertRaises(OverflowError, f, '00:00:100000000000000000:00')
self.assertRaises(OverflowError, f, '00:00:00.100000000000000000')
@skip_if_crdb
@skip_if_crdb("infinity date")
def test_adapt_infinity_tz(self):
t = self.execute("select 'infinity'::timestamp")
self.assert_(t.tzinfo is None)
@ -425,7 +425,7 @@ class DatetimeTests(ConnectingTestCase, CommonDatetimeTestsMixin):
r = cur.fetchone()[0]
self.assertEqual(r, v, "%s -> %s != %s" % (s, r, v))
@skip_if_crdb
@skip_if_crdb("interval style")
@skip_before_postgres(8, 4)
def test_interval_iso_8601_not_supported(self):
# We may end up supporting, but no pressure for it

28
tests/test_extras_dictcursor.py
View File

@ -64,7 +64,7 @@ class _DictCursorBase(ConnectingTestCase):
class ExtrasDictCursorTests(_DictCursorBase):
"""Test if DictCursor extension class works."""
@skip_if_crdb
@skip_if_crdb("named cursor")
def testDictConnCursorArgs(self):
self.conn.close()
self.conn = self.connect(connection_factory=psycopg2.extras.DictConnection)
@ -132,19 +132,19 @@ class ExtrasDictCursorTests(_DictCursorBase):
return row
self._testWithNamedCursor(getter)
@skip_if_crdb
@skip_if_crdb("named cursor")
@skip_before_postgres(8, 2)
def testDictCursorWithNamedCursorNotGreedy(self):
curs = self.conn.cursor('tmp', cursor_factory=psycopg2.extras.DictCursor)
self._testNamedCursorNotGreedy(curs)
@skip_if_crdb
@skip_if_crdb("named cursor")
@skip_before_postgres(8, 0)
def testDictCursorWithNamedCursorIterRowNumber(self):
curs = self.conn.cursor('tmp', cursor_factory=psycopg2.extras.DictCursor)
self._testIterRowNumber(curs)
@skip_if_crdb
@skip_if_crdb("named cursor")
def _testWithNamedCursor(self, getter):
curs = self.conn.cursor('aname', cursor_factory=psycopg2.extras.DictCursor)
curs.execute("SELECT * FROM ExtrasDictCursorTests")
@ -320,19 +320,19 @@ class ExtrasDictCursorRealTests(_DictCursorBase):
return row
self._testWithNamedCursorReal(getter)
@skip_if_crdb
@skip_if_crdb("named cursor")
@skip_before_postgres(8, 2)
def testDictCursorRealWithNamedCursorNotGreedy(self):
curs = self.conn.cursor('tmp', cursor_factory=psycopg2.extras.RealDictCursor)
self._testNamedCursorNotGreedy(curs)
@skip_if_crdb
@skip_if_crdb("named cursor")
@skip_before_postgres(8, 0)
def testDictCursorRealWithNamedCursorIterRowNumber(self):
curs = self.conn.cursor('tmp', cursor_factory=psycopg2.extras.RealDictCursor)
self._testIterRowNumber(curs)
@skip_if_crdb
@skip_if_crdb("named cursor")
def _testWithNamedCursorReal(self, getter):
curs = self.conn.cursor('aname',
cursor_factory=psycopg2.extras.RealDictCursor)
@ -446,7 +446,7 @@ class NamedTupleCursorTest(ConnectingTestCase):
curs.execute("INSERT INTO nttest VALUES (3, 'baz')")
self.conn.commit()
@skip_if_crdb
@skip_if_crdb("named cursor")
def test_cursor_args(self):
cur = self.conn.cursor('foo', cursor_factory=psycopg2.extras.DictCursor)
self.assertEqual(cur.name, 'foo')
@ -604,7 +604,7 @@ class NamedTupleCursorTest(ConnectingTestCase):
finally:
NamedTupleCursor._make_nt = f_orig
@skip_if_crdb
@skip_if_crdb("named cursor")
@skip_before_postgres(8, 0)
def test_named(self):
curs = self.conn.cursor('tmp')
@ -615,28 +615,28 @@ class NamedTupleCursorTest(ConnectingTestCase):
recs.extend(curs.fetchall())
self.assertEqual(list(range(10)), [t.i for t in recs])
@skip_if_crdb
@skip_if_crdb("named cursor")
def test_named_fetchone(self):
curs = self.conn.cursor('tmp')
curs.execute("""select 42 as i""")
t = curs.fetchone()
self.assertEqual(t.i, 42)
@skip_if_crdb
@skip_if_crdb("named cursor")
def test_named_fetchmany(self):
curs = self.conn.cursor('tmp')
curs.execute("""select 42 as i""")
recs = curs.fetchmany(10)
self.assertEqual(recs[0].i, 42)
@skip_if_crdb
@skip_if_crdb("named cursor")
def test_named_fetchall(self):
curs = self.conn.cursor('tmp')
curs.execute("""select 42 as i""")
recs = curs.fetchall()
self.assertEqual(recs[0].i, 42)
@skip_if_crdb
@skip_if_crdb("named cursor")
@skip_before_postgres(8, 2)
def test_not_greedy(self):
curs = self.conn.cursor('tmp')
@ -651,7 +651,7 @@ class NamedTupleCursorTest(ConnectingTestCase):
self.assert_(recs[1].ts - recs[0].ts < timedelta(seconds=0.005))
self.assert_(recs[2].ts - recs[1].ts > timedelta(seconds=0.0099))
@skip_if_crdb
@skip_if_crdb("named cursor")
@skip_before_postgres(8, 0)
def test_named_rownumber(self):
curs = self.conn.cursor('tmp')

7
tests/test_green.py
View File

@ -68,6 +68,7 @@ class GreenTestCase(ConnectingTestCase):
return stub
@slow
@skip_if_crdb("flush on write flakey")
def test_flush_on_write(self):
# a very large query requires a flush loop to be sent to the backend
conn = self.conn
@ -123,9 +124,9 @@ class GreenTestCase(ConnectingTestCase):
cur.execute, "copy (select 1) to stdout")
@slow
@skip_if_crdb
@skip_if_crdb("notice")
@skip_before_postgres(9, 0)
def test_non_block_after_notification(self):
def test_non_block_after_notice(self):
def wait(conn):
while 1:
state = conn.poll()
@ -218,7 +219,7 @@ class CallbackErrorTestCase(ConnectingTestCase):
self.fail("you should have had a success or an error by now")
@skip_if_crdb
@skip_if_crdb("named cursor")
def test_errors_named_cursor(self):
for i in range(100):
self.to_error = None

4
tests/test_ipaddress.py
View File

@ -71,7 +71,7 @@ class NetworkingTestCase(testutils.ConnectingTestCase):
cur.execute("select %s", [ip.ip_interface('::ffff:102:300/128')])
self.assertEquals(cur.fetchone()[0], '::ffff:102:300/128')
@testutils.skip_if_crdb
@testutils.skip_if_crdb("cidr")
def test_cidr_cast(self):
cur = self.conn.cursor()
psycopg2.extras.register_ipaddress(cur)
@ -89,7 +89,7 @@ class NetworkingTestCase(testutils.ConnectingTestCase):
self.assert_(isinstance(obj, ip.IPv6Network), repr(obj))
self.assertEquals(obj, ip.ip_network('::ffff:102:300/128'))
@testutils.skip_if_crdb
@testutils.skip_if_crdb("cidr")
@testutils.skip_before_postgres(8, 2)
def test_cidr_array_cast(self):
cur = self.conn.cursor()

2
tests/test_lobject.py
View File

@ -38,7 +38,7 @@ from .testutils import (decorate_all_tests, skip_if_tpc_disabled,
def skip_if_no_lo(f):
f = skip_before_postgres(8, 1, "large objects only supported from PG 8.1")(f)
f = skip_if_green("libpq doesn't support LO in async mode")(f)
f = skip_if_crdb(f)
f = skip_if_crdb("large objects")(f)
return f

8
tests/test_module.py
View File

@ -216,7 +216,7 @@ class ExceptionsTestCase(ConnectingTestCase):
gc.collect()
assert(w() is None)
@skip_if_crdb
@skip_if_crdb("copy")
@skip_copy_if_green
def test_diagnostics_copy(self):
f = StringIO()
@ -245,7 +245,7 @@ class ExceptionsTestCase(ConnectingTestCase):
self.assertEqual(diag1.sqlstate, '42601')
self.assertEqual(diag2.sqlstate, '42P01')
@skip_if_crdb
@skip_if_crdb("deferrable")
def test_diagnostics_from_commit(self):
cur = self.conn.cursor()
cur.execute("""
@ -261,7 +261,7 @@ class ExceptionsTestCase(ConnectingTestCase):
e = exc
self.assertEqual(e.diag.sqlstate, '23503')
@skip_if_crdb
@skip_if_crdb("diagnostic")
@skip_before_postgres(9, 3)
def test_9_3_diagnostics(self):
cur = self.conn.cursor()
@ -302,7 +302,7 @@ class ExceptionsTestCase(ConnectingTestCase):
self.assertEqual(e.pgcode, e1.pgcode)
self.assert_(e1.cursor is None)
@skip_if_crdb
@skip_if_crdb("connect any db")
def test_pickle_connection_error(self):
# segfaults on psycopg 2.5.1 - see ticket #170
try:

2
tests/test_notify.py
View File

@ -38,7 +38,7 @@ import select
from subprocess import Popen, PIPE
@skip_if_crdb
@skip_if_crdb("notify")
class NotifiesTests(ConnectingTestCase):
def autocommit(self, conn):

4
tests/test_quote.py
View File

@ -121,7 +121,7 @@ class QuotingTestCase(ConnectingTestCase):
self.assertEqual(res, data)
self.assert_(not self.conn.notices)
@skip_if_crdb
@skip_if_crdb("encoding")
def test_latin1(self):
self.conn.set_client_encoding('LATIN1')
curs = self.conn.cursor()
@ -147,7 +147,7 @@ class QuotingTestCase(ConnectingTestCase):
self.assertEqual(res, data)
self.assert_(not self.conn.notices)
@skip_if_crdb
@skip_if_crdb("encoding")
def test_koi8(self):
self.conn.set_client_encoding('KOI8')
curs = self.conn.cursor()

2
tests/test_sql.py
View File

@ -152,7 +152,7 @@ class SqlFormatTests(ConnectingTestCase):
self.assertEqual(cur.fetchall(),
[(10, 'a', 'b', 'c'), (20, 'd', 'e', 'f')])
@skip_if_crdb
@skip_if_crdb("copy")
@skip_copy_if_green
@skip_before_postgres(8, 2)
def test_copy(self):

10
tests/test_transaction.py
View File

@ -26,22 +26,20 @@
import threading
import unittest
from .testutils import ConnectingTestCase, skip_before_postgres, slow
from .testutils import crdb_version, skip_if_crdb
from .testutils import skip_if_crdb
import psycopg2
from psycopg2.extensions import (
ISOLATION_LEVEL_SERIALIZABLE, STATUS_BEGIN, STATUS_READY)
@skip_if_crdb
class TransactionTests(ConnectingTestCase):
def setUp(self):
ConnectingTestCase.setUp(self)
skip_if_crdb("isolation level", self.conn)
self.conn.set_isolation_level(ISOLATION_LEVEL_SERIALIZABLE)
curs = self.conn.cursor()
if crdb_version(self.conn) is not None:
self.skipTest("features not supported on CockroachDB")
curs.execute('''
CREATE TEMPORARY TABLE table1 (
id int PRIMARY KEY
@ -96,7 +94,6 @@ class TransactionTests(ConnectingTestCase):
self.assertEqual(curs.fetchone()[0], 1)
@skip_if_crdb
class DeadlockSerializationTests(ConnectingTestCase):
"""Test deadlock and serialization failure errors."""
@ -107,8 +104,7 @@ class DeadlockSerializationTests(ConnectingTestCase):
def setUp(self):
ConnectingTestCase.setUp(self)
if crdb_version(self.conn) is not None:
self.skipTest("features not supported on CockroachDB")
skip_if_crdb("isolation level", self.conn)
curs = self.conn.cursor()
# Drop table if it already exists

8
tests/test_types_basic.py
View File

@ -149,14 +149,14 @@ class TypesBasicTests(ConnectingTestCase):
buf2 = self.execute("SELECT %s::bytea AS foo", (buf,))
self.assertEqual(s, buf2.tobytes())
@skip_if_crdb
@skip_if_crdb("nested array")
def testArray(self):
s = self.execute("SELECT %s AS foo", ([[1, 2], [3, 4]],))
self.failUnlessEqual(s, [[1, 2], [3, 4]])
s = self.execute("SELECT %s AS foo", (['one', 'two', 'three'],))
self.failUnlessEqual(s, ['one', 'two', 'three'])
@skip_if_crdb
@skip_if_crdb("nested array")
def testEmptyArrayRegression(self):
# ticket #42
curs = self.conn.cursor()
@ -239,7 +239,7 @@ class TypesBasicTests(ConnectingTestCase):
self.assert_(isinstance(x[0], bytes))
self.assertEqual(x, [b'a', b'b', b'c'])
@skip_if_crdb
@skip_if_crdb("nested array")
@testutils.skip_before_postgres(8, 2)
def testArrayOfNulls(self):
curs = self.conn.cursor()
@ -406,7 +406,7 @@ class TypesBasicTests(ConnectingTestCase):
a = self.execute("select '{1, 2, NULL}'::int4[]")
self.assertEqual(a, [2, 4, 'nada'])
@skip_if_crdb
@skip_if_crdb("cidr")
@testutils.skip_before_postgres(8, 2)
def testNetworkArray(self):
# we don't know these types, but we know their arrays

8
tests/test_types_extras.py
View File

@ -134,7 +134,7 @@ class TypesExtrasTests(ConnectingTestCase):
def skip_if_no_hstore(f):
@wraps(f)
@skip_if_crdb
@skip_if_crdb("hstore")
def skip_if_no_hstore_(self):
oids = HstoreAdapter.get_oids(self.conn)
if oids is None or not oids[0]:
@ -418,7 +418,7 @@ class HstoreTestCase(ConnectingTestCase):
def skip_if_no_composite(f):
@wraps(f)
@skip_if_crdb
@skip_if_crdb("composite")
def skip_if_no_composite_(self):
if self.conn.info.server_version < 80000:
return self.skipTest(
@ -788,7 +788,7 @@ def skip_if_no_json_type(f):
return skip_if_no_json_type_
@skip_if_crdb
@skip_if_crdb("json")
class JsonTestCase(ConnectingTestCase):
def test_adapt(self):
objs = [None, "te'xt", 123, 123.45,
@ -1332,7 +1332,7 @@ class RangeTestCase(unittest.TestCase):
self.assertEqual(result, expected)
@skip_if_crdb
@skip_if_crdb("range")
@skip_before_postgres(9, 2, "range not supported before postgres 9.2")
class RangeCasterTestCase(ConnectingTestCase):

4
tests/test_with.py
View File

@ -203,7 +203,7 @@ class WithCursorTestCase(WithTestCase):
self.assert_(curs.closed)
self.assert_(closes)
@skip_if_crdb
@skip_if_crdb("named cursor")
def test_exception_swallow(self):
# bug #262: __exit__ calls cur.close() that hides the exception
# with another error.
@ -217,7 +217,7 @@ class WithCursorTestCase(WithTestCase):
else:
self.fail("where is my exception?")
@skip_if_crdb
@skip_if_crdb("named cursor")
@skip_before_postgres(8, 2)
def test_named_with_noop(self):
with self.conn.cursor('named'):

33
tests/testutils.py
View File

@ -248,8 +248,7 @@ def skip_if_tpc_disabled(f):
@wraps(f)
def skip_if_tpc_disabled_(self):
cnn = self.connect()
if crdb_version(cnn):
self.skipTest("two phase transction not supported on CockroachDB")
skip_if_crdb("2-phase commit", cnn)
cur = cnn.cursor()
try:
@ -439,15 +438,29 @@ def crdb_version(conn, __crdb_version=[]):
return __crdb_version[0]
@decorate_all_tests
def skip_if_crdb(f):
"""Skip a test or test class if we are testing against CockroachDB."""
def skip_if_crdb(reason, conn=None):
"""Skip a test or test class if we are testing against CockroachDB.
@wraps(f)
def skip_if_crdb_(self, *args, **kwargs):
if crdb_version(self.connect()) is not None:
self.skipTest("not supported on CockroachDB")
return f(self, *args, **kwargs)
Can be used as a decorator for tests function or classes:
@skip_if_crdb("my reason")
class SomeUnitTest(UnitTest):
# ...
Or as a normal function if the *conn* argument is passed.
"""
if conn is not None:
if crdb_version(conn) is not None:
raise unittest.SkipTest("not supported on CockroachDB: %s" % reason)
@decorate_all_tests
def skip_if_crdb_(f):
@wraps(f)
def skip_if_crdb__(self, *args, **kwargs):
skip_if_crdb(reason, self.connect())
return f(self, *args, **kwargs)
return skip_if_crdb__
return skip_if_crdb_