1
1
mirror of https://github.com/psycopg/psycopg2.git synced 2025-02-26 13:40:33 +03:00

Use class decorators to decorate all tests

Test decorators changed so that they can be applied either to a method
or to a class. Of course their double nature is implemented by a
decorator.
This commit is contained in:
Daniele Varrazzo 2018-10-30 00:23:56 +00:00
parent 60935b9b3d
commit 2f24a2e22d
9 changed files with 81 additions and 87 deletions

View File

@ -35,7 +35,7 @@ import psycopg2.errorcodes
from psycopg2 import extensions as ext
from .testutils import (
unittest, decorate_all_tests, skip_if_no_superuser, skip_before_postgres,
unittest, skip_if_no_superuser, skip_before_postgres,
skip_after_postgres, skip_before_libpq, skip_after_libpq,
ConnectingTestCase, skip_if_tpc_disabled, skip_if_windows, slow)
@ -812,6 +812,7 @@ class IsolationLevelsTestCase(ConnectingTestCase):
self.conn.isolation_level
@skip_if_tpc_disabled
class ConnectionTwoPhaseTests(ConnectingTestCase):
def setUp(self):
ConnectingTestCase.setUp(self)
@ -1183,9 +1184,6 @@ class ConnectionTwoPhaseTests(ConnectingTestCase):
self.assertEqual(None, xid.bqual)
decorate_all_tests(ConnectionTwoPhaseTests, skip_if_tpc_disabled)
class TransactionControlTests(ConnectingTestCase):
def test_closed(self):
self.conn.close()

View File

@ -25,8 +25,7 @@
import sys
import string
import unittest
from .testutils import (ConnectingTestCase, decorate_all_tests,
skip_before_postgres, slow, StringIO)
from .testutils import (ConnectingTestCase, skip_before_postgres, slow, StringIO)
from itertools import cycle
from subprocess import Popen, PIPE
@ -63,6 +62,7 @@ class MinimalWrite(_base):
return self.f.write(data)
@skip_copy_if_green
class CopyTests(ConnectingTestCase):
def setUp(self):
@ -378,9 +378,6 @@ conn.close()
curs.copy_to, BrokenWrite(), "tcopy")
decorate_all_tests(CopyTests, skip_copy_if_green)
def test_suite():
return unittest.TestLoader().loadTestsFromName(__name__)

View File

@ -134,6 +134,7 @@ class TestExecuteBatch(FastExecuteTestMixin, testutils.ConnectingTestCase):
self.assertEqual(cur.fetchone(), (3, snowman))
@testutils.skip_before_postgres(8, 2)
class TestExecuteValues(FastExecuteTestMixin, testutils.ConnectingTestCase):
def test_empty(self):
cur = self.conn.cursor()
@ -251,10 +252,6 @@ class TestExecuteValues(FastExecuteTestMixin, testutils.ConnectingTestCase):
self.assertEqual(cur.fetchall(), [(1, 'hi')])
testutils.decorate_all_tests(TestExecuteValues,
testutils.skip_before_postgres(8, 2))
def test_suite():
return unittest.TestLoader().loadTestsFromName(__name__)

View File

@ -26,6 +26,7 @@ import psycopg2
import psycopg2.extras
@testutils.decorate_all_tests
def skip_if_no_ipaddress(f):
@wraps(f)
def skip_if_no_ipaddress_(self):
@ -40,6 +41,7 @@ def skip_if_no_ipaddress(f):
return skip_if_no_ipaddress_
@skip_if_no_ipaddress
class NetworkingTestCase(testutils.ConnectingTestCase):
def test_inet_cast(self):
import ipaddress as ip
@ -126,9 +128,6 @@ class NetworkingTestCase(testutils.ConnectingTestCase):
self.assertEquals(cur.fetchone()[0], '::ffff:102:300/128')
testutils.decorate_all_tests(NetworkingTestCase, skip_if_no_ipaddress)
def test_suite():
return unittest.TestLoader().loadTestsFromName(__name__)

View File

