From 8e1257d7d0eecce5f2f9f3acb187ca4c6c4a6bc2 Mon Sep 17 00:00:00 2001 From: Daniele Varrazzo Date: Tue, 18 Jan 2011 00:57:27 +0000 Subject: [PATCH 1/4] Fixed check for pg_sleep availability in tests The function is available since PG 8.2. --- tests/testutils.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/testutils.py b/tests/testutils.py index e0b8eaf2..c1529581 100644 --- a/tests/testutils.py +++ b/tests/testutils.py @@ -77,7 +77,7 @@ def skip_if_no_pg_sleep(name): if callable(cnn): cnn = cnn() - if cnn.server_version < 80100: + if cnn.server_version < 80200: return self.skipTest( "server version %s doesn't support pg_sleep" % cnn.server_version) From 06059a216fca3a3db0ce1b657c63a38179b617fe Mon Sep 17 00:00:00 2001 From: Daniele Varrazzo Date: Tue, 18 Jan 2011 00:58:33 +0000 Subject: [PATCH 2/4] Fixed hstore test in PG 9.x with standard_conforming_strings off --- tests/types_extras.py | 11 +++++++---- 1 file changed, 7 insertions(+), 4 deletions(-) diff --git a/tests/types_extras.py b/tests/types_extras.py index a174ba4f..d26740e7 100755 --- a/tests/types_extras.py +++ b/tests/types_extras.py @@ -201,13 +201,16 @@ class HstoreTestCase(unittest.TestCase): ii = zip(kk, vv) ii.sort() + def f(*args): + return tuple([filter_scs(self.conn, s) for s in args]) + self.assertEqual(len(ii), len(o)) - self.assertEqual(ii[0], ("E'a'", "E'1'")) - self.assertEqual(ii[1], ("E'b'", "E''''")) - self.assertEqual(ii[2], ("E'c'", "NULL")) + self.assertEqual(ii[0], f("E'a'", "E'1'")) + self.assertEqual(ii[1], f("E'b'", "E''''")) + self.assertEqual(ii[2], f("E'c'", "NULL")) if 'd' in o: encc = u'\xe0'.encode(psycopg2.extensions.encodings[self.conn.encoding]) - self.assertEqual(ii[3], ("E'd'", "E'%s'" % encc)) + self.assertEqual(ii[3], f("E'd'", "E'%s'" % encc)) def test_parse(self): from psycopg2.extras import HstoreAdapter From ca3d9da83b37388e251280238062dcc95ce69325 Mon Sep 17 00:00:00 2001 From: Daniele Varrazzo Date: Tue, 18 Jan 2011 01:46:10 +0000 Subject: [PATCH 3/4] Skip GC bug test if uuid is not available --- tests/bug_gc.py | 14 +++++++++++--- tests/testutils.py | 23 +++++++++++++++++++++++ tests/types_extras.py | 23 +---------------------- 3 files changed, 35 insertions(+), 25 deletions(-) diff --git a/tests/bug_gc.py b/tests/bug_gc.py index 71dac1c8..b3d80d1d 100755 --- a/tests/bug_gc.py +++ b/tests/bug_gc.py @@ -34,15 +34,23 @@ if sys.version_info < (3,): else: import py3tests as tests +from testutils import skip_if_no_uuid + class StolenReferenceTestCase(unittest.TestCase): + def setUp(self): + self.conn = psycopg2.connect(tests.dsn) + + def tearDown(self): + self.conn.close() + + @skip_if_no_uuid def test_stolen_reference_bug(self): def fish(val, cur): gc.collect() return 42 - conn = psycopg2.connect(tests.dsn) UUID = psycopg2.extensions.new_type((2950,), "UUID", fish) - psycopg2.extensions.register_type(UUID, conn) - curs = conn.cursor() + psycopg2.extensions.register_type(UUID, self.conn) + curs = self.conn.cursor() curs.execute("select 'b5219e01-19ab-4994-b71e-149225dc51e4'::uuid") curs.fetchone() diff --git a/tests/testutils.py b/tests/testutils.py index c1529581..cfeb0e05 100644 --- a/tests/testutils.py +++ b/tests/testutils.py @@ -65,6 +65,29 @@ def decorate_all_tests(cls, decorator): setattr(cls, n, decorator(getattr(cls, n))) +def skip_if_no_uuid(f): + """Decorator to skip a test if uuid is not supported by Py/PG.""" + def skip_if_no_uuid_(self): + try: + import uuid + except ImportError: + return self.skipTest("uuid not available in this Python version") + + try: + cur = self.conn.cursor() + cur.execute("select typname from pg_type where typname = 'uuid'") + has = cur.fetchone() + finally: + self.conn.rollback() + + if has: + return f(self) + else: + return self.skipTest("uuid type not available on the server") + + return skip_if_no_uuid_ + + def skip_if_no_pg_sleep(name): """Decorator to skip a test if pg_sleep is not supported by the server. diff --git a/tests/types_extras.py b/tests/types_extras.py index d26740e7..010449b2 100755 --- a/tests/types_extras.py +++ b/tests/types_extras.py @@ -22,34 +22,13 @@ import re import sys from datetime import date -from testutils import unittest +from testutils import unittest, skip_if_no_uuid import psycopg2 import psycopg2.extras import tests -def skip_if_no_uuid(f): - def skip_if_no_uuid_(self): - try: - import uuid - except ImportError: - return self.skipTest("uuid not available in this Python version") - - try: - cur = self.conn.cursor() - cur.execute("select typname from pg_type where typname = 'uuid'") - has = cur.fetchone() - finally: - self.conn.rollback() - - if has: - return f(self) - else: - return self.skipTest("uuid type not available on the server") - - return skip_if_no_uuid_ - def filter_scs(conn, s): if conn.get_parameter_status("standard_conforming_strings") == 'off': return s From 153c30f24bca8d3c8fda9ed8bc7dbdb9591f58b9 Mon Sep 17 00:00:00 2001 From: Daniele Varrazzo Date: Tue, 18 Jan 2011 01:54:33 +0000 Subject: [PATCH 4/4] Skip composite type tests if the server doesn't support them --- tests/types_extras.py | 18 ++++++++++++++++++ 1 file changed, 18 insertions(+) diff --git a/tests/types_extras.py b/tests/types_extras.py index 010449b2..6b258863 100755 --- a/tests/types_extras.py +++ b/tests/types_extras.py @@ -333,6 +333,18 @@ class HstoreTestCase(unittest.TestCase): ok(dict(zip(ab, ab))) +def skip_if_no_composite(f): + def skip_if_no_composite_(self): + if self.conn.server_version < 80000: + return self.skipTest( + "server version %s doesn't support composite types" + % self.conn.server_version) + + return f(self) + + skip_if_no_composite_.__name__ = f.__name__ + return skip_if_no_composite_ + class AdaptTypeTestCase(unittest.TestCase): def setUp(self): self.conn = psycopg2.connect(tests.dsn) @@ -340,6 +352,7 @@ class AdaptTypeTestCase(unittest.TestCase): def tearDown(self): self.conn.close() + @skip_if_no_composite def test_none_in_record(self): curs = self.conn.cursor() s = curs.mogrify("SELECT %s;", [(42, None)]) @@ -397,6 +410,7 @@ class AdaptTypeTestCase(unittest.TestCase): '^_`abcdefghijklmnopqrstuvwxyz{|}~\x7f")', [None, ''.join(map(chr, range(1, 128)))]) + @skip_if_no_composite def test_cast_composite(self): oid = self._create_type("type_isd", [('anint', 'integer'), ('astring', 'text'), ('adate', 'date')]) @@ -427,6 +441,7 @@ class AdaptTypeTestCase(unittest.TestCase): self.assertEqual(v.astring, "hello") self.assertEqual(v.adate, date(2011,1,2)) + @skip_if_no_composite def test_cast_nested(self): self._create_type("type_is", [("anint", "integer"), ("astring", "text")]) @@ -453,6 +468,7 @@ class AdaptTypeTestCase(unittest.TestCase): else: self.assertEqual(v.anotherpair.apair.astring, "hello") + @skip_if_no_composite def test_register_on_cursor(self): self._create_type("type_ii", [("a", "integer"), ("b", "integer")]) @@ -464,6 +480,7 @@ class AdaptTypeTestCase(unittest.TestCase): curs2.execute("select (1,2)::type_ii") self.assertEqual(curs2.fetchone()[0], "(1,2)") + @skip_if_no_composite def test_register_on_connection(self): self._create_type("type_ii", [("a", "integer"), ("b", "integer")]) @@ -481,6 +498,7 @@ class AdaptTypeTestCase(unittest.TestCase): conn1.close() conn2.close() + @skip_if_no_composite def test_register_globally(self): self._create_type("type_ii", [("a", "integer"), ("b", "integer")])