diff options
Diffstat (limited to 'tests/test_connection.py')
| -rwxr-xr-x | tests/test_connection.py | 120 |
1 files changed, 66 insertions, 54 deletions
diff --git a/tests/test_connection.py b/tests/test_connection.py index 13635f1..f9fdff6 100755 --- a/tests/test_connection.py +++ b/tests/test_connection.py @@ -35,13 +35,11 @@ import psycopg2.errorcodes 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, - libpq_version -) + unittest, decorate_all_tests, skip_if_no_superuser, skip_before_postgres, + skip_after_postgres, skip_before_libpq, skip_after_libpq, + ConnectingTestCase, skip_if_tpc_disabled, skip_if_windows, slow) -from .testconfig import dsn, dbname +from .testconfig import dbhost, dsn, dbname class ConnectionTests(ConnectingTestCase): @@ -805,6 +803,14 @@ class IsolationLevelsTestCase(ConnectingTestCase): self.assertRaises(ValueError, setattr, self.conn, 'isolation_level', 5) self.assertRaises(ValueError, setattr, self.conn, 'isolation_level', 'bah') + def test_attribs_segfault(self): + # bug #790 + for i in range(10000): + self.conn.autocommit + self.conn.readonly + self.conn.deferrable + self.conn.isolation_level + class ConnectionTwoPhaseTests(ConnectingTestCase): def setUp(self): @@ -1411,55 +1417,37 @@ class TransactionControlTests(ConnectingTestCase): 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' - ) + '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$') + 'md594839d658c28a357126f105b9cb14cfc') - self.assertEqual( - ext.encrypt_password( - 'psycopg2', 'ashesh', self.conn, 'scram-sha-256' - )[:14], 'SCRAM-SHA-256$' - ) + @skip_before_postgres(10) + def test_encrypt_server(self): + cur = self.conn.cursor() + cur.execute("SHOW password_encryption;") + server_encryption_algorithm = cur.fetchone()[0] - self.assertRaises(psycopg2.ProgrammingError, - ext.encrypt_password, 'psycopg2', 'ashesh', self.conn, 'abc') + 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$') - @skip_after_postgres(10) - def test_encrypt_password_pre_10(self): self.assertEqual( - ext.encrypt_password('psycopg2', 'ashesh', self.conn), - 'md594839d658c28a357126f105b9cb14cfc' - ) + 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') @@ -1467,20 +1455,31 @@ class TestEncryptPassword(ConnectingTestCase): def test_encrypt_md5(self): self.assertEqual( ext.encrypt_password('psycopg2', 'ashesh', algorithm='md5'), - 'md594839d658c28a357126f105b9cb14cfc' - ) + 'md594839d658c28a357126f105b9cb14cfc') + @skip_before_libpq(10) + def test_encrypt_bad_libpq_10(self): + self.assertRaises(psycopg2.ProgrammingError, + ext.encrypt_password, 'psycopg2', 'ashesh', self.conn, 'abc') + + @skip_after_libpq(10) + def test_encrypt_bad_before_libpq_10(self): + self.assertRaises(psycopg2.NotSupportedError, + ext.encrypt_password, 'psycopg2', 'ashesh', self.conn, 'abc') + + @skip_before_libpq(10) 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') + self.assert_( + ext.encrypt_password( + 'psycopg2', 'ashesh', self.conn, 'scram-sha-256') + .startswith('SCRAM-SHA-256$')) + + @skip_after_libpq(10) + def test_encrypt_scram_pre_10(self): + 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) @@ -1691,6 +1690,19 @@ while True: self.assert_(not err, err) +class TestConnectionProps(ConnectingTestCase): + def test_host(self): + self.assertFalse(self.conn.closed) + expected = dbhost if dbhost else "/" + self.assertIn(expected, self.conn.host) + + def test_host_readonly(self): + self.assertFalse(self.conn.closed) + with self.assertRaises(AttributeError): + self.conn.host = 'override' + + + def test_suite(): return unittest.TestLoader().loadTestsFromName(__name__) |