@ -31,19 +31,11 @@ import psycopg2
import psycopg2.extensions
import unittest
from .testutils import (decorate_all_tests, skip_if_tpc_disabled,
ConnectingTestCase, skip_if_green, slow)
skip_before_postgres, ConnectingTestCase, skip_if_green, slow)
def skip_if_no_lo(f):
@wraps(f)
def skip_if_no_lo_(self):
if self.conn.info.server_version < 80100:
return self.skipTest("large objects only supported from PG 8.1")
else:
return f(self)
return skip_if_no_lo_
skip_if_no_lo = skip_before_postgres(8, 1,
"large objects only supported from PG 8.1")
skip_lo_if_green = skip_if_green("libpq doesn't support LO in async mode")
@ -73,6 +65,8 @@ class LargeObjectTestCase(ConnectingTestCase):
ConnectingTestCase.tearDown(self)
@skip_if_no_lo
@skip_lo_if_green
class LargeObjectTests(LargeObjectTestCase):
def test_create(self):
lo = self.conn.lobject()
@ -399,9 +393,7 @@ class LargeObjectTests(LargeObjectTestCase):
self.assert_(isinstance(lo, lobject_subclass))
decorate_all_tests(LargeObjectTests, skip_if_no_lo, skip_lo_if_green)
@decorate_all_tests
def skip_if_no_truncate(f):
@wraps(f)
def skip_if_no_truncate_(self):
@ -419,6 +411,9 @@ def skip_if_no_truncate(f):
return skip_if_no_truncate_
@skip_if_no_lo
@skip_lo_if_green
@skip_if_no_truncate
class LargeObjectTruncateTests(LargeObjectTestCase):
def test_truncate(self):
lo = self.conn.lobject()
@ -456,10 +451,6 @@ class LargeObjectTruncateTests(LargeObjectTestCase):
self.assertRaises(psycopg2.ProgrammingError, lo.truncate)
decorate_all_tests(LargeObjectTruncateTests,
skip_if_no_lo, skip_lo_if_green, skip_if_no_truncate)
def _has_lo64(conn):
"""Return (bool, msg) about the lo64 support"""
if conn.info.server_version < 90300:
@ -472,6 +463,7 @@ def _has_lo64(conn):
return True, "this server and build support the lo64 API"
@decorate_all_tests
def skip_if_no_lo64(f):
@wraps(f)
def skip_if_no_lo64_(self):
@ -484,6 +476,10 @@ def skip_if_no_lo64(f):
return skip_if_no_lo64_
@skip_if_no_lo
@skip_lo_if_green
@skip_if_no_truncate
@skip_if_no_lo64
class LargeObject64Tests(LargeObjectTestCase):
def test_seek_tell_truncate_greater_than_2gb(self):
lo = self.conn.lobject()
@ -495,10 +491,7 @@ class LargeObject64Tests(LargeObjectTestCase):
self.assertEqual(lo.tell(), length)
decorate_all_tests(LargeObject64Tests,
skip_if_no_lo, skip_lo_if_green, skip_if_no_truncate, skip_if_no_lo64)
@decorate_all_tests
def skip_if_lo64(f):
@wraps(f)
def skip_if_lo64_(self):
@ -511,6 +504,10 @@ def skip_if_lo64(f):
return skip_if_lo64_
@skip_if_no_lo
@skip_lo_if_green
@skip_if_no_truncate
@skip_if_lo64
class LargeObjectNot64Tests(LargeObjectTestCase):
def test_seek_larger_than_2gb(self):
lo = self.conn.lobject()
@ -527,10 +524,6 @@ class LargeObjectNot64Tests(LargeObjectTestCase):
lo.truncate, length)
decorate_all_tests(LargeObjectNot64Tests,
skip_if_no_lo, skip_lo_if_green, skip_if_no_truncate, skip_if_lo64)
def test_suite():
return unittest.TestLoader().loadTestsFromName(__name__)

View File

