diff options
author | Mike Bayer <mike_mp@zzzcomputing.com> | 2010-07-11 13:15:51 -0400 |
---|---|---|
committer | Mike Bayer <mike_mp@zzzcomputing.com> | 2010-07-11 13:15:51 -0400 |
commit | a6b62cc3fed5f06d3428b1f6ee13756175ded61b (patch) | |
tree | 6483de311229fae0b40e9caf36b579b4f1c58ff2 /test/engine/test_parseconnect.py | |
parent | 5cce6bf2a86cb318d80953cd3712fcb77bbc52db (diff) | |
download | sqlalchemy-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.py | 182 |
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' |