Deal uniformly with test servers without pg_sleep.

tests/test_async.py

@@ -1,5 +1,5 @@
 #!/usr/bin/env python
-from testutils import unittest
+from testutils import unittest, skip_if_no_pg_sleep
 
 import psycopg2
 from psycopg2 import extensions
@@ -94,17 +94,15 @@ class AsyncTests(unittest.TestCase):
         self.assertFalse(self.conn.isexecuting())
         self.assertEquals(cur.fetchone()[0], "a")
 
+    @skip_if_no_pg_sleep('conn')
     def test_async_callproc(self):
         cur = self.conn.cursor()
-        try:
-            cur.callproc("pg_sleep", (0.1, ))
-            self.assertTrue(self.conn.isexecuting())
+        cur.callproc("pg_sleep", (0.1, ))
+        self.assertTrue(self.conn.isexecuting())
 
-            self.wait(cur)
-            self.assertFalse(self.conn.isexecuting())
-            self.assertEquals(cur.fetchall()[0][0], '')
-        except psycopg2.ProgrammingError:
-            return self.skipTest("PG < 8.1 did not have pg_sleep")
+        self.wait(cur)
+        self.assertFalse(self.conn.isexecuting())
+        self.assertEquals(cur.fetchall()[0][0], '')
 
     def test_async_after_async(self):
         cur = self.conn.cursor()

tests/test_cancel.py

@@ -2,7 +2,7 @@
 
 import time
 import threading
-import unittest
+from testutils import unittest, skip_if_no_pg_sleep
 
 import tests
 import psycopg2
@@ -27,6 +27,7 @@ class CancelTests(unittest.TestCase):
     def test_empty_cancel(self):
         self.conn.cancel()
 
+    @skip_if_no_pg_sleep('conn')
     def test_cancel(self):
         errors = []
 
@@ -62,6 +63,7 @@ class CancelTests(unittest.TestCase):
 
         self.assertEqual(errors, [])
 
+    @skip_if_no_pg_sleep('conn')
     def test_async_cancel(self):
         async_conn = psycopg2.connect(tests.dsn, async=True)
         self.assertRaises(psycopg2.OperationalError, async_conn.cancel)

tests/test_connection.py

@@ -2,7 +2,7 @@
 
 import time
 import threading
-from testutils import unittest, decorate_all_tests
+from testutils import unittest, decorate_all_tests, skip_if_no_pg_sleep
 from operator import attrgetter
 
 import psycopg2
@@ -92,6 +92,7 @@ class ConnectionTests(unittest.TestCase):
         self.assertRaises(psycopg2.NotSupportedError,
             cnn.xid, 42, "foo", "bar")
 
+    @skip_if_no_pg_sleep('connect')
     def test_concurrent_execution(self):
         def slave(cnn):
             cur = cnn.cursor()
@@ -111,6 +112,7 @@ class ConnectionTests(unittest.TestCase):
         self.assert_(time.time() - t0 < 3,
             "something broken in concurrency")
 
+
 class IsolationLevelsTestCase(unittest.TestCase):
 
     def setUp(self):

tests/test_transaction.py

@@ -1,6 +1,6 @@
 #!/usr/bin/env python
 import threading
-from testutils import unittest
+from testutils import unittest, skip_if_no_pg_sleep
 
 import psycopg2
 from psycopg2.extensions import (
@@ -211,15 +211,13 @@ class QueryCancellationTests(unittest.TestCase):
         self.conn = psycopg2.connect(tests.dsn)
         self.conn.set_isolation_level(ISOLATION_LEVEL_SERIALIZABLE)
 
+    @skip_if_no_pg_sleep('conn')
     def test_statement_timeout(self):
         curs = self.conn.cursor()
         # Set a low statement timeout, then sleep for a longer period.
         curs.execute('SET statement_timeout TO 10')
-        try:
-            self.assertRaises(psycopg2.extensions.QueryCanceledError,
-                              curs.execute, 'SELECT pg_sleep(50)')
-        except psycopg2.ProgrammingError:
-            return self.skipTest("pg_sleep not available")
+        self.assertRaises(psycopg2.extensions.QueryCanceledError,
+                          curs.execute, 'SELECT pg_sleep(50)')
 
 
 def test_suite():

tests/testutils.py

@@ -44,4 +44,28 @@ def decorate_all_tests(cls, decorator):
     for n in dir(cls):
         if n.startswith('test'):
             setattr(cls, n, decorator(getattr(cls, n)))
-            
+
+
+def skip_if_no_pg_sleep(name):
+    """Decorator to skip a test if pg_sleep is not supported by the server.
+
+    Pass it the name of an attribute containing a connection or of a method
+    returning a connection.
+    """
+    def skip_if_no_pg_sleep_(f):
+        def skip_if_no_pg_sleep__(self):
+            cnn = getattr(self, name)
+            if callable(cnn):
+                cnn = cnn()
+
+            if cnn.server_version < 80100:
+                return self.skipTest(
+                    "server version %s doesn't support pg_sleep"
+                    % cnn.server_version)
+
+            return f(self)
+
+        skip_if_no_pg_sleep__.__name__ = f.__name__
+        return skip_if_no_pg_sleep__
+
+    return skip_if_no_pg_sleep_