summaryrefslogtreecommitdiff
path: root/tests
diff options
context:
space:
mode:
authorDaniele Varrazzo <daniele.varrazzo@gmail.com>2018-05-20 23:56:29 +0100
committerDaniele Varrazzo <daniele.varrazzo@gmail.com>2018-05-20 23:56:29 +0100
commite8a831dda2decd26f57bb227948da8af9237fb89 (patch)
tree1395c3acd39c74989c0a891e57f01862e8cb698d /tests
parentbe3b1ba1eb84244e97a9d85ba05ef2ca6b5a213b (diff)
parenta0f7027ad578c6c2707430e9a208cc3f7510cad4 (diff)
downloadpsycopg2-e8a831dda2decd26f57bb227948da8af9237fb89.tar.gz
Merge branch 'master' into drop-2to3
Diffstat (limited to 'tests')
-rwxr-xr-xtests/test_connection.py120
-rwxr-xr-xtests/test_cursor.py11
-rwxr-xr-xtests/test_dates.py11
-rwxr-xr-xtests/test_extras_dictcursor.py229
-rwxr-xr-xtests/test_types_basic.py29
-rwxr-xr-xtests/test_types_extras.py4
-rwxr-xr-xtests/test_with.py7
7 files changed, 325 insertions, 86 deletions
diff --git a/tests/test_connection.py b/tests/test_connection.py
index 02db264..13635f1 100755
--- a/tests/test_connection.py
+++ b/tests/test_connection.py
@@ -37,7 +37,9 @@ from psycopg2 import extensions as ext
from .testutils import (
unittest, decorate_all_tests, skip_if_no_superuser,
skip_before_postgres, skip_after_postgres, skip_before_libpq,
- ConnectingTestCase, skip_if_tpc_disabled, skip_if_windows, slow)
+ ConnectingTestCase, skip_if_tpc_disabled, skip_if_windows, slow,
+ libpq_version
+)
from .testconfig import dsn, dbname
@@ -246,6 +248,13 @@ class ConnectionTests(ConnectingTestCase):
else:
del os.environ['PGCLIENTENCODING']
+ def test_connect_no_string(self):
+ class MyString(str):
+ pass
+
+ conn = psycopg2.connect(MyString(dsn))
+ conn.close()
+
def test_weakref(self):
from weakref import ref
import gc
@@ -400,6 +409,13 @@ class ParseDsnTestCase(ConnectingTestCase):
self.assertRaises(TypeError, ext.parse_dsn, None)
self.assertRaises(TypeError, ext.parse_dsn, 42)
+ def test_str_subclass(self):
+ class MyString(str):
+ pass
+
+ res = ext.parse_dsn(MyString("dbname=test"))
+ self.assertEqual(res, {'dbname': 'test'})
+
class MakeDsnTestCase(ConnectingTestCase):
def test_empty_arguments(self):
@@ -1381,6 +1397,102 @@ class TransactionControlTests(ConnectingTestCase):
cur.execute("SHOW default_transaction_read_only;")
self.assertEqual(cur.fetchone()[0], 'off')
+ def test_idempotence_check(self):
+ self.conn.autocommit = False
+ self.conn.readonly = True
+ self.conn.autocommit = True
+ self.conn.readonly = True
+
+ cur = self.conn.cursor()
+ cur.execute("SHOW transaction_read_only")
+ self.assertEqual(cur.fetchone()[0], 'on')
+
+
+class TestEncryptPassword(ConnectingTestCase):
+ @skip_before_postgres(10)
+ def test_encrypt_password_post_9_6(self):
+ cur = self.conn.cursor()
+ cur.execute("SHOW password_encryption;")
+ server_encryption_algorithm = cur.fetchone()[0]
+
+ # MD5 algorithm
+ self.assertEqual(
+ ext.encrypt_password('psycopg2', 'ashesh', self.conn, 'md5'),
+ 'md594839d658c28a357126f105b9cb14cfc'
+ )
+
+ # keywords
+ self.assertEqual(
+ ext.encrypt_password(
+ password='psycopg2', user='ashesh',
+ scope=self.conn, algorithm='md5'),
+ 'md594839d658c28a357126f105b9cb14cfc'
+ )
+ if libpq_version() < 100000:
+ self.assertRaises(
+ psycopg2.NotSupportedError,
+ ext.encrypt_password, 'psycopg2', 'ashesh', self.conn,
+ 'scram-sha-256'
+ )
+ else:
+ enc_password = ext.encrypt_password(
+ 'psycopg2', 'ashesh', self.conn
+ )
+ if server_encryption_algorithm == 'md5':
+ self.assertEqual(
+ enc_password, 'md594839d658c28a357126f105b9cb14cfc'
+ )
+ elif server_encryption_algorithm == 'scram-sha-256':
+ self.assertEqual(enc_password[:14], 'SCRAM-SHA-256$')
+
+ self.assertEqual(
+ ext.encrypt_password(
+ 'psycopg2', 'ashesh', self.conn, 'scram-sha-256'
+ )[:14], 'SCRAM-SHA-256$'
+ )
+
+ self.assertRaises(psycopg2.ProgrammingError,
+ ext.encrypt_password, 'psycopg2', 'ashesh', self.conn, 'abc')
+
+ @skip_after_postgres(10)
+ def test_encrypt_password_pre_10(self):
+ self.assertEqual(
+ ext.encrypt_password('psycopg2', 'ashesh', self.conn),
+ 'md594839d658c28a357126f105b9cb14cfc'
+ )
+
+ self.assertRaises(psycopg2.ProgrammingError,
+ ext.encrypt_password, 'psycopg2', 'ashesh', self.conn, 'abc')
+
+ def test_encrypt_md5(self):
+ self.assertEqual(
+ ext.encrypt_password('psycopg2', 'ashesh', algorithm='md5'),
+ 'md594839d658c28a357126f105b9cb14cfc'
+ )
+
+ def test_encrypt_scram(self):
+ if libpq_version() >= 100000:
+ self.assert_(
+ ext.encrypt_password(
+ 'psycopg2', 'ashesh', self.conn, 'scram-sha-256')
+ .startswith('SCRAM-SHA-256$'))
+ else:
+ self.assertRaises(psycopg2.NotSupportedError,
+ ext.encrypt_password,
+ password='psycopg2', user='ashesh',
+ scope=self.conn, algorithm='scram-sha-256')
+
+ def test_bad_types(self):
+ self.assertRaises(TypeError, ext.encrypt_password)
+ self.assertRaises(TypeError, ext.encrypt_password,
+ 'password', 42, self.conn, 'md5')
+ self.assertRaises(TypeError, ext.encrypt_password,
+ 42, 'user', self.conn, 'md5')
+ self.assertRaises(TypeError, ext.encrypt_password,
+ 42, 'user', 'wat', 'abc')
+ self.assertRaises(TypeError, ext.encrypt_password,
+ 'password', 'user', 'wat', 42)
+
class AutocommitTests(ConnectingTestCase):
def test_closed(self):
@@ -1539,9 +1651,13 @@ import os
import sys
import time
import signal
+import warnings
import threading
-import psycopg2
+# ignore wheel deprecation warning
+with warnings.catch_warnings():
+ warnings.simplefilter('ignore')
+ import psycopg2
def handle_sigabort(sig, frame):
sys.exit(1)
diff --git a/tests/test_cursor.py b/tests/test_cursor.py
index d522386..b48fe7f 100755
--- a/tests/test_cursor.py
+++ b/tests/test_cursor.py
@@ -119,6 +119,12 @@ class CursorTests(ConnectingTestCase):
nref2 = sys.getrefcount(foo)
self.assertEqual(nref1, nref2)
+ def test_modify_closed(self):
+ cur = self.conn.cursor()
+ cur.close()
+ sql = cur.mogrify("select %s", (10,))
+ self.assertEqual(sql, b"select 10")
+
def test_bad_placeholder(self):
cur = self.conn.cursor()
self.assertRaises(psycopg2.ProgrammingError,
@@ -431,6 +437,11 @@ class CursorTests(ConnectingTestCase):
self.assertEqual([(5,), (6,), (7,)], cur2.fetchall())
@skip_before_postgres(8, 0)
+ def test_named_noop_close(self):
+ cur = self.conn.cursor('test')
+ cur.close()
+
+ @skip_before_postgres(8, 0)
def test_scroll(self):
cur = self.conn.cursor()
cur.execute("select generate_series(0,9)")
diff --git a/tests/test_dates.py b/tests/test_dates.py
index 47ef41c..bb5aee3 100755
--- a/tests/test_dates.py
+++ b/tests/test_dates.py
@@ -438,6 +438,14 @@ class DatetimeTests(ConnectingTestCase, CommonDatetimeTestsMixin):
r = cur.fetchone()[0]
self.assertEqual(r, v, "%s -> %s != %s" % (s, r, v))
+ @skip_before_postgres(8, 4)
+ def test_interval_iso_8601_not_supported(self):
+ # We may end up supporting, but no pressure for it
+ cur = self.conn.cursor()
+ cur.execute("set local intervalstyle to iso_8601")
+ cur.execute("select '1 day 2 hours'::interval")
+ self.assertRaises(psycopg2.NotSupportedError, cur.fetchone)
+
# Only run the datetime tests if psycopg was compiled with support.
if not hasattr(psycopg2.extensions, 'PYDATETIME'):
@@ -639,7 +647,8 @@ class FromTicksTestCase(unittest.TestCase):
def test_date_value_error_sec_59_99(self):
from datetime import date
s = psycopg2.DateFromTicks(1273173119.99992)
- self.assertEqual(s.adapted, date(2010, 5, 6))
+ # 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
diff --git a/tests/test_extras_dictcursor.py b/tests/test_extras_dictcursor.py
index 75c2277..a9201f1 100755
--- a/tests/test_extras_dictcursor.py
+++ b/tests/test_extras_dictcursor.py
@@ -19,12 +19,11 @@ from datetime import timedelta
import psycopg2
import psycopg2.extras
import unittest
-from .testutils import ConnectingTestCase, skip_before_postgres
+from .testutils import ConnectingTestCase, skip_before_postgres, \
+ skip_before_python, skip_from_python
-class ExtrasDictCursorTests(ConnectingTestCase):
- """Test if DictCursor extension class works."""
-
+class _DictCursorBase(ConnectingTestCase):
def setUp(self):
ConnectingTestCase.setUp(self)
curs = self.conn.cursor()
@@ -32,6 +31,30 @@ class ExtrasDictCursorTests(ConnectingTestCase):
curs.execute("INSERT INTO ExtrasDictCursorTests VALUES ('bar')")
self.conn.commit()
+ def _testIterRowNumber(self, curs):
+ # Only checking for dataset < itersize:
+ # see CursorTests.test_iter_named_cursor_rownumber
+ curs.itersize = 20
+ curs.execute("""select * from generate_series(1,10)""")
+ for i, r in enumerate(curs):
+ self.assertEqual(i + 1, curs.rownumber)
+
+ def _testNamedCursorNotGreedy(self, curs):
+ curs.itersize = 2
+ curs.execute("""select clock_timestamp() as ts from generate_series(1,3)""")
+ recs = []
+ for t in curs:
+ time.sleep(0.01)
+ recs.append(t)
+
+ # check that the dataset was not fetched in a single gulp
+ self.assert_(recs[1]['ts'] - recs[0]['ts'] < timedelta(seconds=0.005))
+ self.assert_(recs[2]['ts'] - recs[1]['ts'] > timedelta(seconds=0.0099))
+
+
+class ExtrasDictCursorTests(_DictCursorBase):
+ """Test if DictCursor extension class works."""
+
def testDictConnCursorArgs(self):
self.conn.close()
self.conn = self.connect(connection_factory=psycopg2.extras.DictConnection)
@@ -81,35 +104,6 @@ class ExtrasDictCursorTests(ConnectingTestCase):
self.failUnless(row[0] == 'bar')
return row
- def testDictCursorWithPlainCursorRealFetchOne(self):
- self._testWithPlainCursorReal(lambda curs: curs.fetchone())
-
- def testDictCursorWithPlainCursorRealFetchMany(self):
- self._testWithPlainCursorReal(lambda curs: curs.fetchmany(100)[0])
-
- def testDictCursorWithPlainCursorRealFetchManyNoarg(self):
- self._testWithPlainCursorReal(lambda curs: curs.fetchmany()[0])
-
- def testDictCursorWithPlainCursorRealFetchAll(self):
- self._testWithPlainCursorReal(lambda curs: curs.fetchall()[0])
-
- def testDictCursorWithPlainCursorRealIter(self):
- def getter(curs):
- for row in curs:
- return row
- self._testWithPlainCursorReal(getter)
-
- @skip_before_postgres(8, 0)
- def testDictCursorWithPlainCursorRealIterRowNumber(self):
- curs = self.conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor)
- self._testIterRowNumber(curs)
-
- def _testWithPlainCursorReal(self, getter):
- curs = self.conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor)
- curs.execute("SELECT * FROM ExtrasDictCursorTests")
- row = getter(curs)
- self.failUnless(row['foo'] == 'bar')
-
def testDictCursorWithNamedCursorFetchOne(self):
self._testWithNamedCursor(lambda curs: curs.fetchone())
@@ -145,6 +139,94 @@ class ExtrasDictCursorTests(ConnectingTestCase):
self.failUnless(row['foo'] == 'bar')
self.failUnless(row[0] == 'bar')
+ def testPickleDictRow(self):
+ import pickle
+ curs = self.conn.cursor(cursor_factory=psycopg2.extras.DictCursor)
+ curs.execute("select 10 as a, 20 as b")
+ r = curs.fetchone()
+ d = pickle.dumps(r)
+ r1 = pickle.loads(d)
+ self.assertEqual(r, r1)
+ self.assertEqual(r[0], r1[0])
+ self.assertEqual(r[1], r1[1])
+ self.assertEqual(r['a'], r1['a'])
+ self.assertEqual(r['b'], r1['b'])
+ self.assertEqual(r._index, r1._index)
+
+ @skip_from_python(3)
+ def test_iter_methods_2(self):
+ curs = self.conn.cursor(cursor_factory=psycopg2.extras.DictCursor)
+ curs.execute("select 10 as a, 20 as b")
+ r = curs.fetchone()
+ self.assert_(isinstance(r.keys(), list))
+ self.assertEqual(len(r.keys()), 2)
+ self.assert_(isinstance(r.values(), tuple)) # sic?
+ self.assertEqual(len(r.values()), 2)
+ self.assert_(isinstance(r.items(), list))
+ self.assertEqual(len(r.items()), 2)
+
+ self.assert_(not isinstance(r.iterkeys(), list))
+ self.assertEqual(len(list(r.iterkeys())), 2)
+ self.assert_(not isinstance(r.itervalues(), list))
+ self.assertEqual(len(list(r.itervalues())), 2)
+ self.assert_(not isinstance(r.iteritems(), list))
+ self.assertEqual(len(list(r.iteritems())), 2)
+
+ @skip_before_python(3)
+ def test_iter_methods_3(self):
+ curs = self.conn.cursor(cursor_factory=psycopg2.extras.DictCursor)
+ curs.execute("select 10 as a, 20 as b")
+ r = curs.fetchone()
+ self.assert_(not isinstance(r.keys(), list))
+ self.assertEqual(len(list(r.keys())), 2)
+ self.assert_(not isinstance(r.values(), list))
+ self.assertEqual(len(list(r.values())), 2)
+ self.assert_(not isinstance(r.items(), list))
+ self.assertEqual(len(list(r.items())), 2)
+
+
+class ExtrasDictCursorRealTests(_DictCursorBase):
+ def testDictCursorWithPlainCursorRealFetchOne(self):
+ self._testWithPlainCursorReal(lambda curs: curs.fetchone())
+
+ def testDictCursorWithPlainCursorRealFetchMany(self):
+ self._testWithPlainCursorReal(lambda curs: curs.fetchmany(100)[0])
+
+ def testDictCursorWithPlainCursorRealFetchManyNoarg(self):
+ self._testWithPlainCursorReal(lambda curs: curs.fetchmany()[0])
+
+ def testDictCursorWithPlainCursorRealFetchAll(self):
+ self._testWithPlainCursorReal(lambda curs: curs.fetchall()[0])
+
+ def testDictCursorWithPlainCursorRealIter(self):
+ def getter(curs):
+ for row in curs:
+ return row
+ self._testWithPlainCursorReal(getter)
+
+ @skip_before_postgres(8, 0)
+ def testDictCursorWithPlainCursorRealIterRowNumber(self):
+ curs = self.conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor)
+ self._testIterRowNumber(curs)
+
+ def _testWithPlainCursorReal(self, getter):
+ curs = self.conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor)
+ curs.execute("SELECT * FROM ExtrasDictCursorTests")
+ row = getter(curs)
+ self.failUnless(row['foo'] == 'bar')
+
+ def testPickleRealDictRow(self):
+ import pickle
+ curs = self.conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor)
+ curs.execute("select 10 as a, 20 as b")
+ r = curs.fetchone()
+ d = pickle.dumps(r)
+ r1 = pickle.loads(d)
+ self.assertEqual(r, r1)
+ self.assertEqual(r['a'], r1['a'])
+ self.assertEqual(r['b'], r1['b'])
+ self.assertEqual(r._column_mapping, r1._column_mapping)
+
def testDictCursorRealWithNamedCursorFetchOne(self):
self._testWithNamedCursorReal(lambda curs: curs.fetchone())
@@ -180,51 +262,36 @@ class ExtrasDictCursorTests(ConnectingTestCase):
row = getter(curs)
self.failUnless(row['foo'] == 'bar')
- def _testNamedCursorNotGreedy(self, curs):
- curs.itersize = 2
- curs.execute("""select clock_timestamp() as ts from generate_series(1,3)""")
- recs = []
- for t in curs:
- time.sleep(0.01)
- recs.append(t)
-
- # check that the dataset was not fetched in a single gulp
- self.assert_(recs[1]['ts'] - recs[0]['ts'] < timedelta(seconds=0.005))
- self.assert_(recs[2]['ts'] - recs[1]['ts'] > timedelta(seconds=0.0099))
-
- def _testIterRowNumber(self, curs):
- # Only checking for dataset < itersize:
- # see CursorTests.test_iter_named_cursor_rownumber
- curs.itersize = 20
- curs.execute("""select * from generate_series(1,10)""")
- for i, r in enumerate(curs):
- self.assertEqual(i + 1, curs.rownumber)
-
- def testPickleDictRow(self):
- import pickle
- curs = self.conn.cursor(cursor_factory=psycopg2.extras.DictCursor)
+ @skip_from_python(3)
+ def test_iter_methods_2(self):
+ curs = self.conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor)
curs.execute("select 10 as a, 20 as b")
r = curs.fetchone()
- d = pickle.dumps(r)
- r1 = pickle.loads(d)
- self.assertEqual(r, r1)
- self.assertEqual(r[0], r1[0])
- self.assertEqual(r[1], r1[1])
- self.assertEqual(r['a'], r1['a'])
- self.assertEqual(r['b'], r1['b'])
- self.assertEqual(r._index, r1._index)
-
- def testPickleRealDictRow(self):
- import pickle
+ self.assert_(isinstance(r.keys(), list))
+ self.assertEqual(len(r.keys()), 2)
+ self.assert_(isinstance(r.values(), list))
+ self.assertEqual(len(r.values()), 2)
+ self.assert_(isinstance(r.items(), list))
+ self.assertEqual(len(r.items()), 2)
+
+ self.assert_(not isinstance(r.iterkeys(), list))
+ self.assertEqual(len(list(r.iterkeys())), 2)
+ self.assert_(not isinstance(r.itervalues(), list))
+ self.assertEqual(len(list(r.itervalues())), 2)
+ self.assert_(not isinstance(r.iteritems(), list))
+ self.assertEqual(len(list(r.iteritems())), 2)
+
+ @skip_before_python(3)
+ def test_iter_methods_3(self):
curs = self.conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor)
curs.execute("select 10 as a, 20 as b")
r = curs.fetchone()
- d = pickle.dumps(r)
- r1 = pickle.loads(d)
- self.assertEqual(r, r1)
- self.assertEqual(r['a'], r1['a'])
- self.assertEqual(r['b'], r1['b'])
- self.assertEqual(r._column_mapping, r1._column_mapping)
+ self.assert_(not isinstance(r.keys(), list))
+ self.assertEqual(len(list(r.keys())), 2)
+ self.assert_(not isinstance(r.values(), list))
+ self.assertEqual(len(list(r.values())), 2)
+ self.assert_(not isinstance(r.items(), list))
+ self.assertEqual(len(list(r.items())), 2)
class NamedTupleCursorTest(ConnectingTestCase):
@@ -349,6 +416,22 @@ class NamedTupleCursorTest(ConnectingTestCase):
curs.execute("update nttest set s = s")
self.assertRaises(psycopg2.ProgrammingError, curs.fetchall)
+ def test_bad_col_names(self):
+ curs = self.conn.cursor()
+ curs.execute('select 1 as "foo.bar_baz", 2 as "?column?", 3 as "3"')
+ rv = curs.fetchone()
+ self.assertEqual(rv.foo_bar_baz, 1)
+ self.assertEqual(rv.f_column_, 2)
+ self.assertEqual(rv.f3, 3)
+
+ @skip_before_python(3)
+ @skip_before_postgres(8)
+ def test_nonascii_name(self):
+ curs = self.conn.cursor()
+ curs.execute('select 1 as \xe5h\xe9')
+ rv = curs.fetchone()
+ self.assertEqual(getattr(rv, '\xe5h\xe9'), 1)
+
def test_minimal_generation(self):
# Instrument the class to verify it gets called the minimum number of times.
from psycopg2.extras import NamedTupleCursor
diff --git a/tests/test_types_basic.py b/tests/test_types_basic.py
index a93265d..76b9aa3 100755
--- a/tests/test_types_basic.py
+++ b/tests/test_types_basic.py
@@ -224,16 +224,31 @@ class TypesBasicTests(ConnectingTestCase):
curs.execute("insert into na (boola) values (%s)", ([True, None],))
curs.execute("insert into na (boola) values (%s)", ([None, None],))
- # TODO: array of array of nulls are not supported yet
- # curs.execute("insert into na (textaa) values (%s)", ([[None]],))
+ curs.execute("insert into na (textaa) values (%s)", ([[None]],))
curs.execute("insert into na (textaa) values (%s)", ([['a', None]],))
- # curs.execute("insert into na (textaa) values (%s)", ([[None, None]],))
- # curs.execute("insert into na (intaa) values (%s)", ([[None]],))
+ curs.execute("insert into na (textaa) values (%s)", ([[None, None]],))
+
+ curs.execute("insert into na (intaa) values (%s)", ([[None]],))
curs.execute("insert into na (intaa) values (%s)", ([[42, None]],))
- # curs.execute("insert into na (intaa) values (%s)", ([[None, None]],))
- # curs.execute("insert into na (boolaa) values (%s)", ([[None]],))
+ curs.execute("insert into na (intaa) values (%s)", ([[None, None]],))
+
+ curs.execute("insert into na (boolaa) values (%s)", ([[None]],))
curs.execute("insert into na (boolaa) values (%s)", ([[True, None]],))
- # curs.execute("insert into na (boolaa) values (%s)", ([[None, None]],))
+ curs.execute("insert into na (boolaa) values (%s)", ([[None, None]],))
+
+ @testutils.skip_before_postgres(8, 2)
+ def testNestedArrays(self):
+ curs = self.conn.cursor()
+ for a in [
+ [[1]],
+ [[None]],
+ [[None, None, None]],
+ [[None, None], [1, None]],
+ [[None, None], [None, None]],
+ [[[None, None], [None, None]]],
+ ]:
+ curs.execute("select %s::int[]", (a,))
+ self.assertEqual(curs.fetchone()[0], a)
@testutils.skip_from_python(3)
def testTypeRoundtripBuffer(self):
diff --git a/tests/test_types_extras.py b/tests/test_types_extras.py
index 5cb1353..cda163b 100755
--- a/tests/test_types_extras.py
+++ b/tests/test_types_extras.py
@@ -179,8 +179,8 @@ class HstoreTestCase(ConnectingTestCase):
m = re.match(br'hstore\(ARRAY\[([^\]]+)\], ARRAY\[([^\]]+)\]\)', q)
self.assert_(m, repr(q))
- kk = m.group(1).split(b", ")
- vv = m.group(2).split(b", ")
+ kk = m.group(1).split(b",")
+ vv = m.group(2).split(b",")
ii = list(zip(kk, vv))
ii.sort()
diff --git a/tests/test_with.py b/tests/test_with.py
index 1392d85..f26f8f9 100755
--- a/tests/test_with.py
+++ b/tests/test_with.py
@@ -26,7 +26,7 @@ import psycopg2
import psycopg2.extensions as ext
import unittest
-from .testutils import ConnectingTestCase
+from .testutils import ConnectingTestCase, skip_before_postgres
class WithTestCase(ConnectingTestCase):
@@ -215,6 +215,11 @@ class WithCursorTestCase(WithTestCase):
else:
self.fail("where is my exception?")
+ @skip_before_postgres(8, 0)
+ def test_named_with_noop(self):
+ with self.conn.cursor('named') as cur:
+ pass
+
def test_suite():
return unittest.TestLoader().loadTestsFromName(__name__)