psycopg/psycopg2
1
1
Fork 0
mirror of https://github.com/psycopg/psycopg2.git synced 2025-02-19 18:30:33 +03:00
Code Issues Packages Projects Releases Wiki Activity

Added test suite specific for make_dsn

Browse Source
This commit is contained in:
Daniele Varrazzo 2016-03-03 15:31:37 +00:00
parent 2c55a1bd53
commit 52087a79d9
1 changed files with 82 additions and 35 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

103
tests/test_connection.py
View File

@ -32,6 +32,7 @@ from StringIO import StringIO
import psycopg2 import psycopg2
import psycopg2.errorcodes import psycopg2.errorcodes
import psycopg2.extensions import psycopg2.extensions
ext = psycopg2.extensions
from testutils import unittest, decorate_all_tests, skip_if_no_superuser from testutils import unittest, decorate_all_tests, skip_if_no_superuser
from testutils import skip_before_postgres, skip_after_postgres, skip_before_libpq from testutils import skip_before_postgres, skip_after_postgres, skip_before_libpq
@ -313,16 +314,15 @@ class ConnectionTests(ConnectingTestCase):
class ParseDsnTestCase(ConnectingTestCase): class ParseDsnTestCase(ConnectingTestCase):
def test_parse_dsn(self): def test_parse_dsn(self):
from psycopg2 import ProgrammingError from psycopg2 import ProgrammingError
from psycopg2.extensions import parse_dsn
self.assertEqual(parse_dsn('dbname=test user=tester password=secret'), self.assertEqual(ext.parse_dsn('dbname=test user=tester password=secret'),
dict(user='tester', password='secret', dbname='test'), dict(user='tester', password='secret', dbname='test'),
"simple DSN parsed") "simple DSN parsed")
self.assertRaises(ProgrammingError, parse_dsn, self.assertRaises(ProgrammingError, ext.parse_dsn,
"dbname=test 2 user=tester password=secret") "dbname=test 2 user=tester password=secret")
self.assertEqual(parse_dsn("dbname='test 2' user=tester password=secret"), self.assertEqual(ext.parse_dsn("dbname='test 2' user=tester password=secret"),
dict(user='tester', password='secret', dbname='test 2'), dict(user='tester', password='secret', dbname='test 2'),
"DSN with quoting parsed") "DSN with quoting parsed")
@ -332,7 +332,7 @@ class ParseDsnTestCase(ConnectingTestCase):
raised = False raised = False
try: try:
# unterminated quote after dbname: # unterminated quote after dbname:
parse_dsn("dbname='test 2 user=tester password=secret") ext.parse_dsn("dbname='test 2 user=tester password=secret")
except ProgrammingError, e: except ProgrammingError, e:
raised = True raised = True
self.assertTrue(str(e).find('secret') < 0, self.assertTrue(str(e).find('secret') < 0,
@ -343,16 +343,14 @@ class ParseDsnTestCase(ConnectingTestCase):
@skip_before_libpq(9, 2) @skip_before_libpq(9, 2)
def test_parse_dsn_uri(self): def test_parse_dsn_uri(self):
from psycopg2.extensions import parse_dsn self.assertEqual(ext.parse_dsn('postgresql://tester:secret@/test'),
self.assertEqual(parse_dsn('postgresql://tester:secret@/test'),
dict(user='tester', password='secret', dbname='test'), dict(user='tester', password='secret', dbname='test'),
"valid URI dsn parsed") "valid URI dsn parsed")
raised = False raised = False
try: try:
# extra '=' after port value # extra '=' after port value
parse_dsn(dsn='postgresql://tester:secret@/test?port=1111=x') ext.parse_dsn(dsn='postgresql://tester:secret@/test?port=1111=x')
except psycopg2.ProgrammingError, e: except psycopg2.ProgrammingError, e:
raised = True raised = True
self.assertTrue(str(e).find('secret') < 0, self.assertTrue(str(e).find('secret') < 0,
@ -362,24 +360,76 @@ class ParseDsnTestCase(ConnectingTestCase):
self.assertTrue(raised, "ProgrammingError raised due to invalid URI") self.assertTrue(raised, "ProgrammingError raised due to invalid URI")
def test_unicode_value(self): def test_unicode_value(self):
from psycopg2.extensions import parse_dsn
snowman = u"\u2603" snowman = u"\u2603"
d = parse_dsn('dbname=' + snowman) d = ext.parse_dsn('dbname=' + snowman)
if sys.version_info[0] < 3: if sys.version_info[0] < 3:
self.assertEqual(d['dbname'], snowman.encode('utf8')) self.assertEqual(d['dbname'], snowman.encode('utf8'))
else: else:
self.assertEqual(d['dbname'], snowman) self.assertEqual(d['dbname'], snowman)
def test_unicode_key(self): def test_unicode_key(self):
from psycopg2.extensions import parse_dsn
snowman = u"\u2603" snowman = u"\u2603"
self.assertRaises(psycopg2.ProgrammingError, parse_dsn, self.assertRaises(psycopg2.ProgrammingError, ext.parse_dsn,
snowman + '=' + snowman) snowman + '=' + snowman)
def test_bad_param(self): def test_bad_param(self):
from psycopg2.extensions import parse_dsn self.assertRaises(TypeError, ext.parse_dsn, None)
self.assertRaises(TypeError, parse_dsn, None) self.assertRaises(TypeError, ext.parse_dsn, 42)
self.assertRaises(TypeError, parse_dsn, 42)
class MakeDsnTestCase(ConnectingTestCase):
def assertDsnEqual(self, dsn1, dsn2):
self.assertEqual(set(dsn1.split()), set(dsn2.split()))
def test_there_has_to_be_something(self):
self.assertRaises(TypeError, ext.make_dsn)
def test_empty_param(self):
dsn = ext.make_dsn(database='sony', password='')
self.assertDsnEqual(dsn, "dbname=sony password=''")
def test_escape(self):
dsn = ext.make_dsn(database='hello world')
self.assertEqual(dsn, "dbname='hello world'")
dsn = ext.make_dsn(database=r'back\slash')
self.assertEqual(dsn, r"dbname=back\\slash")
dsn = ext.make_dsn(database="quo'te")
self.assertEqual(dsn, r"dbname=quo\'te")
dsn = ext.make_dsn(database="with\ttab")
self.assertEqual(dsn, "dbname='with\ttab'")
dsn = ext.make_dsn(database=r"\every thing'")
self.assertEqual(dsn, r"dbname='\\every thing\''")
def test_params_merging(self):
dsn = ext.make_dsn('dbname=foo', database='bar')
self.assertEqual(dsn, 'dbname=bar')
dsn = ext.make_dsn('dbname=foo', user='postgres')
self.assertDsnEqual(dsn, 'dbname=foo user=postgres')
def test_no_dsn_munging(self):
dsn = ext.make_dsn('nosuchparam=whatevs')
self.assertEqual(dsn, 'nosuchparam=whatevs')
dsn = ext.make_dsn(nosuchparam='whatevs')
self.assertEqual(dsn, 'nosuchparam=whatevs')
self.assertRaises(psycopg2.ProgrammingError,
ext.make_dsn, 'nosuchparam=whatevs', andthis='either')
@skip_before_libpq(9, 2)
def test_url_is_cool(self):
dsn = ext.make_dsn('postgresql://tester:secret@/test')
self.assertEqual(dsn, 'postgresql://tester:secret@/test')
dsn = ext.make_dsn('postgresql://tester:secret@/test',
application_name='woot')
self.assertDsnEqual(dsn,
'dbname=test user=tester password=secret application_name=woot')
class IsolationLevelsTestCase(ConnectingTestCase): class IsolationLevelsTestCase(ConnectingTestCase):
@ -761,13 +811,13 @@ class ConnectionTwoPhaseTests(ConnectingTestCase):
def test_status_after_recover(self): def test_status_after_recover(self):
cnn = self.connect() cnn = self.connect()
self.assertEqual(psycopg2.extensions.STATUS_READY, cnn.status) self.assertEqual(psycopg2.extensions.STATUS_READY, cnn.status)
xns = cnn.tpc_recover() cnn.tpc_recover()
self.assertEqual(psycopg2.extensions.STATUS_READY, cnn.status) self.assertEqual(psycopg2.extensions.STATUS_READY, cnn.status)
cur = cnn.cursor() cur = cnn.cursor()
cur.execute("select 1") cur.execute("select 1")
self.assertEqual(psycopg2.extensions.STATUS_BEGIN, cnn.status) self.assertEqual(psycopg2.extensions.STATUS_BEGIN, cnn.status)
xns = cnn.tpc_recover() cnn.tpc_recover()
self.assertEqual(psycopg2.extensions.STATUS_BEGIN, cnn.status) self.assertEqual(psycopg2.extensions.STATUS_BEGIN, cnn.status)
def test_recovered_xids(self): def test_recovered_xids(self):
@ -825,8 +875,7 @@ class ConnectionTwoPhaseTests(ConnectingTestCase):
cnn.close() cnn.close()
cnn = self.connect() cnn = self.connect()
xids = [ xid for xid in cnn.tpc_recover() xids = [x for x in cnn.tpc_recover() if x.database == dbname]
if xid.database == dbname ]
self.assertEqual(1, len(xids)) self.assertEqual(1, len(xids))
xid = xids[0] xid = xids[0]
self.assertEqual(xid.format_id, fid) self.assertEqual(xid.format_id, fid)
@ -847,8 +896,7 @@ class ConnectionTwoPhaseTests(ConnectingTestCase):
cnn.close() cnn.close()
cnn = self.connect() cnn = self.connect()
xids = [ xid for xid in cnn.tpc_recover() xids = [x for x in cnn.tpc_recover() if x.database == dbname]
if xid.database == dbname ]
self.assertEqual(1, len(xids)) self.assertEqual(1, len(xids))
xid = xids[0] xid = xids[0]
self.assertEqual(xid.format_id, None) self.assertEqual(xid.format_id, None)
@ -893,8 +941,7 @@ class ConnectionTwoPhaseTests(ConnectingTestCase):
cnn.tpc_begin(x1) cnn.tpc_begin(x1)
cnn.tpc_prepare() cnn.tpc_prepare()
cnn.reset() cnn.reset()
xid = [ xid for xid in cnn.tpc_recover() xid = [x for x in cnn.tpc_recover() if x.database == dbname][0]
if xid.database == dbname ][0]
self.assertEqual(10, xid.format_id) self.assertEqual(10, xid.format_id)
self.assertEqual('uni', xid.gtrid) self.assertEqual('uni', xid.gtrid)
self.assertEqual('code', xid.bqual) self.assertEqual('code', xid.bqual)
@ -909,8 +956,7 @@ class ConnectionTwoPhaseTests(ConnectingTestCase):
cnn.tpc_prepare() cnn.tpc_prepare()
cnn.reset() cnn.reset()
xid = [ xid for xid in cnn.tpc_recover() xid = [x for x in cnn.tpc_recover() if x.database == dbname][0]
if xid.database == dbname ][0]
self.assertEqual(None, xid.format_id) self.assertEqual(None, xid.format_id)
self.assertEqual('transaction-id', xid.gtrid) self.assertEqual('transaction-id', xid.gtrid)
self.assertEqual(None, xid.bqual) self.assertEqual(None, xid.bqual)
@ -929,7 +975,7 @@ class ConnectionTwoPhaseTests(ConnectingTestCase):
cnn.reset() cnn.reset()
xids = cnn.tpc_recover() xids = cnn.tpc_recover()
xid = [ xid for xid in xids if xid.database == dbname ][0] xid = [x for x in xids if x.database == dbname][0]
self.assertEqual(None, xid.format_id) self.assertEqual(None, xid.format_id)
self.assertEqual('dict-connection', xid.gtrid) self.assertEqual('dict-connection', xid.gtrid)
self.assertEqual(None, xid.bqual) self.assertEqual(None, xid.bqual)
@ -1182,7 +1228,8 @@ class ReplicationTest(ConnectingTestCase):
@skip_before_postgres(9, 0) @skip_before_postgres(9, 0)
def test_replication_not_supported(self): def test_replication_not_supported(self):
conn = self.repl_connect() conn = self.repl_connect()
if conn is None: return if conn is None:
return
cur = conn.cursor() cur = conn.cursor()
f = StringIO() f = StringIO()
self.assertRaises(psycopg2.NotSupportedError, self.assertRaises(psycopg2.NotSupportedError,