+
from sqlalchemy.interfaces import ConnectionProxy
-import re
from sqlalchemy.engine.default import DefaultDialect
from sqlalchemy.engine.base import Connection
import testing
+import re
class AssertRule(object):
def process_execute(self, clauseelement, *multiparams, **params):
import testlib.config as config
from testlib.compat import _function_named
-from testlib import assertsql
# Delayed imports
MetaData = None
return True
def assert_sql_execution(self, db, callable_, *rules):
+ from testlib import assertsql
assertsql.asserter.add_rules(rules)
try:
callable_()
assertsql.asserter.clear_rules()
def assert_sql(self, db, callable_, list_, with_sequences=None):
+ from testlib import assertsql
+
if with_sequences is not None and config.db.name in ('firebird', 'oracle', 'postgres'):
rules = with_sequences
else:
self.assert_sql_execution(db, callable_, *newrules)
def assert_sql_count(self, db, callable_, count):
+ from testlib import assertsql
self.assert_sql_execution(db, callable_, assertsql.CountStatements(count))