diff options
| -rwxr-xr-x | tests/test_connection.py | 26 | ||||
| -rwxr-xr-x | tests/test_psycopg2_dbapi20.py | 2 | ||||
| -rw-r--r-- | tests/testutils.py | 27 |
3 files changed, 29 insertions, 26 deletions
diff --git a/tests/test_connection.py b/tests/test_connection.py index f182190..83661d9 100755 --- a/tests/test_connection.py +++ b/tests/test_connection.py @@ -307,30 +307,6 @@ class IsolationLevelsTestCase(unittest.TestCase): self.assertEqual(2, cur2.fetchone()[0]) -def skip_if_tpc_disabled(f): - """Skip a test if the server has tpc support disabled.""" - def skip_if_tpc_disabled_(self): - cnn = self.connect() - cur = cnn.cursor() - try: - cur.execute("SHOW max_prepared_transactions;") - except psycopg2.ProgrammingError: - return self.skipTest( - "server too old: two phase transactions not supported.") - else: - mtp = int(cur.fetchone()[0]) - cnn.close() - - if not mtp: - return self.skipTest( - "server not configured for two phase transactions. " - "set max_prepared_transactions to > 0 to run the test") - return f(self) - - skip_if_tpc_disabled_.__name__ = f.__name__ - return skip_if_tpc_disabled_ - - class ConnectionTwoPhaseTests(unittest.TestCase): def setUp(self): self._conns = [] @@ -346,7 +322,6 @@ class ConnectionTwoPhaseTests(unittest.TestCase): if not conn.closed: conn.close() - def clear_test_xacts(self): """Rollback all the prepared transaction in the testing db.""" cnn = self.connect() @@ -700,6 +675,7 @@ class ConnectionTwoPhaseTests(unittest.TestCase): cnn.tpc_prepare() self.assertRaises(psycopg2.ProgrammingError, cnn.cancel) +from testutils import skip_if_tpc_disabled decorate_all_tests(ConnectionTwoPhaseTests, skip_if_tpc_disabled) diff --git a/tests/test_psycopg2_dbapi20.py b/tests/test_psycopg2_dbapi20.py index 5eec97a..222d934 100755 --- a/tests/test_psycopg2_dbapi20.py +++ b/tests/test_psycopg2_dbapi20.py @@ -24,7 +24,7 @@ import dbapi20 import dbapi20_tpc -from test_connection import skip_if_tpc_disabled +from testutils import skip_if_tpc_disabled from testutils import unittest, decorate_all_tests import psycopg2 diff --git a/tests/testutils.py b/tests/testutils.py index 6d04fc0..e0b8eaf 100644 --- a/tests/testutils.py +++ b/tests/testutils.py @@ -88,3 +88,30 @@ def skip_if_no_pg_sleep(name): return skip_if_no_pg_sleep__ return skip_if_no_pg_sleep_ + + +def skip_if_tpc_disabled(f): + """Skip a test if the server has tpc support disabled.""" + def skip_if_tpc_disabled_(self): + from psycopg2 import ProgrammingError + cnn = self.connect() + cur = cnn.cursor() + try: + cur.execute("SHOW max_prepared_transactions;") + except ProgrammingError: + return self.skipTest( + "server too old: two phase transactions not supported.") + else: + mtp = int(cur.fetchone()[0]) + cnn.close() + + if not mtp: + return self.skipTest( + "server not configured for two phase transactions. " + "set max_prepared_transactions to > 0 to run the test") + return f(self) + + skip_if_tpc_disabled_.__name__ = f.__name__ + return skip_if_tpc_disabled_ + + |