@ -26,7 +26,6 @@ from . import dbapi20
from . import dbapi20_tpc
from .testutils import skip_if_tpc_disabled
import unittest
from .testutils import decorate_all_tests
import psycopg2
from .testconfig import dsn
@ -70,6 +69,7 @@ class Psycopg2Tests(dbapi20.DatabaseAPI20Test):
pass
@skip_if_tpc_disabled
class Psycopg2TPCTests(dbapi20_tpc.TwoPhaseCommitTests, unittest.TestCase):
driver = psycopg2
@ -77,9 +77,6 @@ class Psycopg2TPCTests(dbapi20_tpc.TwoPhaseCommitTests, unittest.TestCase):
return psycopg2.connect(dsn=dsn)
decorate_all_tests(Psycopg2TPCTests, skip_if_tpc_disabled)
def test_suite():
return unittest.TestLoader().loadTestsFromName(__name__)

View File

@ -456,6 +456,20 @@ class AdaptSubclassTest(unittest.TestCase):
self.assertEqual(ext.adapt(foo((1, 2, 3))).getquoted(), 'bar')
@decorate_all_tests
def skip_if_cant_cast(f):
@wraps(f)
def skip_if_cant_cast_(self, *args, **kwargs):
if self._cast is None:
return self.skipTest("can't test bytea parser: %s - %s"
% (self._exc.__class__.__name__, self._exc))
return f(self, *args, **kwargs)
return skip_if_cant_cast_
@skip_if_cant_cast
class ByteaParserTest(unittest.TestCase):
"""Unit test for our bytea format parser."""
def setUp(self):
@ -543,21 +557,6 @@ class ByteaParserTest(unittest.TestCase):
self.assertEqual(rv, tgt)
def skip_if_cant_cast(f):
@wraps(f)
def skip_if_cant_cast_(self, *args, **kwargs):
if self._cast is None:
return self.skipTest("can't test bytea parser: %s - %s"
% (self._exc.__class__.__name__, self._exc))
return f(self, *args, **kwargs)
return skip_if_cant_cast_
decorate_all_tests(ByteaParserTest, skip_if_cant_cast)
def test_suite():
return unittest.TestLoader().loadTestsFromName(__name__)

View File

