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

Some extra cursors test skipped on CockroachDB

Browse Source 
Skip named cursor tests
This commit is contained in:
Daniele Varrazzo 2020-07-22 02:14:18 +01:00
parent 701637b5fa
commit a9153ac373
2 changed files with 21 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
tests/test_extras_dictcursor.py
View File

@ -27,13 +27,15 @@ import psycopg2.extras
from psycopg2.extras import NamedTupleConnection, NamedTupleCursor
from .testutils import ConnectingTestCase, skip_before_postgres, \
skip_before_python, skip_from_python
skip_before_python, skip_from_python, crdb_version, skip_if_crdb
class _DictCursorBase(ConnectingTestCase):
def setUp(self):
ConnectingTestCase.setUp(self)
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 ExtrasDictCursorTests (foo text)")
curs.execute("INSERT INTO ExtrasDictCursorTests VALUES ('bar')")
self.conn.commit()
@ -62,6 +64,7 @@ class _DictCursorBase(ConnectingTestCase):
class ExtrasDictCursorTests(_DictCursorBase):
"""Test if DictCursor extension class works."""
@skip_if_crdb
def testDictConnCursorArgs(self):
self.conn.close()
self.conn = self.connect(connection_factory=psycopg2.extras.DictConnection)
@ -129,16 +132,19 @@ class ExtrasDictCursorTests(_DictCursorBase):
return row
self._testWithNamedCursor(getter)
@skip_if_crdb
@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_before_postgres(8, 0)
def testDictCursorWithNamedCursorIterRowNumber(self):
curs = self.conn.cursor('tmp', cursor_factory=psycopg2.extras.DictCursor)
self._testIterRowNumber(curs)
@skip_if_crdb
def _testWithNamedCursor(self, getter):
curs = self.conn.cursor('aname', cursor_factory=psycopg2.extras.DictCursor)
curs.execute("SELECT * FROM ExtrasDictCursorTests")
@ -314,16 +320,19 @@ class ExtrasDictCursorRealTests(_DictCursorBase):
return row
self._testWithNamedCursorReal(getter)
@skip_if_crdb
@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_before_postgres(8, 0)
def testDictCursorRealWithNamedCursorIterRowNumber(self):
curs = self.conn.cursor('tmp', cursor_factory=psycopg2.extras.RealDictCursor)
self._testIterRowNumber(curs)
@skip_if_crdb
def _testWithNamedCursorReal(self, getter):
curs = self.conn.cursor('aname',
cursor_factory=psycopg2.extras.RealDictCursor)
@ -429,12 +438,15 @@ class NamedTupleCursorTest(ConnectingTestCase):
self.conn = self.connect(connection_factory=NamedTupleConnection)
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 nttest (i int, s text)")
curs.execute("INSERT INTO nttest VALUES (1, 'foo')")
curs.execute("INSERT INTO nttest VALUES (2, 'bar')")
curs.execute("INSERT INTO nttest VALUES (3, 'baz')")
self.conn.commit()
@skip_if_crdb
def test_cursor_args(self):
cur = self.conn.cursor('foo', cursor_factory=psycopg2.extras.DictCursor)
self.assertEqual(cur.name, 'foo')
@ -592,6 +604,7 @@ class NamedTupleCursorTest(ConnectingTestCase):
finally:
NamedTupleCursor._make_nt = f_orig
@skip_if_crdb
@skip_before_postgres(8, 0)
def test_named(self):
curs = self.conn.cursor('tmp')
@ -602,24 +615,28 @@ class NamedTupleCursorTest(ConnectingTestCase):
recs.extend(curs.fetchall())
self.assertEqual(list(range(10)), [t.i for t in recs])
@skip_if_crdb
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
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
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_before_postgres(8, 2)
def test_not_greedy(self):
curs = self.conn.cursor('tmp')
@ -634,6 +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_before_postgres(8, 0)
def test_named_rownumber(self):
curs = self.conn.cursor('tmp')

4
tests/testutils.py
View File

@ -444,10 +444,10 @@ def skip_if_crdb(f):
"""Skip a test or test class if we are testing against CockroachDB."""
@wraps(f)
def skip_if_crdb_(self):
def skip_if_crdb_(self, *args, **kwargs):
if crdb_version(self.connect()) is not None:
self.skipTest("not supported on CockroachDB")
return f(self)
return f(self, *args, **kwargs)
return skip_if_crdb_