summaryrefslogtreecommitdiff
path: root/tests
diff options
context:
space:
mode:
Diffstat (limited to 'tests')
-rwxr-xr-xtests/test_copy.py5
-rwxr-xr-xtests/test_cursor.py3
-rwxr-xr-xtests/test_extras_dictcursor.py41
-rwxr-xr-xtests/test_types_extras.py22
-rw-r--r--tests/testutils.py27
5 files changed, 7 insertions, 91 deletions
diff --git a/tests/test_copy.py b/tests/test_copy.py
index 9662321..c5e7913 100755
--- a/tests/test_copy.py
+++ b/tests/test_copy.py
@@ -25,7 +25,7 @@
import sys
import string
from testutils import (unittest, ConnectingTestCase, decorate_all_tests,
- skip_if_no_iobase, skip_before_postgres, slow)
+ skip_before_postgres, slow)
from cStringIO import StringIO
from itertools import cycle, izip
from subprocess import Popen, PIPE
@@ -131,7 +131,6 @@ class CopyTests(ConnectingTestCase):
finally:
curs.close()
- @skip_if_no_iobase
def test_copy_text(self):
self.conn.set_client_encoding('latin1')
self._create_temp_table() # the above call closed the xn
@@ -154,7 +153,6 @@ class CopyTests(ConnectingTestCase):
f.seek(0)
self.assertEqual(f.readline().rstrip(), about)
- @skip_if_no_iobase
def test_copy_bytes(self):
self.conn.set_client_encoding('latin1')
self._create_temp_table() # the above call closed the xn
@@ -176,7 +174,6 @@ class CopyTests(ConnectingTestCase):
f.seek(0)
self.assertEqual(f.readline().rstrip(), about)
- @skip_if_no_iobase
def test_copy_expert_textiobase(self):
self.conn.set_client_encoding('latin1')
self._create_temp_table() # the above call closed the xn
diff --git a/tests/test_cursor.py b/tests/test_cursor.py
index 10b8d71..dbc72e1 100755
--- a/tests/test_cursor.py
+++ b/tests/test_cursor.py
@@ -27,7 +27,7 @@ import pickle
import psycopg2
import psycopg2.extensions
from testutils import (unittest, ConnectingTestCase, skip_before_postgres,
- skip_if_no_namedtuple, skip_if_no_getrefcount, slow, skip_if_no_superuser,
+ skip_if_no_getrefcount, slow, skip_if_no_superuser,
skip_if_windows)
import psycopg2.extras
@@ -377,7 +377,6 @@ class CursorTests(ConnectingTestCase):
for i, rec in enumerate(curs):
self.assertEqual(i + 1, curs.rownumber)
- @skip_if_no_namedtuple
def test_namedtuple_description(self):
curs = self.conn.cursor()
curs.execute("""select
diff --git a/tests/test_extras_dictcursor.py b/tests/test_extras_dictcursor.py
index 20393c6..2c867d3 100755
--- a/tests/test_extras_dictcursor.py
+++ b/tests/test_extras_dictcursor.py
@@ -19,7 +19,6 @@ from datetime import timedelta
import psycopg2
import psycopg2.extras
from testutils import unittest, ConnectingTestCase, skip_before_postgres
-from testutils import skip_if_no_namedtuple
class ExtrasDictCursorTests(ConnectingTestCase):
@@ -232,11 +231,6 @@ class NamedTupleCursorTest(ConnectingTestCase):
ConnectingTestCase.setUp(self)
from psycopg2.extras import NamedTupleConnection
- try:
- from collections import namedtuple # noqa
- except ImportError:
- return
-
self.conn = self.connect(connection_factory=NamedTupleConnection)
curs = self.conn.cursor()
curs.execute("CREATE TEMPORARY TABLE nttest (i int, s text)")
@@ -245,13 +239,11 @@ class NamedTupleCursorTest(ConnectingTestCase):
curs.execute("INSERT INTO nttest VALUES (3, 'baz')")
self.conn.commit()
- @skip_if_no_namedtuple
def test_cursor_args(self):
cur = self.conn.cursor('foo', cursor_factory=psycopg2.extras.DictCursor)
self.assertEqual(cur.name, 'foo')
self.assert_(isinstance(cur, psycopg2.extras.DictCursor))
- @skip_if_no_namedtuple
def test_fetchone(self):
curs = self.conn.cursor()
curs.execute("select * from nttest order by 1")
@@ -263,7 +255,6 @@ class NamedTupleCursorTest(ConnectingTestCase):
self.assertEqual(curs.rownumber, 1)
self.assertEqual(curs.rowcount, 3)
- @skip_if_no_namedtuple
def test_fetchmany_noarg(self):
curs = self.conn.cursor()
curs.arraysize = 2
@@ -277,7 +268,6 @@ class NamedTupleCursorTest(ConnectingTestCase):
self.assertEqual(curs.rownumber, 2)
self.assertEqual(curs.rowcount, 3)
- @skip_if_no_namedtuple
def test_fetchmany(self):
curs = self.conn.cursor()
curs.execute("select * from nttest order by 1")
@@ -290,7 +280,6 @@ class NamedTupleCursorTest(ConnectingTestCase):
self.assertEqual(curs.rownumber, 2)
self.assertEqual(curs.rowcount, 3)
- @skip_if_no_namedtuple
def test_fetchall(self):
curs = self.conn.cursor()
curs.execute("select * from nttest order by 1")
@@ -305,7 +294,6 @@ class NamedTupleCursorTest(ConnectingTestCase):
self.assertEqual(curs.rownumber, 3)
self.assertEqual(curs.rowcount, 3)
- @skip_if_no_namedtuple
def test_executemany(self):
curs = self.conn.cursor()
curs.executemany("delete from nttest where i = %s",
@@ -316,7 +304,6 @@ class NamedTupleCursorTest(ConnectingTestCase):
self.assertEqual(res[0].i, 3)
self.assertEqual(res[0].s, 'baz')
- @skip_if_no_namedtuple
def test_iter(self):
curs = self.conn.cursor()
curs.execute("select * from nttest order by 1")
@@ -342,26 +329,6 @@ class NamedTupleCursorTest(ConnectingTestCase):
self.assertEqual(curs.rownumber, 3)
self.assertEqual(curs.rowcount, 3)
- def test_error_message(self):
- try:
- from collections import namedtuple # noqa
- except ImportError:
- # an import error somewhere
- from psycopg2.extras import NamedTupleConnection
- try:
- self.conn = self.connect(
- connection_factory=NamedTupleConnection)
- curs = self.conn.cursor()
- curs.execute("select 1")
- curs.fetchone()
- except ImportError:
- pass
- else:
- self.fail("expecting ImportError")
- else:
- return self.skipTest("namedtuple available")
-
- @skip_if_no_namedtuple
def test_record_updated(self):
curs = self.conn.cursor()
curs.execute("select 1 as foo;")
@@ -373,7 +340,6 @@ class NamedTupleCursorTest(ConnectingTestCase):
self.assertEqual(r.bar, 2)
self.assertRaises(AttributeError, getattr, r, 'foo')
- @skip_if_no_namedtuple
def test_no_result_no_surprise(self):
curs = self.conn.cursor()
curs.execute("update nttest set s = s")
@@ -382,7 +348,6 @@ class NamedTupleCursorTest(ConnectingTestCase):
curs.execute("update nttest set s = s")
self.assertRaises(psycopg2.ProgrammingError, curs.fetchall)
- @skip_if_no_namedtuple
def test_minimal_generation(self):
# Instrument the class to verify it gets called the minimum number of times.
from psycopg2.extras import NamedTupleCursor
@@ -416,7 +381,6 @@ class NamedTupleCursorTest(ConnectingTestCase):
finally:
NamedTupleCursor._make_nt = f_orig
- @skip_if_no_namedtuple
@skip_before_postgres(8, 0)
def test_named(self):
curs = self.conn.cursor('tmp')
@@ -427,28 +391,24 @@ class NamedTupleCursorTest(ConnectingTestCase):
recs.extend(curs.fetchall())
self.assertEqual(range(10), [t.i for t in recs])
- @skip_if_no_namedtuple
def test_named_fetchone(self):
curs = self.conn.cursor('tmp')
curs.execute("""select 42 as i""")
t = curs.fetchone()
self.assertEqual(t.i, 42)
- @skip_if_no_namedtuple
def test_named_fetchmany(self):
curs = self.conn.cursor('tmp')
curs.execute("""select 42 as i""")
recs = curs.fetchmany(10)
self.assertEqual(recs[0].i, 42)
- @skip_if_no_namedtuple
def test_named_fetchall(self):
curs = self.conn.cursor('tmp')
curs.execute("""select 42 as i""")
recs = curs.fetchall()
self.assertEqual(recs[0].i, 42)
- @skip_if_no_namedtuple
@skip_before_postgres(8, 2)
def test_not_greedy(self):
curs = self.conn.cursor('tmp')
@@ -463,7 +423,6 @@ class NamedTupleCursorTest(ConnectingTestCase):
self.assert_(recs[1].ts - recs[0].ts < timedelta(seconds=0.005))
self.assert_(recs[2].ts - recs[1].ts > timedelta(seconds=0.0099))
- @skip_if_no_namedtuple
@skip_before_postgres(8, 0)
def test_named_rownumber(self):
curs = self.conn.cursor('tmp')
diff --git a/tests/test_types_extras.py b/tests/test_types_extras.py
index e1db2c2..a5aa7d2 100755
--- a/tests/test_types_extras.py
+++ b/tests/test_types_extras.py
@@ -541,16 +541,10 @@ class AdaptTypeTestCase(ConnectingTestCase):
self.assertEqual(v[0], 10)
self.assertEqual(v[1], "hello")
self.assertEqual(v[2], date(2011, 1, 2))
-
- try:
- from collections import namedtuple # noqa
- except ImportError:
- pass
- else:
- self.assert_(t.type is not tuple)
- self.assertEqual(v.anint, 10)
- self.assertEqual(v.astring, "hello")
- self.assertEqual(v.adate, date(2011, 1, 2))
+ self.assert_(t.type is not tuple)
+ self.assertEqual(v.anint, 10)
+ self.assertEqual(v.astring, "hello")
+ self.assertEqual(v.adate, date(2011, 1, 2))
@skip_if_no_composite
def test_empty_string(self):
@@ -591,13 +585,7 @@ class AdaptTypeTestCase(ConnectingTestCase):
v = curs.fetchone()[0]
self.assertEqual(r, v)
-
- try:
- from collections import namedtuple # noqa
- except ImportError:
- pass
- else:
- self.assertEqual(v.anotherpair.apair.astring, "hello")
+ self.assertEqual(v.anotherpair.apair.astring, "hello")
@skip_if_no_composite
def test_register_on_cursor(self):
diff --git a/tests/testutils.py b/tests/testutils.py
index 033aba8..ba31e3b 100644
--- a/tests/testutils.py
+++ b/tests/testutils.py
@@ -243,33 +243,6 @@ def skip_if_tpc_disabled(f):
return skip_if_tpc_disabled_
-def skip_if_no_namedtuple(f):
- @wraps(f)
- def skip_if_no_namedtuple_(self):
- try:
- from collections import namedtuple # noqa
- except ImportError:
- return self.skipTest("collections.namedtuple not available")
- else:
- return f(self)
-
- return skip_if_no_namedtuple_
-
-
-def skip_if_no_iobase(f):
- """Skip a test if io.TextIOBase is not available."""
- @wraps(f)
- def skip_if_no_iobase_(self):
- try:
- from io import TextIOBase # noqa
- except ImportError:
- return self.skipTest("io.TextIOBase not found.")
- else:
- return f(self)
-
- return skip_if_no_iobase_
-
-
def skip_before_postgres(*ver):
"""Skip a test on PostgreSQL before a certain version."""
ver = ver + (0,) * (3 - len(ver))