-from sqlalchemy.testing import eq_, assert_raises, assert_raises_message, \
- config, is_
-import re
-from sqlalchemy.testing.util import picklers
-from sqlalchemy.interfaces import ConnectionProxy
-from sqlalchemy import MetaData, Integer, String, INT, VARCHAR, func, \
- bindparam, select, event, TypeDecorator, create_engine, Sequence
-from sqlalchemy.sql import column, literal
-from sqlalchemy.testing.schema import Table, Column
+from sqlalchemy.testing import eq_, assert_raises_message
+from sqlalchemy import select
import sqlalchemy as tsa
-from sqlalchemy import testing
from sqlalchemy.testing import engines
-from sqlalchemy import util
-from sqlalchemy.testing.engines import testing_engine
import logging.handlers
-from sqlalchemy.dialects.oracle.zxjdbc import ReturningParam
-from sqlalchemy.engine import result as _result, default
-from sqlalchemy.engine.base import Engine
from sqlalchemy.testing import fixtures
-from sqlalchemy.testing.mock import Mock, call, patch
+
class LogParamsTest(fixtures.TestBase):
__only_on__ = 'sqlite'
__requires__ = 'ad_hoc_engines',
def setup(self):
- self.eng = engines.testing_engine(options={'echo':True})
+ self.eng = engines.testing_engine(options={'echo': True})
self.eng.execute("create table foo (data string)")
self.buf = logging.handlers.BufferingHandler(100)
for log in [
def test_log_large_dict(self):
self.eng.execute(
"INSERT INTO foo (data) values (:data)",
- [{"data":str(i)} for i in range(100)]
+ [{"data": str(i)} for i in range(100)]
)
eq_(
self.buf.buffer[1].message,
"100 total bound parameter sets ... {'data': '98'}, {'data': '99'}\]",
lambda: self.eng.execute(
"INSERT INTO nonexistent (data) values (:data)",
- [{"data":str(i)} for i in range(100)]
+ [{"data": str(i)} for i in range(100)]
)
)
)
)
+
class LoggingNameTest(fixtures.TestBase):
__requires__ = 'ad_hoc_engines',
assert name in (
'sqlalchemy.engine.base.Engine.%s' % eng_name,
'sqlalchemy.pool.%s.%s' %
- (eng.pool.__class__.__name__, pool_name)
+ (eng.pool.__class__.__name__, pool_name)
)
def _assert_no_name_in_execute(self, eng):
def _named_engine(self, **kw):
options = {
- 'logging_name':'myenginename',
- 'pool_logging_name':'mypoolname',
- 'echo':True
+ 'logging_name': 'myenginename',
+ 'pool_logging_name': 'mypoolname',
+ 'echo': True
}
options.update(kw)
return engines.testing_engine(options=options)
def _unnamed_engine(self, **kw):
- kw.update({'echo':True})
+ kw.update({'echo': True})
return engines.testing_engine(options=kw)
def setup(self):
eng = self._unnamed_engine(echo='debug', echo_pool='debug')
self._assert_no_name_in_execute(eng)
+
class EchoTest(fixtures.TestBase):
__requires__ = 'ad_hoc_engines',