diff options
author | Mike Bayer <mike_mp@zzzcomputing.com> | 2014-12-05 11:35:42 -0500 |
---|---|---|
committer | Mike Bayer <mike_mp@zzzcomputing.com> | 2014-12-05 11:35:42 -0500 |
commit | 6e53e866dea4eba630128e856573ca1076b91611 (patch) | |
tree | c33b9f981d8a691f5b5c0f60b0df39322b84cdf6 | |
parent | edef95379777a9c84ee7dbcbc9a3b58849aa8930 (diff) | |
download | sqlalchemy-6e53e866dea4eba630128e856573ca1076b91611.tar.gz |
- pep8 cleanup
-rw-r--r-- | test/engine/test_parseconnect.py | 135 |
1 files changed, 77 insertions, 58 deletions
diff --git a/test/engine/test_parseconnect.py b/test/engine/test_parseconnect.py index 391b92144..d8f202f99 100644 --- a/test/engine/test_parseconnect.py +++ b/test/engine/test_parseconnect.py @@ -1,12 +1,13 @@ from sqlalchemy.testing import assert_raises, eq_, assert_raises_message -from sqlalchemy.util.compat import configparser, StringIO import sqlalchemy.engine.url as url from sqlalchemy import create_engine, engine_from_config, exc, pool from sqlalchemy.engine.default import DefaultDialect import sqlalchemy as tsa from sqlalchemy.testing import fixtures from sqlalchemy import testing -from sqlalchemy.testing.mock import Mock, MagicMock, patch +from sqlalchemy.testing.mock import Mock, MagicMock + +dialect = None class ParseConnectTest(fixtures.TestBase): @@ -31,21 +32,25 @@ class ParseConnectTest(fixtures.TestBase): 'dbtype://username:password@/database', 'dbtype:////usr/local/_xtest@example.com/members.db', 'dbtype://username:apples%2Foranges@hostspec/database', - 'dbtype://username:password@[2001:da8:2004:1000:202:116:160:90]/database?foo=bar', - 'dbtype://username:password@[2001:da8:2004:1000:202:116:160:90]:80/database?foo=bar' - ): + 'dbtype://username:password@[2001:da8:2004:1000:202:116:160:90]' + '/database?foo=bar', + 'dbtype://username:password@[2001:da8:2004:1000:202:116:160:90]:80' + '/database?foo=bar' + ): u = url.make_url(text) assert u.drivername in ('dbtype', 'dbtype+apitype') assert u.username in ('username', None) assert u.password in ('password', 'apples/oranges', None) - assert u.host in ('hostspec', '127.0.0.1', - '2001:da8:2004:1000:202:116:160:90', '', None), u.host - assert u.database in ('database', - '/usr/local/_xtest@example.com/members.db', - '/usr/db_file.db', ':memory:', '', - 'foo/bar/im/a/file', - 'E:/work/src/LEM/db/hello.db', None), u.database + assert u.host in ( + 'hostspec', '127.0.0.1', + '2001:da8:2004:1000:202:116:160:90', '', None), u.host + assert u.database in ( + 'database', + '/usr/local/_xtest@example.com/members.db', + '/usr/db_file.db', ':memory:', '', + 'foo/bar/im/a/file', + 'E:/work/src/LEM/db/hello.db', None), u.database eq_(str(u), text) def test_rfc1738_password(self): @@ -53,13 +58,17 @@ class ParseConnectTest(fixtures.TestBase): eq_(u.password, "pass word + other:words") eq_(str(u), "dbtype://user:pass word + other%3Awords@host/dbname") - u = url.make_url('dbtype://username:apples%2Foranges@hostspec/database') + u = url.make_url( + 'dbtype://username:apples%2Foranges@hostspec/database') eq_(u.password, "apples/oranges") eq_(str(u), 'dbtype://username:apples%2Foranges@hostspec/database') - u = url.make_url('dbtype://username:apples%40oranges%40%40@hostspec/database') + u = url.make_url( + 'dbtype://username:apples%40oranges%40%40@hostspec/database') eq_(u.password, "apples@oranges@@") - eq_(str(u), 'dbtype://username:apples%40oranges%40%40@hostspec/database') + eq_( + str(u), + 'dbtype://username:apples%40oranges%40%40@hostspec/database') u = url.make_url('dbtype://username%40:@hostspec/database') eq_(u.password, '') @@ -70,23 +79,23 @@ class ParseConnectTest(fixtures.TestBase): eq_(u.password, 'pass/word') eq_(str(u), 'dbtype://username:pass%2Fword@hostspec/database') + class DialectImportTest(fixtures.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', - ): + 'mysql', + 'firebird', + 'postgresql', + 'sqlite', + 'oracle', + 'mssql'): exec ('from sqlalchemy.dialects import %s\ndialect = ' '%s.dialect()' % (name, name), globals()) eq_(dialect.name, name) + class CreateEngineTest(fixtures.TestBase): """test that create_engine arguments of different types get propagated properly""" @@ -97,26 +106,28 @@ class CreateEngineTest(fixtures.TestBase): create_engine('postgresql://scott:tiger@somehost/test?foobe' 'r=12&lala=18&fooz=somevalue', module=dbapi, _initialize=False) - c = e.connect() + 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) - c = e.connect() - + create_engine( + 'postgresql://scott:tiger@somehost/test?fooz=' + 'somevalue', connect_args={ + 'foober': 12, + 'lala': 18, 'hoho': {'this': 'dict'}}, + module=dbapi, _initialize=False) + e.connect() 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 @@ -125,7 +136,6 @@ class CreateEngineTest(fixtures.TestBase): 'z=somevalue') assert e.echo is True - def test_engine_from_config_custom(self): from sqlalchemy import util from sqlalchemy.dialects import registry @@ -143,8 +153,9 @@ class CreateEngineTest(fixtures.TestBase): global dialect dialect = MyDialect - registry.register("mockdialect.barb", - ".".join(tokens[0:-1]), tokens[-1]) + registry.register( + "mockdialect.barb", + ".".join(tokens[0:-1]), tokens[-1]) config = { "sqlalchemy.url": "mockdialect+barb://", @@ -155,7 +166,6 @@ class CreateEngineTest(fixtures.TestBase): eq_(e.dialect.foobar, 5) eq_(e.dialect.bathoho, False) - def test_custom(self): dbapi = MockDBAPI(foober=12, lala=18, hoho={'this': 'dict'}, fooz='somevalue') @@ -169,7 +179,7 @@ class CreateEngineTest(fixtures.TestBase): e = create_engine('postgresql://', creator=connect, module=dbapi, _initialize=False) - c = e.connect() + e.connect() def test_recycle(self): dbapi = MockDBAPI(foober=12, lala=18, hoho={'this': 'dict'}, @@ -188,8 +198,9 @@ class CreateEngineTest(fixtures.TestBase): (True, pool.reset_rollback), (False, pool.reset_none), ]: - e = create_engine('postgresql://', pool_reset_on_return=value, - module=dbapi, _initialize=False) + e = create_engine( + 'postgresql://', pool_reset_on_return=value, + module=dbapi, _initialize=False) assert e.pool._reset_on_return is expected assert_raises( @@ -217,7 +228,7 @@ class CreateEngineTest(fixtures.TestBase): 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, @@ -233,14 +244,14 @@ class CreateEngineTest(fixtures.TestBase): dbapi = MockDBAPI() dbapi.Error = sqlite3.Error, dbapi.ProgrammingError = sqlite3.ProgrammingError - dbapi.connect = Mock(side_effect=sqlite3.ProgrammingError("random error")) + dbapi.connect = Mock( + side_effect=sqlite3.ProgrammingError("random error")) try: create_engine('sqlite://', module=dbapi).connect() assert False except tsa.exc.DBAPIError as de: assert not de.connection_invalidated - @testing.requires.sqlite def test_dont_touch_non_dbapi_exception_on_connect(self): e = create_engine('sqlite://') @@ -260,10 +271,12 @@ class CreateEngineTest(fixtures.TestBase): eq_(is_disconnect.call_count, 0) def test_ensure_dialect_does_is_disconnect_no_conn(self): - """test that is_disconnect() doesn't choke if no connection, cursor given.""" + """test that is_disconnect() doesn't choke if no connection, + cursor given.""" dialect = testing.db.dialect dbapi = dialect.dbapi - assert not dialect.is_disconnect(dbapi.OperationalError("test"), None, None) + assert not dialect.is_disconnect( + dbapi.OperationalError("test"), None, None) @testing.requires.sqlite def test_invalidate_on_connect(self): @@ -280,8 +293,9 @@ class CreateEngineTest(fixtures.TestBase): dbapi = MockDBAPI() dbapi.Error = sqlite3.Error, dbapi.ProgrammingError = sqlite3.ProgrammingError - dbapi.connect = Mock(side_effect=sqlite3.ProgrammingError( - "Cannot operate on a closed database.")) + dbapi.connect = Mock( + side_effect=sqlite3.ProgrammingError( + "Cannot operate on a closed database.")) try: create_engine('sqlite://', module=dbapi).connect() assert False @@ -313,7 +327,7 @@ class CreateEngineTest(fixtures.TestBase): echo_pool=None, module=mock_dbapi, _initialize=False, - ) + ) assert e.pool._recycle == 50 # these args work for QueuePool @@ -325,7 +339,7 @@ class CreateEngineTest(fixtures.TestBase): poolclass=tsa.pool.QueuePool, module=mock_dbapi, _initialize=False, - ) + ) # but not SingletonThreadPool @@ -338,7 +352,8 @@ class CreateEngineTest(fixtures.TestBase): poolclass=tsa.pool.SingletonThreadPool, module=mock_sqlite_dbapi, _initialize=False, - ) + ) + class TestRegNewDBAPI(fixtures.TestBase): def test_register_base(self): @@ -361,7 +376,8 @@ class TestRegNewDBAPI(fixtures.TestBase): global dialect dialect = MockDialect - registry.register("mockdialect.foob", ".".join(tokens[0:-1]), tokens[-1]) + registry.register( + "mockdialect.foob", ".".join(tokens[0:-1]), tokens[-1]) e = create_engine("mockdialect+foob://") assert isinstance(e.dialect, MockDialect) @@ -373,13 +389,16 @@ class TestRegNewDBAPI(fixtures.TestBase): e = create_engine("mysql+my_mock_dialect://") assert isinstance(e.dialect, MockDialect) + class MockDialect(DefaultDialect): @classmethod def dbapi(cls, **kw): return MockDBAPI() + def MockDBAPI(**assert_kwargs): connection = Mock(get_server_version_info=Mock(return_value='5.0')) + def connect(*args, **kwargs): for k in assert_kwargs: assert k in kwargs, 'key %s not present in dictionary' % k @@ -389,12 +408,12 @@ def MockDBAPI(**assert_kwargs): return connection return MagicMock( - sqlite_version_info=(99, 9, 9,), - version_info=(99, 9, 9,), - sqlite_version='99.9.9', - paramstyle='named', - connect=Mock(side_effect=connect) - ) + sqlite_version_info=(99, 9, 9,), + version_info=(99, 9, 9,), + sqlite_version='99.9.9', + paramstyle='named', + connect=Mock(side_effect=connect) + ) mock_dbapi = MockDBAPI() mock_sqlite_dbapi = msd = MockDBAPI() |