1
1
mirror of https://github.com/psycopg/psycopg2.git synced 2025-02-25 21:20:32 +03:00

Test decorator moved into the test utilities module

This commit is contained in:
Daniele Varrazzo 2011-01-09 23:44:58 +00:00
parent 15a09da96d
commit 38641b93ea
3 changed files with 29 additions and 26 deletions

View File

@ -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)

View File

@ -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

View File

@ -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_