diff options
Diffstat (limited to 'test/engine/test_parseconnect.py')
| -rw-r--r-- | test/engine/test_parseconnect.py | 230 |
1 files changed, 230 insertions, 0 deletions
diff --git a/test/engine/test_parseconnect.py b/test/engine/test_parseconnect.py new file mode 100644 index 000000000..6b7ac37b2 --- /dev/null +++ b/test/engine/test_parseconnect.py @@ -0,0 +1,230 @@ +import ConfigParser, StringIO +import sqlalchemy.engine.url as url +from sqlalchemy import create_engine, engine_from_config +import sqlalchemy as tsa +from sqlalchemy.test import TestBase + + +class ParseConnectTest(TestBase): + def test_rfc1738(self): + for text in ( + 'dbtype://username:password@hostspec:110//usr/db_file.db', + 'dbtype://username:password@hostspec/database', + 'dbtype://username:password@hostspec', + 'dbtype://username:password@/database', + 'dbtype://username@hostspec', + 'dbtype://username:password@127.0.0.1:1521', + 'dbtype://hostspec/database', + 'dbtype://hostspec', + 'dbtype://hostspec/?arg1=val1&arg2=val2', + 'dbtype:///database', + 'dbtype:///:memory:', + 'dbtype:///foo/bar/im/a/file', + 'dbtype:///E:/work/src/LEM/db/hello.db', + 'dbtype:///E:/work/src/LEM/db/hello.db?foo=bar&hoho=lala', + 'dbtype://', + 'dbtype://username:password@/db', + 'dbtype:////usr/local/mailman/lists/_xtest@example.com/members.db', + 'dbtype://username:apples%2Foranges@hostspec/mydatabase', + ): + u = url.make_url(text) + print u, text + print "username=", u.username, "password=", u.password, "database=", u.database, "host=", u.host + assert u.drivername == 'dbtype' + assert u.username == 'username' or u.username is None + assert u.password == 'password' or u.password == 'apples/oranges' or u.password is None + assert u.host == 'hostspec' or u.host == '127.0.0.1' or (not u.host) + assert str(u) == text + +class CreateEngineTest(TestBase): + """test that create_engine arguments of different types get propagated properly""" + def test_connect_query(self): + dbapi = MockDBAPI(foober='12', lala='18', fooz='somevalue') + + # start the postgres dialect, but put our mock DBAPI as the module instead of psycopg + e = create_engine('postgres://scott:tiger@somehost/test?foober=12&lala=18&fooz=somevalue', module=dbapi) + c = e.connect() + + def test_kwargs(self): + dbapi = MockDBAPI(foober=12, lala=18, hoho={'this':'dict'}, fooz='somevalue') + + # start the postgres dialect, but put our mock DBAPI as the module instead of psycopg + e = create_engine('postgres://scott:tiger@somehost/test?fooz=somevalue', connect_args={'foober':12, 'lala':18, 'hoho':{'this':'dict'}}, module=dbapi) + c = e.connect() + + def test_coerce_config(self): + raw = r""" +[prefixed] +sqlalchemy.url=postgres://scott:tiger@somehost/test?fooz=somevalue +sqlalchemy.convert_unicode=0 +sqlalchemy.echo=false +sqlalchemy.echo_pool=1 +sqlalchemy.max_overflow=2 +sqlalchemy.pool_recycle=50 +sqlalchemy.pool_size=2 +sqlalchemy.pool_threadlocal=1 +sqlalchemy.pool_timeout=10 +[plain] +url=postgres://scott:tiger@somehost/test?fooz=somevalue +convert_unicode=0 +echo=0 +echo_pool=1 +max_overflow=2 +pool_recycle=50 +pool_size=2 +pool_threadlocal=1 +pool_timeout=10 +""" + ini = ConfigParser.ConfigParser() + ini.readfp(StringIO.StringIO(raw)) + + expected = { + 'url': 'postgres://scott:tiger@somehost/test?fooz=somevalue', + 'convert_unicode': 0, + 'echo': False, + 'echo_pool': True, + 'max_overflow': 2, + 'pool_recycle': 50, + 'pool_size': 2, + 'pool_threadlocal': True, + 'pool_timeout': 10, + } + + prefixed = dict(ini.items('prefixed')) + self.assert_(tsa.engine._coerce_config(prefixed, 'sqlalchemy.') == expected) + + plain = dict(ini.items('plain')) + self.assert_(tsa.engine._coerce_config(plain, '') == expected) + + def test_engine_from_config(self): + dbapi = MockDBAPI() + + config = { + 'sqlalchemy.url':'postgres://scott:tiger@somehost/test?fooz=somevalue', + 'sqlalchemy.pool_recycle':'50', + 'sqlalchemy.echo':'true' + } + + e = engine_from_config(config, module=dbapi) + assert e.pool._recycle == 50 + assert e.url == url.make_url('postgres://scott:tiger@somehost/test?fooz=somevalue') + assert e.echo is True + + def test_custom(self): + dbapi = MockDBAPI(foober=12, lala=18, hoho={'this':'dict'}, fooz='somevalue') + + def connect(): + return dbapi.connect(foober=12, lala=18, fooz='somevalue', hoho={'this':'dict'}) + + # start the postgres dialect, but put our mock DBAPI as the module instead of psycopg + e = create_engine('postgres://', creator=connect, module=dbapi) + c = e.connect() + + def test_recycle(self): + dbapi = MockDBAPI(foober=12, lala=18, hoho={'this':'dict'}, fooz='somevalue') + e = create_engine('postgres://', pool_recycle=472, module=dbapi) + assert e.pool._recycle == 472 + + def test_badargs(self): + # good arg, use MockDBAPI to prevent oracle import errors + e = create_engine('oracle://', use_ansi=True, module=MockDBAPI()) + + try: + e = create_engine("foobar://", module=MockDBAPI()) + assert False + except ImportError: + assert True + + # 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, module=MockDBAPI()) + 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('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 tsa.exc.DBAPIError: + assert True + + def test_urlattr(self): + """test the url attribute on ``Engine``.""" + + e = create_engine('mysql://scott:tiger@localhost/test', module=MockDBAPI()) + u = url.make_url('mysql://scott:tiger@localhost/test') + e2 = create_engine(u, module=MockDBAPI()) + assert e.url.drivername == e2.url.drivername == 'mysql' + assert e.url.username == e2.url.username == 'scott' + assert e2.url is u + + def test_poolargs(self): + """test that connection pool args make it thru""" + e = create_engine('postgres://', creator=None, pool_recycle=50, echo_pool=None, module=MockDBAPI()) + assert e.pool._recycle == 50 + + # these args work for QueuePool + e = create_engine('postgres://', max_overflow=8, pool_timeout=60, poolclass=tsa.pool.QueuePool, module=MockDBAPI()) + + try: + # but not SingletonThreadPool + e = create_engine('sqlite://', max_overflow=8, pool_timeout=60, poolclass=tsa.pool.SingletonThreadPool) + assert False + except TypeError: + assert True + +class MockDBAPI(object): + def __init__(self, **kwargs): + self.kwargs = kwargs + self.paramstyle = 'named' + def connect(self, **kwargs): + print kwargs, self.kwargs + for k in self.kwargs: + assert k in kwargs, "key %s not present in dictionary" % k + assert kwargs[k]==self.kwargs[k], "value %s does not match %s" % (kwargs[k], self.kwargs[k]) + return MockConnection() +class MockConnection(object): + def close(self): + pass + def cursor(self): + return MockCursor() +class MockCursor(object): + def close(self): + pass +mock_dbapi = MockDBAPI() + |