summaryrefslogtreecommitdiff
path: root/test/engine/test_parseconnect.py
diff options
context:
space:
mode:
authorMike Bayer <mike_mp@zzzcomputing.com>2010-07-11 13:15:51 -0400
committerMike Bayer <mike_mp@zzzcomputing.com>2010-07-11 13:15:51 -0400
commita6b62cc3fed5f06d3428b1f6ee13756175ded61b (patch)
tree6483de311229fae0b40e9caf36b579b4f1c58ff2 /test/engine/test_parseconnect.py
parent5cce6bf2a86cb318d80953cd3712fcb77bbc52db (diff)
downloadsqlalchemy-a6b62cc3fed5f06d3428b1f6ee13756175ded61b.tar.gz
Python-tidy test/engine and test/aaa_profiling, 80% auto + 20% manual intervention
Diffstat (limited to 'test/engine/test_parseconnect.py')
-rw-r--r--test/engine/test_parseconnect.py182
1 files changed, 122 insertions, 60 deletions
diff --git a/test/engine/test_parseconnect.py b/test/engine/test_parseconnect.py
index c00a83136..7fb8d8a90 100644
--- a/test/engine/test_parseconnect.py
+++ b/test/engine/test_parseconnect.py
@@ -26,46 +26,56 @@ class ParseConnectTest(TestBase):
'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:////usr/local/mailman/lists/_xtest@example.com/memb'
+ 'ers.db',
'dbtype://username:apples%2Foranges@hostspec/mydatabase',
- ):
+ ):
u = url.make_url(text)
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 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 DialectImportTest(TestBase):
def test_import_base_dialects(self):
+
# the globals() somehow makes it for the exec() + nose3.
- for name in ('mysql', 'firebird', 'postgresql', 'sqlite', 'oracle', 'mssql'):
- exec("from sqlalchemy.dialects import %s\n"
- "dialect = %s.dialect()"
- % (name, name), globals())
+
+ for name in (
+ 'mysql',
+ 'firebird',
+ 'postgresql',
+ 'sqlite',
+ 'oracle',
+ 'mssql',
+ ):
+ exec ('from sqlalchemy.dialects import %s\ndialect = '
+ '%s.dialect()' % (name, name), globals())
eq_(dialect.name, name)
class CreateEngineTest(TestBase):
- """test that create_engine arguments of different types get propagated properly"""
+ """test that create_engine arguments of different types get
+ propagated properly"""
+
def test_connect_query(self):
dbapi = MockDBAPI(foober='12', lala='18', fooz='somevalue')
-
- e = create_engine(
- 'postgresql://scott:tiger@somehost/test?foober=12&lala=18&fooz=somevalue',
- module=dbapi,
- _initialize=False
- )
+ e = \
+ create_engine('postgresql://scott:tiger@somehost/test?foobe'
+ 'r=12&lala=18&fooz=somevalue', module=dbapi,
+ _initialize=False)
c = e.connect()
def test_kwargs(self):
- dbapi = MockDBAPI(foober=12, lala=18, hoho={'this':'dict'}, fooz='somevalue')
-
- e = create_engine(
- 'postgresql://scott:tiger@somehost/test?fooz=somevalue',
- connect_args={'foober':12, 'lala':18, 'hoho':{'this':'dict'}},
- module=dbapi,
- _initialize=False
- )
+ dbapi = MockDBAPI(foober=12, lala=18, hoho={'this': 'dict'},
+ fooz='somevalue')
+ e = \
+ create_engine('postgresql://scott:tiger@somehost/test?fooz='
+ 'somevalue', connect_args={'foober': 12,
+ 'lala': 18, 'hoho': {'this': 'dict'}},
+ module=dbapi, _initialize=False)
c = e.connect()
def test_coerce_config(self):
@@ -107,7 +117,8 @@ pool_timeout=10
}
prefixed = dict(ini.items('prefixed'))
- self.assert_(tsa.engine._coerce_config(prefixed, 'sqlalchemy.') == expected)
+ self.assert_(tsa.engine._coerce_config(prefixed, 'sqlalchemy.')
+ == expected)
plain = dict(ini.items('plain'))
self.assert_(tsa.engine._coerce_config(plain, '') == expected)
@@ -115,53 +126,73 @@ pool_timeout=10
def test_engine_from_config(self):
dbapi = mock_dbapi
- config = {
- 'sqlalchemy.url':'postgresql://scott:tiger@somehost/test?fooz=somevalue',
- 'sqlalchemy.pool_recycle':'50',
- 'sqlalchemy.echo':'true'
- }
+ config = \
+ {'sqlalchemy.url': 'postgresql://scott:tiger@somehost/test'\
+ '?fooz=somevalue', 'sqlalchemy.pool_recycle': '50',
+ 'sqlalchemy.echo': 'true'}
e = engine_from_config(config, module=dbapi, _initialize=False)
assert e.pool._recycle == 50
- assert e.url == url.make_url('postgresql://scott:tiger@somehost/test?fooz=somevalue')
+ assert e.url \
+ == url.make_url('postgresql://scott:tiger@somehost/test?foo'
+ 'z=somevalue')
assert e.echo is True
def test_custom(self):
- dbapi = MockDBAPI(foober=12, lala=18, hoho={'this':'dict'}, fooz='somevalue')
+ 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'})
+ return dbapi.connect(foober=12, lala=18, fooz='somevalue',
+ hoho={'this': 'dict'})
- # start the postgresql dialect, but put our mock DBAPI as the module instead of psycopg
- e = create_engine('postgresql://', creator=connect, module=dbapi, _initialize=False)
+ # start the postgresql dialect, but put our mock DBAPI as the
+ # module instead of psycopg
+
+ e = create_engine('postgresql://', creator=connect,
+ module=dbapi, _initialize=False)
c = e.connect()
def test_recycle(self):
- dbapi = MockDBAPI(foober=12, lala=18, hoho={'this':'dict'}, fooz='somevalue')
- e = create_engine('postgresql://', pool_recycle=472, module=dbapi, _initialize=False)
+ dbapi = MockDBAPI(foober=12, lala=18, hoho={'this': 'dict'},
+ fooz='somevalue')
+ e = create_engine('postgresql://', pool_recycle=472,
+ module=dbapi, _initialize=False)
assert e.pool._recycle == 472
def test_badargs(self):
- assert_raises(ImportError, create_engine, "foobar://", module=mock_dbapi)
-
- # bad arg
- assert_raises(TypeError, create_engine, 'postgresql://', use_ansi=True, module=mock_dbapi)
+ assert_raises(ImportError, create_engine, 'foobar://',
+ module=mock_dbapi)
# bad arg
- assert_raises(TypeError, create_engine, 'oracle://', lala=5, use_ansi=True, module=mock_dbapi)
- assert_raises(TypeError, create_engine, 'postgresql://', lala=5, module=mock_dbapi)
+ assert_raises(TypeError, create_engine, 'postgresql://',
+ use_ansi=True, module=mock_dbapi)
- assert_raises(TypeError, create_engine,'sqlite://', lala=5, module=mock_sqlite_dbapi)
+ # bad arg
- assert_raises(TypeError, create_engine, 'mysql+mysqldb://', use_unicode=True, module=mock_dbapi)
+ assert_raises(
+ TypeError,
+ create_engine,
+ 'oracle://',
+ lala=5,
+ use_ansi=True,
+ module=mock_dbapi,
+ )
+ assert_raises(TypeError, create_engine, 'postgresql://',
+ lala=5, module=mock_dbapi)
+ assert_raises(TypeError, create_engine, 'sqlite://', lala=5,
+ module=mock_sqlite_dbapi)
+ assert_raises(TypeError, create_engine, 'mysql+mysqldb://',
+ use_unicode=True, module=mock_dbapi)
# sqlite uses SingletonThreadPool which doesnt have max_overflow
- assert_raises(TypeError, create_engine, 'sqlite://', max_overflow=5,
- module=mock_sqlite_dbapi)
+ assert_raises(TypeError, create_engine, 'sqlite://',
+ max_overflow=5, module=mock_sqlite_dbapi)
try:
- e = create_engine('sqlite://', connect_args={'use_unicode':True}, convert_unicode=True)
+ e = create_engine('sqlite://', connect_args={'use_unicode'
+ : True}, convert_unicode=True)
except ImportError:
# no sqlite
pass
@@ -172,7 +203,8 @@ pool_timeout=10
def test_urlattr(self):
"""test the url attribute on ``Engine``."""
- e = create_engine('mysql://scott:tiger@localhost/test', module=mock_dbapi, _initialize=False)
+ e = create_engine('mysql://scott:tiger@localhost/test',
+ module=mock_dbapi, _initialize=False)
u = url.make_url('mysql://scott:tiger@localhost/test')
e2 = create_engine(u, module=mock_dbapi, _initialize=False)
assert e.url.drivername == e2.url.drivername == 'mysql'
@@ -181,40 +213,70 @@ pool_timeout=10
def test_poolargs(self):
"""test that connection pool args make it thru"""
- e = create_engine('postgresql://', creator=None, pool_recycle=50, echo_pool=None, module=mock_dbapi, _initialize=False)
+
+ e = create_engine(
+ 'postgresql://',
+ creator=None,
+ pool_recycle=50,
+ echo_pool=None,
+ module=mock_dbapi,
+ _initialize=False,
+ )
assert e.pool._recycle == 50
# these args work for QueuePool
- e = create_engine('postgresql://',
- max_overflow=8, pool_timeout=60,
- poolclass=tsa.pool.QueuePool, module=mock_dbapi,
- _initialize=False)
+
+ e = create_engine(
+ 'postgresql://',
+ max_overflow=8,
+ pool_timeout=60,
+ poolclass=tsa.pool.QueuePool,
+ module=mock_dbapi,
+ _initialize=False,
+ )
# but not SingletonThreadPool
- assert_raises(TypeError, create_engine, 'sqlite://', max_overflow=8, pool_timeout=60,
- poolclass=tsa.pool.SingletonThreadPool, module=mock_sqlite_dbapi,
- _initialize=False)
+
+ assert_raises(
+ TypeError,
+ create_engine,
+ 'sqlite://',
+ max_overflow=8,
+ pool_timeout=60,
+ poolclass=tsa.pool.SingletonThreadPool,
+ module=mock_sqlite_dbapi,
+ _initialize=False,
+ )
class MockDBAPI(object):
def __init__(self, **kwargs):
self.kwargs = kwargs
self.paramstyle = 'named'
+
def connect(self, *args, **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])
+ 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 get_server_info(self):
- return "5.0"
+ return '5.0'
+
def close(self):
pass
+
def cursor(self):
return MockCursor()
+
class MockCursor(object):
def close(self):
pass
+
mock_dbapi = MockDBAPI()
mock_sqlite_dbapi = msd = MockDBAPI()
-msd.version_info = msd.sqlite_version_info = (99, 9, 9)
+msd.version_info = msd.sqlite_version_info = 99, 9, 9
msd.sqlite_version = '99.9.9'