@ -24,8 +24,7 @@ from pickle import dumps, loads
import unittest
from .testutils import (skip_if_no_uuid, skip_before_postgres,
ConnectingTestCase, decorate_all_tests, py3_raises_typeerror, slow,
skip_from_python)
ConnectingTestCase, py3_raises_typeerror, slow, skip_from_python)
import psycopg2
import psycopg2.extras
@ -1049,6 +1048,7 @@ def skip_if_no_jsonb_type(f):
return skip_before_postgres(9, 4)(f)
@skip_if_no_jsonb_type
class JsonbTestCase(ConnectingTestCase):
@staticmethod
def myloads(s):
@ -1139,9 +1139,6 @@ class JsonbTestCase(ConnectingTestCase):
self.assertEqual(curs.fetchone()[0], None)
decorate_all_tests(JsonbTestCase, skip_if_no_jsonb_type)
class RangeTestCase(unittest.TestCase):
def test_noparam(self):
from psycopg2.extras import Range
@ -1435,19 +1432,7 @@ class RangeTestCase(unittest.TestCase):
self.assertEqual(result, expected)
def skip_if_no_range(f):
@wraps(f)
def skip_if_no_range_(self):
if self.conn.info.server_version < 90200:
return self.skipTest(
"server version %s doesn't support range types"
% self.conn.info.server_version)
return f(self)
return skip_if_no_range_
@skip_before_postgres(9, 2, "range not supported before postgres 9.2")
class RangeCasterTestCase(ConnectingTestCase):
builtin_ranges = ('int4range', 'int8range', 'numrange',
@ -1766,9 +1751,6 @@ class RangeCasterTestCase(ConnectingTestCase):
del ext.adapters[r.range, ext.ISQLQuote]
decorate_all_tests(RangeCasterTestCase, skip_if_no_range)
def test_suite():
return unittest.TestLoader().loadTestsFromName(__name__)

View File

@ -25,6 +25,7 @@
import re
import os
import sys
import types
import select
import platform
import unittest
@ -174,16 +175,31 @@ class ConnectingTestCase(unittest.TestCase):
raise Exception("Unexpected result from poll: %r", state)
def decorate_all_tests(cls, *decorators):
def decorate_all_tests(obj, *decorators):
"""
Apply all the *decorators* to all the tests defined in the TestCase *cls*.
Apply all the *decorators* to all the tests defined in the TestCase *obj*.
The decorator can also be applied to a decorator: if *obj* is a function,
return a new decorator which can be applied either to a method or to a
class, in which case it will decorate all the tests.
"""
for n in dir(cls):
if isinstance(obj, types.FunctionType):
def decorator(func_or_cls):
if isinstance(func_or_cls, types.FunctionType):
return obj(func_or_cls)
else:
decorate_all_tests(func_or_cls, obj)
return func_or_cls
return decorator
for n in dir(obj):
if n.startswith('test'):
for d in decorators:
setattr(cls, n, d(getattr(cls, n)))
setattr(obj, n, d(getattr(obj, n)))
@decorate_all_tests
def skip_if_no_uuid(f):
"""Decorator to skip a test if uuid is not supported by Py/PG."""
@wraps(f)
@ -203,6 +219,7 @@ def skip_if_no_uuid(f):
return skip_if_no_uuid_
@decorate_all_tests
def skip_if_tpc_disabled(f):
"""Skip a test if the server has tpc support disabled."""
@wraps(f)
@ -230,13 +247,19 @@ def skip_if_tpc_disabled(f):
def skip_before_postgres(*ver):
"""Skip a test on PostgreSQL before a certain version."""
reason = None
if isinstance(ver[-1], str):
ver, reason = ver[:-1], ver[-1]
ver = ver + (0,) * (3 - len(ver))
@decorate_all_tests
def skip_before_postgres_(f):
@wraps(f)
def skip_before_postgres__(self):
if self.conn.info.server_version < int("%d%02d%02d" % ver):
return self.skipTest("skipped because PostgreSQL %s"
return self.skipTest(
reason or "skipped because PostgreSQL %s"
% self.conn.info.server_version)
else:
return f(self)
@ -249,6 +272,7 @@ def skip_after_postgres(*ver):
"""Skip a test on PostgreSQL after (including) a certain version."""
ver = ver + (0,) * (3 - len(ver))
@decorate_all_tests
def skip_after_postgres_(f):
@wraps(f)
def skip_after_postgres__(self):
@ -274,6 +298,7 @@ def skip_before_libpq(*ver):
"""Skip a test if libpq we're linked to is older than a certain version."""
ver = ver + (0,) * (3 - len(ver))
@decorate_all_tests
def skip_before_libpq_(f):
@wraps(f)
def skip_before_libpq__(self):
@ -291,6 +316,7 @@ def skip_after_libpq(*ver):
"""Skip a test if libpq we're linked to is newer than a certain version."""
ver = ver + (0,) * (3 - len(ver))
@decorate_all_tests
def skip_after_libpq_(f):
@wraps(f)
def skip_after_libpq__(self):
@ -306,6 +332,7 @@ def skip_after_libpq(*ver):
def skip_before_python(*ver):
"""Skip a test on Python before a certain version."""
@decorate_all_tests
def skip_before_python_(f):
@wraps(f)
def skip_before_python__(self):
@ -321,6 +348,7 @@ def skip_before_python(*ver):
def skip_from_python(*ver):
"""Skip a test on Python after (including) a certain version."""
@decorate_all_tests
def skip_from_python_(f):
@wraps(f)
def skip_from_python__(self):
@ -334,6 +362,7 @@ def skip_from_python(*ver):
return skip_from_python_
@decorate_all_tests
def skip_if_no_superuser(f):
"""Skip a test if the database user running the test is not a superuser"""
@wraps(f)
@ -352,6 +381,7 @@ def skip_if_no_superuser(f):
def skip_if_green(reason):
@decorate_all_tests
def skip_if_green_(f):
@wraps(f)
def skip_if_green__(self):
@ -368,6 +398,7 @@ def skip_if_green(reason):
skip_copy_if_green = skip_if_green("copy in async mode currently not supported")
@decorate_all_tests
def skip_if_no_getrefcount(f):
@wraps(f)
def skip_if_no_getrefcount_(self):
@ -378,6 +409,7 @@ def skip_if_no_getrefcount(f):
return skip_if_no_getrefcount_
@decorate_all_tests
def skip_if_windows(f):
"""Skip a test if run on windows"""
@wraps(f)