mirror of
https://github.com/psycopg/psycopg2.git
synced 2025-02-07 12:50:32 +03:00
Skip GC bug test if uuid is not available
This commit is contained in:
parent
06059a216f
commit
ca3d9da83b
|
@ -34,15 +34,23 @@ if sys.version_info < (3,):
|
||||||
else:
|
else:
|
||||||
import py3tests as tests
|
import py3tests as tests
|
||||||
|
|
||||||
|
from testutils import skip_if_no_uuid
|
||||||
|
|
||||||
class StolenReferenceTestCase(unittest.TestCase):
|
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 test_stolen_reference_bug(self):
|
||||||
def fish(val, cur):
|
def fish(val, cur):
|
||||||
gc.collect()
|
gc.collect()
|
||||||
return 42
|
return 42
|
||||||
conn = psycopg2.connect(tests.dsn)
|
|
||||||
UUID = psycopg2.extensions.new_type((2950,), "UUID", fish)
|
UUID = psycopg2.extensions.new_type((2950,), "UUID", fish)
|
||||||
psycopg2.extensions.register_type(UUID, conn)
|
psycopg2.extensions.register_type(UUID, self.conn)
|
||||||
curs = conn.cursor()
|
curs = self.conn.cursor()
|
||||||
curs.execute("select 'b5219e01-19ab-4994-b71e-149225dc51e4'::uuid")
|
curs.execute("select 'b5219e01-19ab-4994-b71e-149225dc51e4'::uuid")
|
||||||
curs.fetchone()
|
curs.fetchone()
|
||||||
|
|
||||||
|
|
|
@ -65,6 +65,29 @@ def decorate_all_tests(cls, decorator):
|
||||||
setattr(cls, n, decorator(getattr(cls, n)))
|
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):
|
def skip_if_no_pg_sleep(name):
|
||||||
"""Decorator to skip a test if pg_sleep is not supported by the server.
|
"""Decorator to skip a test if pg_sleep is not supported by the server.
|
||||||
|
|
||||||
|
|
|
@ -22,34 +22,13 @@ import re
|
||||||
import sys
|
import sys
|
||||||
from datetime import date
|
from datetime import date
|
||||||
|
|
||||||
from testutils import unittest
|
from testutils import unittest, skip_if_no_uuid
|
||||||
|
|
||||||
import psycopg2
|
import psycopg2
|
||||||
import psycopg2.extras
|
import psycopg2.extras
|
||||||
import tests
|
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):
|
def filter_scs(conn, s):
|
||||||
if conn.get_parameter_status("standard_conforming_strings") == 'off':
|
if conn.get_parameter_status("standard_conforming_strings") == 'off':
|
||||||
return s
|
return s
|
||||||
|
|
Loading…
Reference in New Issue
Block a user