diff options
| author | Daniele Varrazzo <daniele.varrazzo@gmail.com> | 2018-05-20 23:56:29 +0100 |
|---|---|---|
| committer | Daniele Varrazzo <daniele.varrazzo@gmail.com> | 2018-05-20 23:56:29 +0100 |
| commit | e8a831dda2decd26f57bb227948da8af9237fb89 (patch) | |
| tree | 1395c3acd39c74989c0a891e57f01862e8cb698d /tests | |
| parent | be3b1ba1eb84244e97a9d85ba05ef2ca6b5a213b (diff) | |
| parent | a0f7027ad578c6c2707430e9a208cc3f7510cad4 (diff) | |
| download | psycopg2-e8a831dda2decd26f57bb227948da8af9237fb89.tar.gz | |
Merge branch 'master' into drop-2to3
Diffstat (limited to 'tests')
| -rwxr-xr-x | tests/test_connection.py | 120 | ||||
| -rwxr-xr-x | tests/test_cursor.py | 11 | ||||
| -rwxr-xr-x | tests/test_dates.py | 11 | ||||
| -rwxr-xr-x | tests/test_extras_dictcursor.py | 229 | ||||
| -rwxr-xr-x | tests/test_types_basic.py | 29 | ||||
| -rwxr-xr-x | tests/test_types_extras.py | 4 | ||||
| -rwxr-xr-x | tests/test_with.py | 7 |
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__) |
