1 files changed, 48 insertions, 3 deletions
diff --git a/tests/testutils.py b/tests/testutils.py
index cd034e5..6de3a4b 100644
--- a/tests/testutils.py
+++ b/tests/testutils.py
@@ -1,6 +1,25 @@
-# Utility module for psycopg2 testing.
-# 
-# Copyright (C) 2010 Daniele Varrazzo <daniele.varrazzo@gmail.com>
+# testutils.py - utility module for psycopg2 testing.
+#
+# Copyright (C) 2010-2011 Daniele Varrazzo  <daniele.varrazzo@gmail.com>
+#
+# psycopg2 is free software: you can redistribute it and/or modify it
+# under the terms of the GNU Lesser General Public License as published
+# by the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# In addition, as a special exception, the copyright holders give
+# permission to link this program with the OpenSSL library (or with
+# modified versions of OpenSSL that use the same license as OpenSSL),
+# and distribute linked combinations including the two.
+#
+# You must obey the GNU Lesser General Public License in all respects for
+# all of the code used other than OpenSSL.
+#
+# psycopg2 is distributed in the hope that it will be useful, but WITHOUT
+# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+# FITNESS FOR A PARTICULAR PURPOSE.  See the GNU Lesser General Public
+# License for more details.
+
 
 # Use unittest2 if available. Otherwise mock a skip facility with warnings.
 
@@ -83,6 +102,32 @@ def skip_if_no_pg_sleep(name):
 
     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_
+
+
 def skip_on_python2(f):
     """Skip a test on Python 3 and following."""
     def skip_on_python2_(self):