From ecee1fb16cf6b7b5d01187191ea23260b8bcef2a Mon Sep 17 00:00:00 2001 From: Mike Bayer Date: Sun, 12 Nov 2006 20:50:51 +0000 Subject: - create_engine() reworked to be strict about incoming **kwargs. all keyword arguments must be consumed by one of the dialect, connection pool, and engine constructors, else a TypeError is thrown which describes the full set of invalid kwargs in relation to the selected dialect/pool/engine configuration. --- test/engine/parseconnect.py | 69 +++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 69 insertions(+) (limited to 'test/engine/parseconnect.py') diff --git a/test/engine/parseconnect.py b/test/engine/parseconnect.py index 9af594a98..251e0e64d 100644 --- a/test/engine/parseconnect.py +++ b/test/engine/parseconnect.py @@ -67,6 +67,75 @@ class CreateEngineTest(PersistTest): e = create_engine('postgres://', pool_recycle=472, module=dbapi) assert e.connection_provider._pool._recycle == 472 + def testbadargs(self): + # good arg, use MockDBAPI to prevent oracle import errors + e = create_engine('oracle://', use_ansi=True, module=MockDBAPI()) + + # bad arg + try: + e = create_engine('postgres://', use_ansi=True, module=MockDBAPI()) + assert False + except TypeError: + assert True + + # bad arg + try: + e = create_engine('oracle://', lala=5, use_ansi=True, module=MockDBAPI()) + assert False + except TypeError: + assert True + + try: + e = create_engine('postgres://', lala=5, module=MockDBAPI()) + assert False + except TypeError: + assert True + + try: + e = create_engine('sqlite://', lala=5) + assert False + except TypeError: + assert True + + try: + e = create_engine('mysql://', use_unicode=True) + assert False + except TypeError: + assert True + + try: + # sqlite uses SingletonThreadPool which doesnt have max_overflow + e = create_engine('sqlite://', max_overflow=5) + assert False + except TypeError: + assert True + + e = create_engine('sqlite://', echo=True) + e = create_engine('mysql://', module=MockDBAPI(), connect_args={'use_unicode':True}, convert_unicode=True) + + e = create_engine('sqlite://', connect_args={'use_unicode':True}, convert_unicode=True) + try: + c = e.connect() + assert False + except exceptions.DBAPIError: + assert True + + def testpoolargs(self): + """test that connection pool args make it thru""" + e = create_engine('postgres://', creator=None, pool_recycle=-1, echo_pool=None, auto_close_cursors=False, disallow_open_cursors=True, module=MockDBAPI()) + assert e.connection_provider._pool.auto_close_cursors is False + assert e.connection_provider._pool.disallow_open_cursors is True + + # these args work for QueuePool + e = create_engine('postgres://', max_overflow=8, pool_timeout=60, poolclass=pool.QueuePool, module=MockDBAPI()) + + try: + # but not SingletonThreadPool + e = create_engine('sqlite://', max_overflow=8, pool_timeout=60, poolclass=pool.SingletonThreadPool) + assert False + except TypeError: + assert True + class MockDBAPI(object): def __init__(self, **kwargs): self.kwargs = kwargs -- cgit v1.2.1