mirror of
https://github.com/psycopg/psycopg2.git
synced 2025-02-07 12:50:32 +03:00
Move imports to the top of the module across tests
Allows removing many duplicate imports and better follows PEP8 guidelines: https://www.python.org/dev/peps/pep-0008/#imports > Imports are always put at the top of the file, just after any module > comments and docstrings, and before module globals and constants.
This commit is contained in:
Jon Dufresne
Daniele Varrazzo
parent
194efc4375
commit
d90ad8627d
|
|
@ -24,6 +24,7 @@
|
|||
# License for more details.
|
||||
|
||||
import time
|
||||
from select import select
|
||||
|
||||
import psycopg2
|
||||
from psycopg2 import extras
|
||||
|
|
@ -205,7 +206,6 @@ class AsyncReplicationTest(ReplicationTestCase):
|
|||
self.assertRaises(psycopg2.ProgrammingError, cur.consume_stream, consume)
|
||||
|
||||
def process_stream():
|
||||
from select import select
|
||||
while True:
|
||||
msg = cur.read_message()
|
||||
if msg:
|
||||
|
|
|
|||
|
|
@ -22,18 +22,24 @@
|
|||
# FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public
|
||||
# License for more details.
|
||||
|
||||
import re
|
||||
import os
|
||||
import sys
|
||||
import time
|
||||
import ctypes
|
||||
import threading
|
||||
import gc
|
||||
import os
|
||||
import re
|
||||
import subprocess as sp
|
||||
import sys
|
||||
import threading
|
||||
import time
|
||||
from collections import deque
|
||||
from operator import attrgetter
|
||||
from weakref import ref
|
||||
|
||||
import psycopg2
|
||||
import psycopg2.errorcodes
|
||||
from psycopg2 import extensions as ext
|
||||
from psycopg2 import ProgrammingError
|
||||
from psycopg2.extensions import Xid
|
||||
from psycopg2.extras import RealDictConnection
|
||||
|
||||
from .testutils import (
|
||||
unittest, skip_if_no_superuser, skip_before_postgres,
|
||||
|
|
@ -149,8 +155,6 @@ class ConnectionTests(ConnectingTestCase):
|
|||
|
||||
@slow
|
||||
def test_notices_deque(self):
|
||||
from collections import deque
|
||||
|
||||
conn = self.conn
|
||||
self.conn.notices = deque()
|
||||
cur = conn.cursor()
|
||||
|
|
@ -255,8 +259,6 @@ class ConnectionTests(ConnectingTestCase):
|
|||
conn.close()
|
||||
|
||||
def test_weakref(self):
|
||||
from weakref import ref
|
||||
import gc
|
||||
conn = psycopg2.connect(dsn)
|
||||
w = ref(conn)
|
||||
conn.close()
|
||||
|
|
@ -362,8 +364,6 @@ class ConnectionTests(ConnectingTestCase):
|
|||
|
||||
class ParseDsnTestCase(ConnectingTestCase):
|
||||
def test_parse_dsn(self):
|
||||
from psycopg2 import ProgrammingError
|
||||
|
||||
self.assertEqual(
|
||||
ext.parse_dsn('dbname=test user=tester password=secret'),
|
||||
dict(user='tester', password='secret', dbname='test'),
|
||||
|
|
@ -1124,16 +1124,12 @@ class ConnectionTwoPhaseTests(ConnectingTestCase):
|
|||
cnn.tpc_rollback(xid)
|
||||
|
||||
def test_xid_construction(self):
|
||||
from psycopg2.extensions import Xid
|
||||
|
||||
x1 = Xid(74, 'foo', 'bar')
|
||||
self.assertEqual(74, x1.format_id)
|
||||
self.assertEqual('foo', x1.gtrid)
|
||||
self.assertEqual('bar', x1.bqual)
|
||||
|
||||
def test_xid_from_string(self):
|
||||
from psycopg2.extensions import Xid
|
||||
|
||||
x2 = Xid.from_string('42_Z3RyaWQ=_YnF1YWw=')
|
||||
self.assertEqual(42, x2.format_id)
|
||||
self.assertEqual('gtrid', x2.gtrid)
|
||||
|
|
@ -1145,8 +1141,6 @@ class ConnectionTwoPhaseTests(ConnectingTestCase):
|
|||
self.assertEqual(None, x3.bqual)
|
||||
|
||||
def test_xid_to_string(self):
|
||||
from psycopg2.extensions import Xid
|
||||
|
||||
x1 = Xid.from_string('42_Z3RyaWQ=_YnF1YWw=')
|
||||
self.assertEqual(str(x1), '42_Z3RyaWQ=_YnF1YWw=')
|
||||
|
||||
|
|
@ -1186,7 +1180,6 @@ class ConnectionTwoPhaseTests(ConnectingTestCase):
|
|||
self.assertRaises(psycopg2.ProgrammingError, cnn.cancel)
|
||||
|
||||
def test_tpc_recover_non_dbapi_connection(self):
|
||||
from psycopg2.extras import RealDictConnection
|
||||
cnn = self.connect(connection_factory=RealDictConnection)
|
||||
cnn.tpc_begin('dict-connection')
|
||||
cnn.tpc_prepare()
|
||||
|
|
|
|||
|
|
@ -28,6 +28,9 @@ import pickle
|
|||
import psycopg2
|
||||
import psycopg2.extensions
|
||||
import unittest
|
||||
from datetime import date
|
||||
from decimal import Decimal
|
||||
from weakref import ref
|
||||
from .testutils import (ConnectingTestCase, skip_before_postgres,
|
||||
skip_if_no_getrefcount, slow, skip_if_no_superuser,
|
||||
skip_if_windows)
|
||||
|
|
@ -101,8 +104,6 @@ class CursorTests(ConnectingTestCase):
|
|||
cur.mogrify(u"SELECT %s;", (snowman,)))
|
||||
|
||||
def test_mogrify_decimal_explodes(self):
|
||||
from decimal import Decimal
|
||||
|
||||
conn = self.conn
|
||||
cur = conn.cursor()
|
||||
self.assertEqual(b'SELECT 10.3;',
|
||||
|
|
@ -143,10 +144,8 @@ class CursorTests(ConnectingTestCase):
|
|||
self.assertEqual(42, curs.cast(20, '42'))
|
||||
self.assertAlmostEqual(3.14, curs.cast(700, '3.14'))
|
||||
|
||||
from decimal import Decimal
|
||||
self.assertEqual(Decimal('123.45'), curs.cast(1700, '123.45'))
|
||||
|
||||
from datetime import date
|
||||
self.assertEqual(date(2011, 1, 2), curs.cast(1082, '2011-01-02'))
|
||||
self.assertEqual("who am i?", curs.cast(705, 'who am i?')) # unknown
|
||||
|
||||
|
|
@ -166,7 +165,6 @@ class CursorTests(ConnectingTestCase):
|
|||
self.assertEqual("foofoo", curs2.cast(705, 'foo'))
|
||||
|
||||
def test_weakref(self):
|
||||
from weakref import ref
|
||||
curs = self.conn.cursor()
|
||||
w = ref(curs)
|
||||
del curs
|
||||
|
|
|
|||
|
|
@ -23,7 +23,10 @@
|
|||
# License for more details.
|
||||
|
||||
import math
|
||||
from datetime import date, datetime, time, timedelta
|
||||
|
||||
import psycopg2
|
||||
import psycopg2.tz
|
||||
from psycopg2.tz import FixedOffsetTimezone, ZERO
|
||||
import unittest
|
||||
from .testutils import ConnectingTestCase, skip_before_postgres
|
||||
|
|
@ -130,7 +133,6 @@ class DatetimeTests(ConnectingTestCase, CommonDatetimeTestsMixin):
|
|||
self.assertEqual(value.microsecond, 123456)
|
||||
|
||||
def check_time_tz(self, str_offset, offset):
|
||||
from datetime import time, timedelta
|
||||
base = time(13, 30, 29)
|
||||
base_str = '13:30:29'
|
||||
|
||||
|
|
@ -161,7 +163,6 @@ class DatetimeTests(ConnectingTestCase, CommonDatetimeTestsMixin):
|
|||
self.check_time_tz("-01:15:59", -60 * (60 + 16))
|
||||
|
||||
def check_datetime_tz(self, str_offset, offset):
|
||||
from datetime import datetime, timedelta
|
||||
base = datetime(2007, 1, 1, 13, 30, 29)
|
||||
base_str = '2007-01-01 13:30:29'
|
||||
|
||||
|
|
@ -229,25 +230,21 @@ class DatetimeTests(ConnectingTestCase, CommonDatetimeTestsMixin):
|
|||
self.assertEqual(str(value), '9999-12-31')
|
||||
|
||||
def test_adapt_date(self):
|
||||
from datetime import date
|
||||
value = self.execute('select (%s)::date::text',
|
||||
[date(2007, 1, 1)])
|
||||
self.assertEqual(value, '2007-01-01')
|
||||
|
||||
def test_adapt_time(self):
|
||||
from datetime import time
|
||||
value = self.execute('select (%s)::time::text',
|
||||
[time(13, 30, 29)])
|
||||
self.assertEqual(value, '13:30:29')
|
||||
|
||||
def test_adapt_datetime(self):
|
||||
from datetime import datetime
|
||||
value = self.execute('select (%s)::timestamp::text',
|
||||
[datetime(2007, 1, 1, 13, 30, 29)])
|
||||
self.assertEqual(value, '2007-01-01 13:30:29')
|
||||
|
||||
def test_adapt_timedelta(self):
|
||||
from datetime import timedelta
|
||||
value = self.execute('select extract(epoch from (%s)::interval)',
|
||||
[timedelta(days=42, seconds=45296,
|
||||
microseconds=123456)])
|
||||
|
|
@ -256,7 +253,6 @@ class DatetimeTests(ConnectingTestCase, CommonDatetimeTestsMixin):
|
|||
self.assertEqual(int(round((value - seconds) * 1000000)), 123456)
|
||||
|
||||
def test_adapt_negative_timedelta(self):
|
||||
from datetime import timedelta
|
||||
value = self.execute('select extract(epoch from (%s)::interval)',
|
||||
[timedelta(days=-42, seconds=45296,
|
||||
microseconds=123456)])
|
||||
|
|
@ -275,17 +271,13 @@ class DatetimeTests(ConnectingTestCase, CommonDatetimeTestsMixin):
|
|||
self.assertEqual(type(o1[0]), type(o2[0]))
|
||||
|
||||
def test_type_roundtrip_date(self):
|
||||
from datetime import date
|
||||
self._test_type_roundtrip(date(2010, 5, 3))
|
||||
|
||||
def test_type_roundtrip_datetime(self):
|
||||
from datetime import datetime
|
||||
dt = self._test_type_roundtrip(datetime(2010, 5, 3, 10, 20, 30))
|
||||
self.assertEqual(None, dt.tzinfo)
|
||||
|
||||
def test_type_roundtrip_datetimetz(self):
|
||||
from datetime import datetime
|
||||
import psycopg2.tz
|
||||
tz = psycopg2.tz.FixedOffsetTimezone(8 * 60)
|
||||
dt1 = datetime(2010, 5, 3, 10, 20, 30, tzinfo=tz)
|
||||
dt2 = self._test_type_roundtrip(dt1)
|
||||
|
|
@ -293,13 +285,10 @@ class DatetimeTests(ConnectingTestCase, CommonDatetimeTestsMixin):
|
|||
self.assertEqual(dt1, dt2)
|
||||
|
||||
def test_type_roundtrip_time(self):
|
||||
from datetime import time
|
||||
tm = self._test_type_roundtrip(time(10, 20, 30))
|
||||
self.assertEqual(None, tm.tzinfo)
|
||||
|
||||
def test_type_roundtrip_timetz(self):
|
||||
from datetime import time
|
||||
import psycopg2.tz
|
||||
tz = psycopg2.tz.FixedOffsetTimezone(8 * 60)
|
||||
tm1 = time(10, 20, 30, tzinfo=tz)
|
||||
tm2 = self._test_type_roundtrip(tm1)
|
||||
|
|
@ -307,34 +296,26 @@ class DatetimeTests(ConnectingTestCase, CommonDatetimeTestsMixin):
|
|||
self.assertEqual(tm1, tm2)
|
||||
|
||||
def test_type_roundtrip_interval(self):
|
||||
from datetime import timedelta
|
||||
self._test_type_roundtrip(timedelta(seconds=30))
|
||||
|
||||
def test_type_roundtrip_date_array(self):
|
||||
from datetime import date
|
||||
self._test_type_roundtrip_array(date(2010, 5, 3))
|
||||
|
||||
def test_type_roundtrip_datetime_array(self):
|
||||
from datetime import datetime
|
||||
self._test_type_roundtrip_array(datetime(2010, 5, 3, 10, 20, 30))
|
||||
|
||||
def test_type_roundtrip_datetimetz_array(self):
|
||||
from datetime import datetime
|
||||
self._test_type_roundtrip_array(
|
||||
datetime(2010, 5, 3, 10, 20, 30, tzinfo=FixedOffsetTimezone(0)))
|
||||
|
||||
def test_type_roundtrip_time_array(self):
|
||||
from datetime import time
|
||||
self._test_type_roundtrip_array(time(10, 20, 30))
|
||||
|
||||
def test_type_roundtrip_interval_array(self):
|
||||
from datetime import timedelta
|
||||
self._test_type_roundtrip_array(timedelta(seconds=30))
|
||||
|
||||
@skip_before_postgres(8, 1)
|
||||
def test_time_24(self):
|
||||
from datetime import time
|
||||
|
||||
t = self.execute("select '24:00'::time;")
|
||||
self.assertEqual(t, time(0, 0))
|
||||
|
||||
|
|
@ -399,8 +380,6 @@ class DatetimeTests(ConnectingTestCase, CommonDatetimeTestsMixin):
|
|||
self.assertRaises(OverflowError, f, '00:00:00.100000000000000000')
|
||||
|
||||
def test_adapt_infinity_tz(self):
|
||||
from datetime import datetime
|
||||
|
||||
t = self.execute("select 'infinity'::timestamp")
|
||||
self.assert_(t.tzinfo is None)
|
||||
self.assert_(t > datetime(4000, 1, 1))
|
||||
|
|
@ -425,7 +404,6 @@ class DatetimeTests(ConnectingTestCase, CommonDatetimeTestsMixin):
|
|||
psycopg2.STRING.values, 'WAT', psycopg2.extensions.INTERVAL),
|
||||
cur)
|
||||
|
||||
from datetime import timedelta
|
||||
for s, v in [
|
||||
('0', timedelta(0)),
|
||||
('1', timedelta(microseconds=1)),
|
||||
|
|
@ -629,20 +607,17 @@ class FromTicksTestCase(unittest.TestCase):
|
|||
# bug "TimestampFromTicks() throws ValueError (2-2.0.14)"
|
||||
# reported by Jozsef Szalay on 2010-05-06
|
||||
def test_timestamp_value_error_sec_59_99(self):
|
||||
from datetime import datetime
|
||||
s = psycopg2.TimestampFromTicks(1273173119.99992)
|
||||
self.assertEqual(s.adapted,
|
||||
datetime(2010, 5, 6, 14, 11, 59, 999920,
|
||||
tzinfo=FixedOffsetTimezone(-5 * 60)))
|
||||
|
||||
def test_date_value_error_sec_59_99(self):
|
||||
from datetime import date
|
||||
s = psycopg2.DateFromTicks(1273173119.99992)
|
||||
# The returned date is local
|
||||
self.assert_(s.adapted in [date(2010, 5, 6), date(2010, 5, 7)])
|
||||
|
||||
def test_time_value_error_sec_59_99(self):
|
||||
from datetime import time
|
||||
s = psycopg2.TimeFromTicks(1273173119.99992)
|
||||
self.assertEqual(s.adapted.replace(hour=0),
|
||||
time(0, 11, 59, 999920))
|
||||
|
|
|
|||
|
|
@ -26,6 +26,9 @@ import unittest
|
|||
from .testutils import ConnectingTestCase
|
||||
|
||||
import psycopg2
|
||||
from psycopg2 import errors
|
||||
from psycopg2._psycopg import sqlstate_errors
|
||||
from psycopg2.errors import UndefinedTable
|
||||
|
||||
|
||||
class ErrorsTests(ConnectingTestCase):
|
||||
|
|
@ -36,14 +39,12 @@ class ErrorsTests(ConnectingTestCase):
|
|||
except psycopg2.Error as exc:
|
||||
e = exc
|
||||
|
||||
from psycopg2.errors import UndefinedTable
|
||||
self.assert_(isinstance(e, UndefinedTable), type(e))
|
||||
self.assert_(isinstance(e, self.conn.ProgrammingError))
|
||||
|
||||
def test_exception_class_fallback(self):
|
||||
cur = self.conn.cursor()
|
||||
|
||||
from psycopg2._psycopg import sqlstate_errors
|
||||
x = sqlstate_errors.pop('42P01')
|
||||
try:
|
||||
cur.execute("select * from nonexist")
|
||||
|
|
@ -55,17 +56,12 @@ class ErrorsTests(ConnectingTestCase):
|
|||
self.assertEqual(type(e), self.conn.ProgrammingError)
|
||||
|
||||
def test_lookup(self):
|
||||
from psycopg2 import errors
|
||||
|
||||
self.assertIs(errors.lookup('42P01'), errors.UndefinedTable)
|
||||
|
||||
with self.assertRaises(KeyError):
|
||||
errors.lookup('XXXXX')
|
||||
|
||||
def test_has_base_exceptions(self):
|
||||
import psycopg2
|
||||
from psycopg2 import errors
|
||||
|
||||
excs = []
|
||||
for n in dir(psycopg2):
|
||||
obj = getattr(psycopg2, n)
|
||||
|
|
|
|||
|
|
@ -16,10 +16,14 @@
|
|||
|
||||
import time
|
||||
import pickle
|
||||
from datetime import timedelta
|
||||
import psycopg2
|
||||
import psycopg2.extras
|
||||
import unittest
|
||||
from datetime import timedelta
|
||||
|
||||
import psycopg2
|
||||
from psycopg2.compat import lru_cache
|
||||
import psycopg2.extras
|
||||
from psycopg2.extras import NamedTupleConnection, NamedTupleCursor
|
||||
|
||||
from .testutils import ConnectingTestCase, skip_before_postgres, \
|
||||
skip_before_python, skip_from_python
|
||||
|
||||
|
|
@ -358,7 +362,6 @@ class ExtrasDictCursorRealTests(_DictCursorBase):
|
|||
class NamedTupleCursorTest(ConnectingTestCase):
|
||||
def setUp(self):
|
||||
ConnectingTestCase.setUp(self)
|
||||
from psycopg2.extras import NamedTupleConnection
|
||||
|
||||
self.conn = self.connect(connection_factory=NamedTupleConnection)
|
||||
curs = self.conn.cursor()
|
||||
|
|
@ -495,7 +498,6 @@ class NamedTupleCursorTest(ConnectingTestCase):
|
|||
|
||||
def test_minimal_generation(self):
|
||||
# Instrument the class to verify it gets called the minimum number of times.
|
||||
from psycopg2.extras import NamedTupleCursor
|
||||
f_orig = NamedTupleCursor._make_nt
|
||||
calls = [0]
|
||||
|
||||
|
|
@ -591,9 +593,6 @@ class NamedTupleCursorTest(ConnectingTestCase):
|
|||
self.assert_(type(r1) is not type(r2))
|
||||
|
||||
def test_max_cache(self):
|
||||
from psycopg2.extras import NamedTupleCursor
|
||||
from psycopg2.compat import lru_cache
|
||||
|
||||
old_func = NamedTupleCursor._cached_make_nt
|
||||
NamedTupleCursor._cached_make_nt = \
|
||||
lru_cache(8)(NamedTupleCursor._do_make_nt)
|
||||
|
|
|
|||
|
|
@ -22,6 +22,7 @@ import unittest
|
|||
import psycopg2
|
||||
import psycopg2.extras
|
||||
import psycopg2.extensions as ext
|
||||
from psycopg2 import sql
|
||||
|
||||
|
||||
class TestPaginate(unittest.TestCase):
|
||||
|
|
@ -84,7 +85,6 @@ class TestExecuteBatch(FastExecuteTestMixin, testutils.ConnectingTestCase):
|
|||
self.assertEqual(cur.fetchall(), [(i, i * 10) for i in range(1000)])
|
||||
|
||||
def test_composed(self):
|
||||
from psycopg2 import sql
|
||||
cur = self.conn.cursor()
|
||||
psycopg2.extras.execute_batch(cur,
|
||||
sql.SQL("insert into {0} (id, val) values (%s, %s)")
|
||||
|
|
@ -181,7 +181,6 @@ class TestExecuteValues(FastExecuteTestMixin, testutils.ConnectingTestCase):
|
|||
self.assertEqual(cur.fetchall(), [(i, i * 10) for i in range(1000)])
|
||||
|
||||
def test_composed(self):
|
||||
from psycopg2 import sql
|
||||
cur = self.conn.cursor()
|
||||
psycopg2.extras.execute_values(cur,
|
||||
sql.SQL("insert into {0} (id, val) values %s")
|
||||
|
|
|
|||
|
|
@ -22,10 +22,12 @@
|
|||
# FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public
|
||||
# License for more details.
|
||||
|
||||
import select
|
||||
import unittest
|
||||
import psycopg2
|
||||
import psycopg2.extensions
|
||||
import psycopg2.extras
|
||||
from psycopg2.extensions import POLL_OK, POLL_READ, POLL_WRITE
|
||||
|
||||
from .testutils import ConnectingTestCase, skip_before_postgres, slow
|
||||
|
||||
|
|
@ -131,9 +133,6 @@ class CallbackErrorTestCase(ConnectingTestCase):
|
|||
|
||||
def crappy_callback(self, conn):
|
||||
"""green callback failing after `self.to_error` time it is called"""
|
||||
import select
|
||||
from psycopg2.extensions import POLL_OK, POLL_READ, POLL_WRITE
|
||||
|
||||
while True:
|
||||
if self.to_error is not None:
|
||||
self.to_error -= 1
|
||||
|
|
|
|||
|
|
@ -22,9 +22,11 @@
|
|||
# FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public
|
||||
# License for more details.
|
||||
|
||||
import gc
|
||||
import os
|
||||
import sys
|
||||
from subprocess import Popen
|
||||
from weakref import ref
|
||||
|
||||
import unittest
|
||||
from .testutils import (skip_before_postgres,
|
||||
|
|
@ -190,9 +192,6 @@ class ExceptionsTestCase(ConnectingTestCase):
|
|||
self.assertEqual(e.diag.severity, 'ERROR')
|
||||
|
||||
def test_diagnostics_life(self):
|
||||
import gc
|
||||
from weakref import ref
|
||||
|
||||
def tmp():
|
||||
cur = self.conn.cursor()
|
||||
try:
|
||||
|
|
|
|||
|
|
@ -23,9 +23,11 @@
|
|||
# License for more details.
|
||||
|
||||
import unittest
|
||||
from collections import deque
|
||||
|
||||
import psycopg2
|
||||
from psycopg2 import extensions
|
||||
from psycopg2.extensions import Notify
|
||||
from .testutils import ConnectingTestCase, slow
|
||||
from .testconfig import dsn
|
||||
|
||||
|
|
@ -163,7 +165,6 @@ conn.close()
|
|||
|
||||
@slow
|
||||
def test_notify_deque(self):
|
||||
from collections import deque
|
||||
self.autocommit(self.conn)
|
||||
self.conn.notifies = deque()
|
||||
self.listen('foo')
|
||||
|
|
@ -209,14 +210,12 @@ conn.close()
|
|||
self.assertEqual((n1 != n2), (d1 != d2))
|
||||
|
||||
def test_compare_tuple(self):
|
||||
from psycopg2.extensions import Notify
|
||||
self.assertEqual((10, 'foo'), Notify(10, 'foo'))
|
||||
self.assertEqual((10, 'foo'), Notify(10, 'foo', 'bar'))
|
||||
self.assertNotEqual((10, 'foo'), Notify(20, 'foo'))
|
||||
self.assertNotEqual((10, 'foo'), Notify(10, 'bar'))
|
||||
|
||||
def test_hash(self):
|
||||
from psycopg2.extensions import Notify
|
||||
self.assertEqual(hash((10, 'foo')), hash(Notify(10, 'foo')))
|
||||
self.assertNotEqual(hash(Notify(10, 'foo', 'bar')),
|
||||
hash(Notify(10, 'foo')))
|
||||
|
|
|
|||
|
|
@ -29,6 +29,7 @@ from .testutils import ConnectingTestCase, unichr
|
|||
|
||||
import psycopg2
|
||||
import psycopg2.extensions
|
||||
from psycopg2.extensions import adapt, quote_ident
|
||||
|
||||
|
||||
class QuotingTestCase(ConnectingTestCase):
|
||||
|
|
@ -194,13 +195,11 @@ class TestQuotedString(ConnectingTestCase):
|
|||
|
||||
class TestQuotedIdentifier(ConnectingTestCase):
|
||||
def test_identifier(self):
|
||||
from psycopg2.extensions import quote_ident
|
||||
self.assertEqual(quote_ident('blah-blah', self.conn), '"blah-blah"')
|
||||
self.assertEqual(quote_ident('quote"inside', self.conn), '"quote""inside"')
|
||||
|
||||
@testutils.skip_before_postgres(8, 0)
|
||||
def test_unicode_ident(self):
|
||||
from psycopg2.extensions import quote_ident
|
||||
snowman = u"\u2603"
|
||||
quoted = '"' + snowman + '"'
|
||||
if sys.version_info[0] < 3:
|
||||
|
|
@ -211,7 +210,6 @@ class TestQuotedIdentifier(ConnectingTestCase):
|
|||
|
||||
class TestStringAdapter(ConnectingTestCase):
|
||||
def test_encoding_default(self):
|
||||
from psycopg2.extensions import adapt
|
||||
a = adapt("hello")
|
||||
self.assertEqual(a.encoding, 'latin1')
|
||||
self.assertEqual(a.getquoted(), b"'hello'")
|
||||
|
|
@ -223,7 +221,6 @@ class TestStringAdapter(ConnectingTestCase):
|
|||
# self.assertEqual(adapt(egrave).getquoted(), "'\xe8'")
|
||||
|
||||
def test_encoding_error(self):
|
||||
from psycopg2.extensions import adapt
|
||||
snowman = u"\u2603"
|
||||
a = adapt(snowman)
|
||||
self.assertRaises(UnicodeEncodeError, a.getquoted)
|
||||
|
|
@ -232,7 +229,6 @@ class TestStringAdapter(ConnectingTestCase):
|
|||
# Note: this works-ish mostly in case when the standard db connection
|
||||
# we test with is utf8, otherwise the encoding chosen by PQescapeString
|
||||
# may give bad results.
|
||||
from psycopg2.extensions import adapt
|
||||
snowman = u"\u2603"
|
||||
a = adapt(snowman)
|
||||
a.encoding = 'utf8'
|
||||
|
|
@ -240,7 +236,6 @@ class TestStringAdapter(ConnectingTestCase):
|
|||
self.assertEqual(a.getquoted(), b"'\xe2\x98\x83'")
|
||||
|
||||
def test_connection_wins_anyway(self):
|
||||
from psycopg2.extensions import adapt
|
||||
snowman = u"\u2603"
|
||||
a = adapt(snowman)
|
||||
a.encoding = 'latin9'
|
||||
|
|
|
|||
|
|
@ -22,7 +22,10 @@
|
|||
# FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public
|
||||
# License for more details.
|
||||
|
||||
from select import select
|
||||
|
||||
import psycopg2
|
||||
from psycopg2 import sql
|
||||
from psycopg2.extras import (
|
||||
PhysicalReplicationConnection, LogicalReplicationConnection, StopReplication)
|
||||
|
||||
|
|
@ -141,7 +144,6 @@ class ReplicationTest(ReplicationTestCase):
|
|||
@skip_before_postgres(9, 4) # slots require 9.4
|
||||
@skip_repl_if_green
|
||||
def test_start_replication_expert_sql(self):
|
||||
from psycopg2 import sql
|
||||
conn = self.repl_connect(connection_factory=LogicalReplicationConnection)
|
||||
if conn is None:
|
||||
return
|
||||
|
|
@ -252,7 +254,6 @@ class AsyncReplicationTest(ReplicationTestCase):
|
|||
self.assertRaises(psycopg2.ProgrammingError, cur.consume_stream, consume)
|
||||
|
||||
def process_stream():
|
||||
from select import select
|
||||
while True:
|
||||
msg = cur.read_message()
|
||||
if msg:
|
||||
|
|
|
|||
|
|
@ -33,6 +33,7 @@ from .testutils import ConnectingTestCase, long
|
|||
|
||||
import psycopg2
|
||||
from psycopg2.compat import text_type
|
||||
from psycopg2.extensions import AsIs, adapt, register_adapter
|
||||
|
||||
|
||||
class TypesBasicTests(ConnectingTestCase):
|
||||
|
|
@ -424,8 +425,6 @@ class TypesBasicTests(ConnectingTestCase):
|
|||
|
||||
class AdaptSubclassTest(unittest.TestCase):
|
||||
def test_adapt_subtype(self):
|
||||
from psycopg2.extensions import adapt
|
||||
|
||||
class Sub(str):
|
||||
pass
|
||||
s1 = "hel'lo"
|
||||
|
|
@ -433,8 +432,6 @@ class AdaptSubclassTest(unittest.TestCase):
|
|||
self.assertEqual(adapt(s1).getquoted(), adapt(s2).getquoted())
|
||||
|
||||
def test_adapt_most_specific(self):
|
||||
from psycopg2.extensions import adapt, register_adapter, AsIs
|
||||
|
||||
class A(object):
|
||||
pass
|
||||
|
||||
|
|
@ -454,8 +451,6 @@ class AdaptSubclassTest(unittest.TestCase):
|
|||
|
||||
@testutils.skip_from_python(3)
|
||||
def test_no_mro_no_joy(self):
|
||||
from psycopg2.extensions import adapt, register_adapter, AsIs
|
||||
|
||||
class A:
|
||||
pass
|
||||
|
||||
|
|
@ -470,8 +465,6 @@ class AdaptSubclassTest(unittest.TestCase):
|
|||
|
||||
@testutils.skip_before_python(3)
|
||||
def test_adapt_subtype_3(self):
|
||||
from psycopg2.extensions import adapt, register_adapter, AsIs
|
||||
|
||||
class A:
|
||||
pass
|
||||
|
||||
|
|
|
|||
|
|
@ -29,6 +29,13 @@ from .testutils import (skip_if_no_uuid, skip_before_postgres,
|
|||
import psycopg2
|
||||
import psycopg2.extras
|
||||
import psycopg2.extensions as ext
|
||||
from psycopg2._json import _get_json_oids
|
||||
from psycopg2.extras import (
|
||||
CompositeCaster, DateRange, DateTimeRange, DateTimeTZRange, HstoreAdapter,
|
||||
Inet, Json, NumericRange, Range, RealDictConnection, json,
|
||||
register_composite, register_hstore, register_range,
|
||||
)
|
||||
from psycopg2.tz import FixedOffsetTimezone
|
||||
|
||||
|
||||
class TypesExtrasTests(ConnectingTestCase):
|
||||
|
|
@ -94,7 +101,6 @@ class TypesExtrasTests(ConnectingTestCase):
|
|||
self.failUnless(s is None)
|
||||
|
||||
def test_inet_conform(self):
|
||||
from psycopg2.extras import Inet
|
||||
i = Inet("192.168.1.0/24")
|
||||
a = psycopg2.extensions.adapt(i)
|
||||
a.prepare(self.conn)
|
||||
|
|
@ -126,7 +132,6 @@ class TypesExtrasTests(ConnectingTestCase):
|
|||
def skip_if_no_hstore(f):
|
||||
@wraps(f)
|
||||
def skip_if_no_hstore_(self):
|
||||
from psycopg2.extras import HstoreAdapter
|
||||
oids = HstoreAdapter.get_oids(self.conn)
|
||||
if oids is None or not oids[0]:
|
||||
return self.skipTest("hstore not available in test database")
|
||||
|
|
@ -140,8 +145,6 @@ class HstoreTestCase(ConnectingTestCase):
|
|||
if self.conn.info.server_version >= 90000:
|
||||
return self.skipTest("skipping dict adaptation with PG pre-9 syntax")
|
||||
|
||||
from psycopg2.extras import HstoreAdapter
|
||||
|
||||
o = {'a': '1', 'b': "'", 'c': None}
|
||||
if self.conn.encoding == 'UTF8':
|
||||
o['d'] = u'\xe0'
|
||||
|
|
@ -166,8 +169,6 @@ class HstoreTestCase(ConnectingTestCase):
|
|||
if self.conn.info.server_version < 90000:
|
||||
return self.skipTest("skipping dict adaptation with PG 9 syntax")
|
||||
|
||||
from psycopg2.extras import HstoreAdapter
|
||||
|
||||
o = {'a': '1', 'b': "'", 'c': None}
|
||||
if self.conn.encoding == 'UTF8':
|
||||
o['d'] = u'\xe0'
|
||||
|
|
@ -197,8 +198,6 @@ class HstoreTestCase(ConnectingTestCase):
|
|||
self.assertQuotedEqual(ii[3][1], b"'" + encc + b"'")
|
||||
|
||||
def test_parse(self):
|
||||
from psycopg2.extras import HstoreAdapter
|
||||
|
||||
def ok(s, d):
|
||||
self.assertEqual(HstoreAdapter.parse(s, None), d)
|
||||
|
||||
|
|
@ -228,8 +227,6 @@ class HstoreTestCase(ConnectingTestCase):
|
|||
|
||||
@skip_if_no_hstore
|
||||
def test_register_conn(self):
|
||||
from psycopg2.extras import register_hstore
|
||||
|
||||
register_hstore(self.conn)
|
||||
cur = self.conn.cursor()
|
||||
cur.execute("select null::hstore, ''::hstore, 'a => b'::hstore")
|
||||
|
|
@ -240,8 +237,6 @@ class HstoreTestCase(ConnectingTestCase):
|
|||
|
||||
@skip_if_no_hstore
|
||||
def test_register_curs(self):
|
||||
from psycopg2.extras import register_hstore
|
||||
|
||||
cur = self.conn.cursor()
|
||||
register_hstore(cur)
|
||||
cur.execute("select null::hstore, ''::hstore, 'a => b'::hstore")
|
||||
|
|
@ -253,8 +248,6 @@ class HstoreTestCase(ConnectingTestCase):
|
|||
@skip_if_no_hstore
|
||||
@skip_from_python(3)
|
||||
def test_register_unicode(self):
|
||||
from psycopg2.extras import register_hstore
|
||||
|
||||
register_hstore(self.conn, unicode=True)
|
||||
cur = self.conn.cursor()
|
||||
cur.execute("select null::hstore, ''::hstore, 'a => b'::hstore")
|
||||
|
|
@ -267,8 +260,6 @@ class HstoreTestCase(ConnectingTestCase):
|
|||
|
||||
@skip_if_no_hstore
|
||||
def test_register_globally(self):
|
||||
from psycopg2.extras import register_hstore, HstoreAdapter
|
||||
|
||||
oids = HstoreAdapter.get_oids(self.conn)
|
||||
try:
|
||||
register_hstore(self.conn, globally=True)
|
||||
|
|
@ -291,7 +282,6 @@ class HstoreTestCase(ConnectingTestCase):
|
|||
|
||||
@skip_if_no_hstore
|
||||
def test_roundtrip(self):
|
||||
from psycopg2.extras import register_hstore
|
||||
register_hstore(self.conn)
|
||||
cur = self.conn.cursor()
|
||||
|
||||
|
|
@ -322,7 +312,6 @@ class HstoreTestCase(ConnectingTestCase):
|
|||
@skip_if_no_hstore
|
||||
@skip_from_python(3)
|
||||
def test_roundtrip_unicode(self):
|
||||
from psycopg2.extras import register_hstore
|
||||
register_hstore(self.conn, unicode=True)
|
||||
cur = self.conn.cursor()
|
||||
|
||||
|
|
@ -351,7 +340,6 @@ class HstoreTestCase(ConnectingTestCase):
|
|||
|
||||
# Note: None as conn_or_cursor is just for testing: not public
|
||||
# interface and it may break in future.
|
||||
from psycopg2.extras import register_hstore
|
||||
register_hstore(None, globally=True, oid=oid)
|
||||
try:
|
||||
cur.execute("select null::hstore, ''::hstore, 'a => b'::hstore")
|
||||
|
|
@ -366,7 +354,6 @@ class HstoreTestCase(ConnectingTestCase):
|
|||
@skip_if_no_hstore
|
||||
@skip_before_postgres(8, 3)
|
||||
def test_roundtrip_array(self):
|
||||
from psycopg2.extras import register_hstore
|
||||
register_hstore(self.conn)
|
||||
|
||||
ds = [{}, {'a': 'b', 'c': None}]
|
||||
|
|
@ -392,7 +379,6 @@ class HstoreTestCase(ConnectingTestCase):
|
|||
@skip_if_no_hstore
|
||||
@skip_before_postgres(8, 3)
|
||||
def test_array_cast(self):
|
||||
from psycopg2.extras import register_hstore
|
||||
register_hstore(self.conn)
|
||||
cur = self.conn.cursor()
|
||||
cur.execute("select array['a=>1'::hstore, 'b=>2'::hstore];")
|
||||
|
|
@ -405,7 +391,6 @@ class HstoreTestCase(ConnectingTestCase):
|
|||
cur.execute("select 'hstore'::regtype::oid, 'hstore[]'::regtype::oid")
|
||||
oid, aoid = cur.fetchone()
|
||||
|
||||
from psycopg2.extras import register_hstore
|
||||
register_hstore(None, globally=True, oid=oid, array_oid=aoid)
|
||||
try:
|
||||
cur.execute("""
|
||||
|
|
@ -423,9 +408,6 @@ class HstoreTestCase(ConnectingTestCase):
|
|||
|
||||
@skip_if_no_hstore
|
||||
def test_non_dbapi_connection(self):
|
||||
from psycopg2.extras import RealDictConnection
|
||||
from psycopg2.extras import register_hstore
|
||||
|
||||
conn = self.connect(connection_factory=RealDictConnection)
|
||||
try:
|
||||
register_hstore(conn)
|
||||
|
|
@ -492,8 +474,6 @@ class AdaptTypeTestCase(ConnectingTestCase):
|
|||
ext.register_adapter(type(None), orig_adapter)
|
||||
|
||||
def test_tokenization(self):
|
||||
from psycopg2.extras import CompositeCaster
|
||||
|
||||
def ok(s, v):
|
||||
self.assertEqual(CompositeCaster.tokenize(s), v)
|
||||
|
||||
|
|
@ -690,7 +670,6 @@ class AdaptTypeTestCase(ConnectingTestCase):
|
|||
@skip_if_no_composite
|
||||
def test_wrong_schema(self):
|
||||
oid = self._create_type("type_ii", [("a", "integer"), ("b", "integer")])
|
||||
from psycopg2.extras import CompositeCaster
|
||||
c = CompositeCaster('type_ii', oid, [('a', 23), ('b', 23), ('c', 23)])
|
||||
curs = self.conn.cursor()
|
||||
psycopg2.extensions.register_type(c.typecaster, curs)
|
||||
|
|
@ -745,8 +724,6 @@ class AdaptTypeTestCase(ConnectingTestCase):
|
|||
|
||||
@skip_if_no_composite
|
||||
def test_non_dbapi_connection(self):
|
||||
from psycopg2.extras import RealDictConnection
|
||||
from psycopg2.extras import register_composite
|
||||
self._create_type("type_ii", [("a", "integer"), ("b", "integer")])
|
||||
|
||||
conn = self.connect(connection_factory=RealDictConnection)
|
||||
|
|
@ -772,8 +749,6 @@ class AdaptTypeTestCase(ConnectingTestCase):
|
|||
oid = self._create_type("type_isd",
|
||||
[('anint', 'integer'), ('astring', 'text'), ('adate', 'date')])
|
||||
|
||||
from psycopg2.extras import register_composite, CompositeCaster
|
||||
|
||||
class DictComposite(CompositeCaster):
|
||||
def make(self, values):
|
||||
return dict(zip(self.attnames, values))
|
||||
|
|
@ -832,8 +807,6 @@ def skip_if_no_json_type(f):
|
|||
|
||||
class JsonTestCase(ConnectingTestCase):
|
||||
def test_adapt(self):
|
||||
from psycopg2.extras import json, Json
|
||||
|
||||
objs = [None, "te'xt", 123, 123.45,
|
||||
u'\xe0\u20ac', ['a', 100], {'a': 100}]
|
||||
|
||||
|
|
@ -843,8 +816,6 @@ class JsonTestCase(ConnectingTestCase):
|
|||
psycopg2.extensions.QuotedString(json.dumps(obj)).getquoted())
|
||||
|
||||
def test_adapt_dumps(self):
|
||||
from psycopg2.extras import json, Json
|
||||
|
||||
class DecimalEncoder(json.JSONEncoder):
|
||||
def default(self, obj):
|
||||
if isinstance(obj, Decimal):
|
||||
|
|
@ -860,8 +831,6 @@ class JsonTestCase(ConnectingTestCase):
|
|||
b"'123.45'")
|
||||
|
||||
def test_adapt_subclass(self):
|
||||
from psycopg2.extras import json, Json
|
||||
|
||||
class DecimalEncoder(json.JSONEncoder):
|
||||
def default(self, obj):
|
||||
if isinstance(obj, Decimal):
|
||||
|
|
@ -877,7 +846,6 @@ class JsonTestCase(ConnectingTestCase):
|
|||
self.assertQuotedEqual(curs.mogrify("%s", (MyJson(obj),)), b"'123.45'")
|
||||
|
||||
def test_register_on_dict(self):
|
||||
from psycopg2.extras import Json
|
||||
psycopg2.extensions.register_adapter(dict, Json)
|
||||
|
||||
try:
|
||||
|
|
@ -953,7 +921,6 @@ class JsonTestCase(ConnectingTestCase):
|
|||
|
||||
@skip_if_no_json_type
|
||||
def test_no_conn_curs(self):
|
||||
from psycopg2._json import _get_json_oids
|
||||
oid, array_oid = _get_json_oids(self.conn)
|
||||
|
||||
old = psycopg2.extensions.string_types.get(114)
|
||||
|
|
@ -1141,7 +1108,6 @@ class JsonbTestCase(ConnectingTestCase):
|
|||
|
||||
class RangeTestCase(unittest.TestCase):
|
||||
def test_noparam(self):
|
||||
from psycopg2.extras import Range
|
||||
r = Range()
|
||||
|
||||
self.assert_(not r.isempty)
|
||||
|
|
@ -1153,7 +1119,6 @@ class RangeTestCase(unittest.TestCase):
|
|||
self.assert_(not r.upper_inc)
|
||||
|
||||
def test_empty(self):
|
||||
from psycopg2.extras import Range
|
||||
r = Range(empty=True)
|
||||
|
||||
self.assert_(r.isempty)
|
||||
|
|
@ -1165,7 +1130,6 @@ class RangeTestCase(unittest.TestCase):
|
|||
self.assert_(not r.upper_inc)
|
||||
|
||||
def test_nobounds(self):
|
||||
from psycopg2.extras import Range
|
||||
r = Range(10, 20)
|
||||
self.assertEqual(r.lower, 10)
|
||||
self.assertEqual(r.upper, 20)
|
||||
|
|
@ -1176,7 +1140,6 @@ class RangeTestCase(unittest.TestCase):
|
|||
self.assert_(not r.upper_inc)
|
||||
|
||||
def test_bounds(self):
|
||||
from psycopg2.extras import Range
|
||||
for bounds, lower_inc, upper_inc in [
|
||||
('[)', True, False),
|
||||
('(]', False, True),
|
||||
|
|
@ -1192,7 +1155,6 @@ class RangeTestCase(unittest.TestCase):
|
|||
self.assertEqual(r.upper_inc, upper_inc)
|
||||
|
||||
def test_keywords(self):
|
||||
from psycopg2.extras import Range
|
||||
r = Range(upper=20)
|
||||
self.assertEqual(r.lower, None)
|
||||
self.assertEqual(r.upper, 20)
|
||||
|
|
@ -1212,12 +1174,10 @@ class RangeTestCase(unittest.TestCase):
|
|||
self.assert_(not r.upper_inc)
|
||||
|
||||
def test_bad_bounds(self):
|
||||
from psycopg2.extras import Range
|
||||
self.assertRaises(ValueError, Range, bounds='(')
|
||||
self.assertRaises(ValueError, Range, bounds='[}')
|
||||
|
||||
def test_in(self):
|
||||
from psycopg2.extras import Range
|
||||
r = Range(empty=True)
|
||||
self.assert_(10 not in r)
|
||||
|
||||
|
|
@ -1269,7 +1229,6 @@ class RangeTestCase(unittest.TestCase):
|
|||
self.assert_(21 not in r)
|
||||
|
||||
def test_nonzero(self):
|
||||
from psycopg2.extras import Range
|
||||
self.assert_(Range())
|
||||
self.assert_(Range(10, 20))
|
||||
self.assert_(not Range(empty=True))
|
||||
|
|
@ -1279,7 +1238,6 @@ class RangeTestCase(unittest.TestCase):
|
|||
self.assert_(r1 == r2)
|
||||
self.assert_(hash(r1) == hash(r2))
|
||||
|
||||
from psycopg2.extras import Range
|
||||
assert_equal(Range(empty=True), Range(empty=True))
|
||||
assert_equal(Range(), Range())
|
||||
assert_equal(Range(10, None), Range(10, None))
|
||||
|
|
@ -1296,12 +1254,9 @@ class RangeTestCase(unittest.TestCase):
|
|||
assert_not_equal(Range(10, 20, '[)'), Range(10, 20, '[]'))
|
||||
|
||||
def test_eq_wrong_type(self):
|
||||
from psycopg2.extras import Range
|
||||
self.assertNotEqual(Range(10, 20), ())
|
||||
|
||||
def test_eq_subclass(self):
|
||||
from psycopg2.extras import Range, NumericRange
|
||||
|
||||
class IntRange(NumericRange):
|
||||
pass
|
||||
|
||||
|
|
@ -1316,7 +1271,6 @@ class RangeTestCase(unittest.TestCase):
|
|||
# and consistent.
|
||||
|
||||
def test_lt_ordering(self):
|
||||
from psycopg2.extras import Range
|
||||
self.assert_(Range(empty=True) < Range(0, 4))
|
||||
self.assert_(not Range(1, 2) < Range(0, 4))
|
||||
self.assert_(Range(0, 4) < Range(1, 2))
|
||||
|
|
@ -1332,7 +1286,6 @@ class RangeTestCase(unittest.TestCase):
|
|||
self.assert_(not Range(1, 2) < 1)
|
||||
|
||||
def test_gt_ordering(self):
|
||||
from psycopg2.extras import Range
|
||||
self.assert_(not Range(empty=True) > Range(0, 4))
|
||||
self.assert_(Range(1, 2) > Range(0, 4))
|
||||
self.assert_(not Range(0, 4) > Range(1, 2))
|
||||
|
|
@ -1348,7 +1301,6 @@ class RangeTestCase(unittest.TestCase):
|
|||
self.assert_(Range(1, 2) > 1)
|
||||
|
||||
def test_le_ordering(self):
|
||||
from psycopg2.extras import Range
|
||||
self.assert_(Range(empty=True) <= Range(0, 4))
|
||||
self.assert_(not Range(1, 2) <= Range(0, 4))
|
||||
self.assert_(Range(0, 4) <= Range(1, 2))
|
||||
|
|
@ -1364,7 +1316,6 @@ class RangeTestCase(unittest.TestCase):
|
|||
self.assert_(not Range(1, 2) <= 1)
|
||||
|
||||
def test_ge_ordering(self):
|
||||
from psycopg2.extras import Range
|
||||
self.assert_(not Range(empty=True) >= Range(0, 4))
|
||||
self.assert_(Range(1, 2) >= Range(0, 4))
|
||||
self.assert_(not Range(0, 4) >= Range(1, 2))
|
||||
|
|
@ -1380,8 +1331,6 @@ class RangeTestCase(unittest.TestCase):
|
|||
self.assert_(Range(1, 2) >= 1)
|
||||
|
||||
def test_pickling(self):
|
||||
from psycopg2.extras import Range
|
||||
|
||||
r = Range(0, 4)
|
||||
self.assertEqual(loads(dumps(r)), r)
|
||||
|
||||
|
|
@ -1392,7 +1341,6 @@ class RangeTestCase(unittest.TestCase):
|
|||
Using ``repr`` for all string conversions can be very unreadable for
|
||||
longer types like ``DateTimeTZRange``.
|
||||
'''
|
||||
from psycopg2.extras import Range
|
||||
|
||||
# Using the "u" prefix to make sure we have the proper return types in
|
||||
# Python2
|
||||
|
|
@ -1420,9 +1368,7 @@ class RangeTestCase(unittest.TestCase):
|
|||
Date-Time ranges should return a human-readable string as well on
|
||||
string conversion.
|
||||
'''
|
||||
from psycopg2.extras import DateTimeTZRange
|
||||
from datetime import datetime
|
||||
from psycopg2.tz import FixedOffsetTimezone
|
||||
|
||||
converter = unicode if sys.version_info < (3, 0) else str
|
||||
tz = FixedOffsetTimezone(-5 * 60, "EST")
|
||||
r = DateTimeTZRange(datetime(2010, 1, 1, tzinfo=tz),
|
||||
|
|
@ -1446,7 +1392,6 @@ class RangeCasterTestCase(ConnectingTestCase):
|
|||
self.assertEqual(r, None)
|
||||
|
||||
def test_cast_empty(self):
|
||||
from psycopg2.extras import Range
|
||||
cur = self.conn.cursor()
|
||||
for type in self.builtin_ranges:
|
||||
cur.execute("select 'empty'::%s" % type)
|
||||
|
|
@ -1455,7 +1400,6 @@ class RangeCasterTestCase(ConnectingTestCase):
|
|||
self.assert_(r.isempty)
|
||||
|
||||
def test_cast_inf(self):
|
||||
from psycopg2.extras import Range
|
||||
cur = self.conn.cursor()
|
||||
for type in self.builtin_ranges:
|
||||
cur.execute("select '(,)'::%s" % type)
|
||||
|
|
@ -1466,7 +1410,6 @@ class RangeCasterTestCase(ConnectingTestCase):
|
|||
self.assert_(r.upper_inf)
|
||||
|
||||
def test_cast_numbers(self):
|
||||
from psycopg2.extras import NumericRange
|
||||
cur = self.conn.cursor()
|
||||
for type in ('int4range', 'int8range'):
|
||||
cur.execute("select '(10,20)'::%s" % type)
|
||||
|
|
@ -1492,7 +1435,6 @@ class RangeCasterTestCase(ConnectingTestCase):
|
|||
self.assert_(not r.upper_inc)
|
||||
|
||||
def test_cast_date(self):
|
||||
from psycopg2.extras import DateRange
|
||||
cur = self.conn.cursor()
|
||||
cur.execute("select '(2000-01-01,2012-12-31)'::daterange")
|
||||
r = cur.fetchone()[0]
|
||||
|
|
@ -1506,7 +1448,6 @@ class RangeCasterTestCase(ConnectingTestCase):
|
|||
self.assert_(not r.upper_inc)
|
||||
|
||||
def test_cast_timestamp(self):
|
||||
from psycopg2.extras import DateTimeRange
|
||||
cur = self.conn.cursor()
|
||||
ts1 = datetime(2000, 1, 1)
|
||||
ts2 = datetime(2000, 12, 31, 23, 59, 59, 999)
|
||||
|
|
@ -1522,8 +1463,6 @@ class RangeCasterTestCase(ConnectingTestCase):
|
|||
self.assert_(not r.upper_inc)
|
||||
|
||||
def test_cast_timestamptz(self):
|
||||
from psycopg2.extras import DateTimeTZRange
|
||||
from psycopg2.tz import FixedOffsetTimezone
|
||||
cur = self.conn.cursor()
|
||||
ts1 = datetime(2000, 1, 1, tzinfo=FixedOffsetTimezone(600))
|
||||
ts2 = datetime(2000, 12, 31, 23, 59, 59, 999,
|
||||
|
|
@ -1540,7 +1479,6 @@ class RangeCasterTestCase(ConnectingTestCase):
|
|||
self.assert_(r.upper_inc)
|
||||
|
||||
def test_adapt_number_range(self):
|
||||
from psycopg2.extras import NumericRange
|
||||
cur = self.conn.cursor()
|
||||
|
||||
r = NumericRange(empty=True)
|
||||
|
|
@ -1568,7 +1506,6 @@ class RangeCasterTestCase(ConnectingTestCase):
|
|||
self.assert_(r1.upper_inc)
|
||||
|
||||
def test_adapt_numeric_range(self):
|
||||
from psycopg2.extras import NumericRange
|
||||
cur = self.conn.cursor()
|
||||
|
||||
r = NumericRange(empty=True)
|
||||
|
|
@ -1596,8 +1533,6 @@ class RangeCasterTestCase(ConnectingTestCase):
|
|||
self.assert_(r1.upper_inc)
|
||||
|
||||
def test_adapt_date_range(self):
|
||||
from psycopg2.extras import DateRange, DateTimeRange, DateTimeTZRange
|
||||
from psycopg2.tz import FixedOffsetTimezone
|
||||
cur = self.conn.cursor()
|
||||
|
||||
d1 = date(2012, 1, 1)
|
||||
|
|
@ -1630,7 +1565,6 @@ class RangeCasterTestCase(ConnectingTestCase):
|
|||
self.assert_(r1.upper_inc)
|
||||
|
||||
def test_register_range_adapter(self):
|
||||
from psycopg2.extras import Range, register_range
|
||||
cur = self.conn.cursor()
|
||||
cur.execute("create type textrange as range (subtype=text)")
|
||||
rc = register_range('textrange', 'TextRange', cur)
|
||||
|
|
@ -1660,7 +1594,6 @@ class RangeCasterTestCase(ConnectingTestCase):
|
|||
del ext.adapters[rc.range, ext.ISQLQuote]
|
||||
|
||||
def test_range_escaping(self):
|
||||
from psycopg2.extras import register_range
|
||||
cur = self.conn.cursor()
|
||||
cur.execute("create type textrange as range (subtype=text)")
|
||||
rc = register_range('textrange', 'TextRange', cur)
|
||||
|
|
@ -1714,7 +1647,6 @@ class RangeCasterTestCase(ConnectingTestCase):
|
|||
del ext.adapters[TextRange, ext.ISQLQuote]
|
||||
|
||||
def test_range_not_found(self):
|
||||
from psycopg2.extras import register_range
|
||||
cur = self.conn.cursor()
|
||||
self.assertRaises(psycopg2.ProgrammingError,
|
||||
register_range, 'nosuchrange', 'FailRange', cur)
|
||||
|
|
@ -1728,7 +1660,6 @@ class RangeCasterTestCase(ConnectingTestCase):
|
|||
cur.execute("create type rs.r3 as range (subtype=text)")
|
||||
cur.execute("savepoint x")
|
||||
|
||||
from psycopg2.extras import register_range
|
||||
ra1 = register_range('r1', 'r1', cur)
|
||||
ra2 = register_range('r2', 'r2', cur)
|
||||
rars2 = register_range('rs.r2', 'r2', cur)
|
||||
|
|
|
|||
|
|
@ -33,8 +33,11 @@ import unittest
|
|||
from functools import wraps
|
||||
from ctypes.util import find_library
|
||||
from .testconfig import dsn, repl_dsn
|
||||
from psycopg2 import ProgrammingError
|
||||
from psycopg2.compat import text_type
|
||||
|
||||
from .testconfig import green
|
||||
|
||||
# Python 2/3 compatibility
|
||||
|
||||
if sys.version_info[0] == 2:
|
||||
|
|
@ -241,7 +244,6 @@ def skip_if_tpc_disabled(f):
|
|||
"""Skip a test if the server has tpc support disabled."""
|
||||
@wraps(f)
|
||||
def skip_if_tpc_disabled_(self):
|
||||
from psycopg2 import ProgrammingError
|
||||
cnn = self.connect()
|
||||
cur = cnn.cursor()
|
||||
try:
|
||||
|
|
@ -368,7 +370,6 @@ def skip_if_no_superuser(f):
|
|||
"""Skip a test if the database user running the test is not a superuser"""
|
||||
@wraps(f)
|
||||
def skip_if_no_superuser_(self):
|
||||
from psycopg2 import ProgrammingError
|
||||
try:
|
||||
return f(self)
|
||||
except ProgrammingError as e:
|
||||
|
|
@ -383,7 +384,6 @@ def skip_if_no_superuser(f):
|
|||
|
||||
def skip_if_green(reason):
|
||||
def skip_if_green_(cls):
|
||||
from .testconfig import green
|
||||
decorator = unittest.skipIf(green, reason)
|
||||
return decorator(cls)
|
||||
return skip_if_green_
|
||||
|
|
|
|||
Loading…
Reference in New Issue
Block a user