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

All the sync tests pass on CockroachDB

Browse Source 
Added decorator to skip tests on crdb
This commit is contained in:
Daniele Varrazzo 2020-07-21 01:55:22 +01:00
parent 659910ee81
commit ee34198bf6
2 changed files with 23 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
tests/test_async.py
View File

@ -34,7 +34,7 @@ import psycopg2.errors
from psycopg2 import extensions as ext from psycopg2 import extensions as ext
from .testutils import (ConnectingTestCase, StringIO, skip_before_postgres, from .testutils import (ConnectingTestCase, StringIO, skip_before_postgres,
crdb_version, slow) skip_if_crdb, crdb_version, slow)
class PollableStub(object): class PollableStub(object):
@ -114,7 +114,6 @@ class AsyncTests(ConnectingTestCase):
self.wait(cur) self.wait(cur)
self.assertFalse(self.conn.isexecuting()) self.assertFalse(self.conn.isexecuting())
self.assertEquals(cur.fetchall()[0][0], '')
@slow @slow
def test_async_after_async(self): def test_async_after_async(self):
@ -355,6 +354,7 @@ class AsyncTests(ConnectingTestCase):
self.assertEquals(cur.fetchone()[0], 1) self.assertEquals(cur.fetchone()[0], 1)
@slow @slow
@skip_if_crdb
def test_notify(self): def test_notify(self):
cur = self.conn.cursor() cur = self.conn.cursor()
sync_cur = self.sync_conn.cursor() sync_cur = self.sync_conn.cursor()
@ -399,9 +399,10 @@ class AsyncTests(ConnectingTestCase):
self.assertRaises(psycopg2.IntegrityError, self.wait, cur) self.assertRaises(psycopg2.IntegrityError, self.wait, cur)
cur.execute("insert into table1 values (%s); " cur.execute("insert into table1 values (%s); "
"insert into table1 values (%s)", (2, 2)) "insert into table1 values (%s)", (2, 2))
# this should fail as well # this should fail as well (Postgres behaviour)
self.assertRaises(psycopg2.IntegrityError, self.wait, cur) self.assertRaises(psycopg2.IntegrityError, self.wait, cur)
# but this should work # but this should work
if crdb_version(self.sync_conn) is None:
cur.execute("insert into table1 values (%s)", (2, )) cur.execute("insert into table1 values (%s)", (2, ))
self.wait(cur) self.wait(cur)
# and the cursor should be usable afterwards # and the cursor should be usable afterwards
@ -431,6 +432,7 @@ class AsyncTests(ConnectingTestCase):
self.wait(cur2) self.wait(cur2)
self.assertEquals(cur2.fetchone()[0], 1) self.assertEquals(cur2.fetchone()[0], 1)
@skip_if_crdb
def test_notices(self): def test_notices(self):
del self.conn.notices[:] del self.conn.notices[:]
cur = self.conn.cursor() cur = self.conn.cursor()
@ -455,6 +457,7 @@ class AsyncTests(ConnectingTestCase):
self.wait(self.conn) self.wait(self.conn)
self.assertEqual(cur.fetchone(), (42,)) self.assertEqual(cur.fetchone(), (42,))
@skip_if_crdb
def test_async_connection_error_message(self): def test_async_connection_error_message(self):
try: try:
cnn = psycopg2.connect('dbname=thisdatabasedoesntexist', async_=True) cnn = psycopg2.connect('dbname=thisdatabasedoesntexist', async_=True)
@ -472,6 +475,7 @@ class AsyncTests(ConnectingTestCase):
self.assertRaises(psycopg2.ProgrammingError, self.wait, self.conn) self.assertRaises(psycopg2.ProgrammingError, self.wait, self.conn)
@slow @slow
@skip_if_crdb
@skip_before_postgres(9, 0) @skip_before_postgres(9, 0)
def test_non_block_after_notification(self): def test_non_block_after_notification(self):
from select import select from select import select
@ -505,6 +509,7 @@ class AsyncTests(ConnectingTestCase):
def test_poll_noop(self): def test_poll_noop(self):
self.conn.poll() self.conn.poll()
@skip_if_crdb
@skip_before_postgres(9, 0) @skip_before_postgres(9, 0)
def test_poll_conn_for_notification(self): def test_poll_conn_for_notification(self):
with self.conn.cursor() as cur: with self.conn.cursor() as cur:

13
tests/testutils.py
View File

@ -439,6 +439,19 @@ def crdb_version(conn, __crdb_version=[]):
return __crdb_version[0] return __crdb_version[0]
@decorate_all_tests
def skip_if_crdb(f):
"""Skip a test or test class if we are testing against CockroachDB."""
@wraps(f)
def skip_if_crdb_(self):
if crdb_version(self.connect()) is not None:
self.skipTest("not supported on CockroachDB")
return f(self)
return skip_if_crdb_
class py3_raises_typeerror(object): class py3_raises_typeerror(object):
def __enter__(self): def __enter__(self):
pass pass