"""
-from sqlalchemy import util, topological
+from sqlalchemy import util
+from sqlalchemy.util import topological
from sqlalchemy.orm import attributes, interfaces
from sqlalchemy.orm import util as mapperutil
from sqlalchemy.orm.util import _state_mapper
import weakref, time, threading
from sqlalchemy import exc, log, event, events, interfaces, util
-from sqlalchemy import queue as sqla_queue
+from sqlalchemy.util import queue as sqla_queue
from sqlalchemy.util import threading, pickle, memoized_property
proxies = {}
-from sqlalchemy import exc, schema, topological, util, sql, types as sqltypes
+from sqlalchemy import exc, schema, util, sql, types as sqltypes
+from sqlalchemy.util import topological
from sqlalchemy.sql import expression, operators, visitors
from itertools import chain
portable_instancemethod, unbound_method_to_callable, \
getargspec_init, format_argspec_init, format_argspec_plus, \
get_func_kwargs, get_cls_kwargs, decorator, as_interface, \
- function_named, memoized_property, memoized_instancemethod, \
+ memoized_property, memoized_instancemethod, \
reset_memoized, group_expirable_memoized_property, importlater, \
monkeypatch_proxied_specials, asbool, bool_or_str, coerce_kw_type,\
duck_type_collection, assert_arg_type, symbol, dictlike_iteritems,\
from deprecations import warn_deprecated, warn_pending_deprecation, \
deprecated, pending_deprecation
-
"""Handle Python version/platform incompatibilities."""
+
import sys
# Py2K
raise TypeError("dictionary does not contain required keys %s" %
', '.join(required - found))
-def function_named(fn, name):
- """Return a function with a given __name__.
-
- Will assign to __name__ and return the original function if possible on
- the Python implementation, otherwise a new function will be constructed.
-
- """
- try:
- fn.__name__ = name
- except TypeError:
- fn = types.FunctionType(fn.func_code, fn.func_globals, name,
- fn.func_defaults, fn.func_closure)
- return fn
-
class memoized_property(object):
"""A read-only @property that is only evaluated once."""
-import sqlalchemy.topological as topological
+from sqlalchemy.util import topological
from test.lib import TestBase
from test.lib.testing import assert_raises, eq_
from test.lib.util import conforms_partial_ordering
from sqlalchemy import exc, util
-
class DependencySortTest(TestBase):
def assert_sort(self, tuples, allitems=None):
import sys, types, weakref
from collections import deque
from test.bootstrap import config
-from sqlalchemy.util import function_named, callable
+from test.lib.util import decorator
+from sqlalchemy.util import callable
from sqlalchemy import event
import re
import warnings
def drop_all_tables(metadata):
testing_reaper.close_all()
metadata.drop_all()
-
-def assert_conns_closed(fn):
- def decorated(*args, **kw):
- try:
- fn(*args, **kw)
- finally:
- testing_reaper.assert_all_closed()
- return function_named(decorated, fn.__name__)
-
-def rollback_open_connections(fn):
+
+@decorator
+def assert_conns_closed(fn, *args, **kw):
+ try:
+ fn(*args, **kw)
+ finally:
+ testing_reaper.assert_all_closed()
+
+@decorator
+def rollback_open_connections(fn, *args, **kw):
"""Decorator that rolls back all open connections after fn execution."""
- def decorated(*args, **kw):
- try:
- fn(*args, **kw)
- finally:
- testing_reaper.rollback_all()
- return function_named(decorated, fn.__name__)
+ try:
+ fn(*args, **kw)
+ finally:
+ testing_reaper.rollback_all()
-def close_first(fn):
+@decorator
+def close_first(fn, *args, **kw):
"""Decorator that closes all connections before fn execution."""
- def decorated(*args, **kw):
- testing_reaper.close_all()
- fn(*args, **kw)
- return function_named(decorated, fn.__name__)
+
+ testing_reaper.close_all()
+ fn(*args, **kw)
-def close_open_connections(fn):
+@decorator
+def close_open_connections(fn, *args, **kw):
"""Decorator that closes all connections after fn execution."""
-
- def decorated(*args, **kw):
- try:
- fn(*args, **kw)
- finally:
- testing_reaper.close_all()
- return function_named(decorated, fn.__name__)
+ try:
+ fn(*args, **kw)
+ finally:
+ testing_reaper.close_all()
def all_dialects(exclude=None):
import sqlalchemy.databases as d
"""
import os, sys
-from test.lib.util import function_named, gc_collect
+from test.lib.util import gc_collect, decorator
from nose import SkipTest
__all__ = 'profiled', 'function_call_count', 'conditional_call_count'
all_targets.add(target)
filename = "%s.prof" % target
-
- def decorator(fn):
- def profiled(*args, **kw):
- if (target not in profile_config['targets'] and
- not target_opts.get('always', None)):
- return fn(*args, **kw)
-
- elapsed, load_stats, result = _profile(
- filename, fn, *args, **kw)
-
- report = target_opts.get('report', profile_config['report'])
- if report:
- sort_ = target_opts.get('sort', profile_config['sort'])
- limit = target_opts.get('limit', profile_config['limit'])
- print "Profile report for target '%s' (%s)" % (
- target, filename)
-
- stats = load_stats()
- stats.sort_stats(*sort_)
- if limit:
- stats.print_stats(limit)
- else:
- stats.print_stats()
- #stats.print_callers()
- os.unlink(filename)
- return result
- return function_named(profiled, fn.__name__)
- return decorator
+
+ @decorator
+ def decorate(fn, *args, **kw):
+ if (target not in profile_config['targets'] and
+ not target_opts.get('always', None)):
+ return fn(*args, **kw)
+
+ elapsed, load_stats, result = _profile(
+ filename, fn, *args, **kw)
+
+ report = target_opts.get('report', profile_config['report'])
+ if report:
+ sort_ = target_opts.get('sort', profile_config['sort'])
+ limit = target_opts.get('limit', profile_config['limit'])
+ print "Profile report for target '%s' (%s)" % (
+ target, filename)
+
+ stats = load_stats()
+ stats.sort_stats(*sort_)
+ if limit:
+ stats.print_stats(limit)
+ else:
+ stats.print_stats()
+ #stats.print_callers()
+ os.unlink(filename)
+ return result
+ return decorate
def function_call_count(count=None, versions={}, variance=0.05):
"""Assert a target for a test case's function call count.
if count is None:
return lambda fn: fn
+
+ @decorator
+ def decorate(fn, *args, **kw):
+ try:
+ filename = "%s.prof" % fn.__name__
- def decorator(fn):
- def counted(*args, **kw):
- try:
- filename = "%s.prof" % fn.__name__
+ elapsed, stat_loader, result = _profile(
+ filename, fn, *args, **kw)
- elapsed, stat_loader, result = _profile(
- filename, fn, *args, **kw)
+ stats = stat_loader()
+ calls = stats.total_calls
- stats = stat_loader()
- calls = stats.total_calls
+ stats.sort_stats('calls', 'cumulative')
+ stats.print_stats()
+ #stats.print_callers()
+ deviance = int(count * variance)
+ if (calls < (count - deviance) or
+ calls > (count + deviance)):
+ raise AssertionError(
+ "Function call count %s not within %s%% "
+ "of expected %s. (Python version %s)" % (
+ calls, (variance * 100), count, py_version))
- stats.sort_stats('calls', 'cumulative')
- stats.print_stats()
- #stats.print_callers()
- deviance = int(count * variance)
- if (calls < (count - deviance) or
- calls > (count + deviance)):
- raise AssertionError(
- "Function call count %s not within %s%% "
- "of expected %s. (Python version %s)" % (
- calls, (variance * 100), count, py_version))
-
- return result
- finally:
- if os.path.exists(filename):
- os.unlink(filename)
- return function_named(counted, fn.__name__)
- return decorator
+ return result
+ finally:
+ if os.path.exists(filename):
+ os.unlink(filename)
+ return decorate
def conditional_call_count(discriminator, categories):
"""Apply a function call count conditionally at runtime.
have a function count penalty not seen in the full suite, due to lazy
initialization in the DB-API, SA, etc.
"""
-
- def decorator(fn):
- def at_runtime(*args, **kw):
- criteria = categories.get(discriminator(), None)
- if criteria is None:
- return fn(*args, **kw)
-
- rewrapped = function_call_count(*criteria)(fn)
- return rewrapped(*args, **kw)
- return function_named(at_runtime, fn.__name__)
- return decorator
+ @decorator
+ def decorate(fn, *args, **kw):
+ criteria = categories.get(discriminator(), None)
+ if criteria is None:
+ return fn(*args, **kw)
+
+ rewrapped = function_call_count(*criteria)(fn)
+ return rewrapped(*args, **kw)
+ return decorate
def _profile(filename, fn, *args, **kw):
from test.bootstrap import config
from test.lib import assertsql, util as testutil
-from sqlalchemy.util import function_named, py3k
+from sqlalchemy.util import py3k, decorator
from engines import drop_all_tables
from sqlalchemy import exc as sa_exc, util, types as sqltypes, schema, \
docstring = getattr(callable_, '__doc__', None) or callable_.__name__
description = docstring.split('\n')[0]
- def decorate(fn):
- fn_name = fn.__name__
- def maybe(*args, **kw):
- if not callable_():
- return fn(*args, **kw)
+ @decorator
+ def decorate(fn, *args, **kw):
+ if not callable_():
+ return fn(*args, **kw)
+ else:
+ try:
+ fn(*args, **kw)
+ except Exception, ex:
+ print ("'%s' failed as expected (condition: %s): %s " % (
+ fn.__name__, description, str(ex)))
+ return True
else:
- try:
- fn(*args, **kw)
- except Exception, ex:
- print ("'%s' failed as expected (condition: %s): %s " % (
- fn_name, description, str(ex)))
- return True
- else:
- raise AssertionError(
- "Unexpected success for '%s' (condition: %s)" %
- (fn_name, description))
- return function_named(maybe, fn_name)
+ raise AssertionError(
+ "Unexpected success for '%s' (condition: %s)" %
+ (fn.__name__, description))
return decorate
-
-def future(fn):
+@decorator
+def future(fn, *args, **kw):
"""Mark a test as expected to unconditionally fail.
Takes no arguments, omit parens when using as a decorator.
"""
- fn_name = fn.__name__
- def decorated(*args, **kw):
- try:
- fn(*args, **kw)
- except Exception, ex:
- print ("Future test '%s' failed as expected: %s " % (
- fn_name, str(ex)))
- return True
- else:
- raise AssertionError(
- "Unexpected success for future test '%s'" % fn_name)
- return function_named(decorated, fn_name)
+ try:
+ fn(*args, **kw)
+ except Exception, ex:
+ print ("Future test '%s' failed as expected: %s " % (
+ fn.__name__, str(ex)))
+ return True
+ else:
+ raise AssertionError(
+ "Unexpected success for future test '%s'" % fn.__name__)
def db_spec(*dbs):
dialects = set([x for x in dbs if '+' not in x])
"""
spec = db_spec(dbs)
-
- def decorate(fn):
- fn_name = fn.__name__
- def maybe(*args, **kw):
- if not spec(config.db):
- return fn(*args, **kw)
+
+ @decorator
+ def decorate(fn, *args, **kw):
+ if not spec(config.db):
+ return fn(*args, **kw)
+ else:
+ try:
+ fn(*args, **kw)
+ except Exception, ex:
+ print ("'%s' failed as expected on DB implementation "
+ "'%s+%s': %s" % (
+ fn.__name__, config.db.name, config.db.driver, reason))
+ return True
else:
- try:
- fn(*args, **kw)
- except Exception, ex:
- print ("'%s' failed as expected on DB implementation "
- "'%s+%s': %s" % (
- fn_name, config.db.name, config.db.driver, reason))
- return True
- else:
- raise AssertionError(
- "Unexpected success for '%s' on DB implementation '%s+%s'" %
- (fn_name, config.db.name, config.db.driver))
- return function_named(maybe, fn_name)
+ raise AssertionError(
+ "Unexpected success for '%s' on DB implementation '%s+%s'" %
+ (fn.__name__, config.db.name, config.db.driver))
return decorate
def fails_on_everything_except(*dbs):
spec = db_spec(*dbs)
- def decorate(fn):
- fn_name = fn.__name__
- def maybe(*args, **kw):
- if spec(config.db):
- return fn(*args, **kw)
+ @decorator
+ def decorate(fn, *args, **kw):
+ if spec(config.db):
+ return fn(*args, **kw)
+ else:
+ try:
+ fn(*args, **kw)
+ except Exception, ex:
+ print ("'%s' failed as expected on DB implementation "
+ "'%s+%s': %s" % (
+ fn.__name__, config.db.name, config.db.driver, str(ex)))
+ return True
else:
- try:
- fn(*args, **kw)
- except Exception, ex:
- print ("'%s' failed as expected on DB implementation "
- "'%s+%s': %s" % (
- fn_name, config.db.name, config.db.driver, str(ex)))
- return True
- else:
- raise AssertionError(
- "Unexpected success for '%s' on DB implementation '%s+%s'" %
- (fn_name, config.db.name, config.db.driver))
- return function_named(maybe, fn_name)
+ raise AssertionError(
+ "Unexpected success for '%s' on DB implementation '%s+%s'" %
+ (fn.__name__, config.db.name, config.db.driver))
return decorate
def crashes(db, reason):
"""
carp = _should_carp_about_exclusion(reason)
spec = db_spec(db)
- def decorate(fn):
- fn_name = fn.__name__
- def maybe(*args, **kw):
- if spec(config.db):
- msg = "'%s' unsupported on DB implementation '%s+%s': %s" % (
- fn_name, config.db.name, config.db.driver, reason)
- print msg
- if carp:
- print >> sys.stderr, msg
- return True
- else:
- return fn(*args, **kw)
- return function_named(maybe, fn_name)
+ @decorator
+ def decorate(fn, *args, **kw):
+ if spec(config.db):
+ msg = "'%s' unsupported on DB implementation '%s+%s': %s" % (
+ fn.__name__, config.db.name, config.db.driver, reason)
+ print msg
+ if carp:
+ print >> sys.stderr, msg
+ return True
+ else:
+ return fn(*args, **kw)
return decorate
def _block_unconditionally(db, reason):
"""
carp = _should_carp_about_exclusion(reason)
spec = db_spec(db)
- def decorate(fn):
- fn_name = fn.__name__
- def maybe(*args, **kw):
- if spec(config.db):
- msg = "'%s' unsupported on DB implementation '%s+%s': %s" % (
- fn_name, config.db.name, config.db.driver, reason)
- print msg
- if carp:
- print >> sys.stderr, msg
- return True
- else:
- return fn(*args, **kw)
- return function_named(maybe, fn_name)
+ @decorator
+ def decorate(fn, *args, **kw):
+ if spec(config.db):
+ msg = "'%s' unsupported on DB implementation '%s+%s': %s" % (
+ fn.__name__, config.db.name, config.db.driver, reason)
+ print msg
+ if carp:
+ print >> sys.stderr, msg
+ return True
+ else:
+ return fn(*args, **kw)
return decorate
def only_on(dbs, reason):
carp = _should_carp_about_exclusion(reason)
spec = db_spec(*util.to_list(dbs))
- def decorate(fn):
- fn_name = fn.__name__
- def maybe(*args, **kw):
- if spec(config.db):
- return fn(*args, **kw)
- else:
- msg = "'%s' unsupported on DB implementation '%s+%s': %s" % (
- fn_name, config.db.name, config.db.driver, reason)
- print msg
- if carp:
- print >> sys.stderr, msg
- return True
- return function_named(maybe, fn_name)
+ @decorator
+ def decorate(fn, *args, **kw):
+ if spec(config.db):
+ return fn(*args, **kw)
+ else:
+ msg = "'%s' unsupported on DB implementation '%s+%s': %s" % (
+ fn.__name__, config.db.name, config.db.driver, reason)
+ print msg
+ if carp:
+ print >> sys.stderr, msg
+ return True
return decorate
def exclude(db, op, spec, reason):
"""
carp = _should_carp_about_exclusion(reason)
- def decorate(fn):
- fn_name = fn.__name__
- def maybe(*args, **kw):
- if _is_excluded(db, op, spec):
- msg = "'%s' unsupported on DB %s version '%s': %s" % (
- fn_name, config.db.name, _server_version(), reason)
- print msg
- if carp:
- print >> sys.stderr, msg
- return True
- else:
- return fn(*args, **kw)
- return function_named(maybe, fn_name)
+ @decorator
+ def decorate(fn, *args, **kw):
+ if _is_excluded(db, op, spec):
+ msg = "'%s' unsupported on DB %s version '%s': %s" % (
+ fn.__name__, config.db.name, _server_version(), reason)
+ print msg
+ if carp:
+ print >> sys.stderr, msg
+ return True
+ else:
+ return fn(*args, **kw)
return decorate
def _should_carp_about_exclusion(reason):
reason = reason or predicate.__name__
carp = _should_carp_about_exclusion(reason)
- def decorate(fn):
- fn_name = fn.__name__
- def maybe(*args, **kw):
- if predicate():
- msg = "'%s' skipped on DB %s version '%s': %s" % (
- fn_name, config.db.name, _server_version(), reason)
- print msg
- if carp:
- print >> sys.stderr, msg
- return True
- else:
- return fn(*args, **kw)
- return function_named(maybe, fn_name)
+ @decorator
+ def decorate(fn, *args, **kw):
+ if predicate():
+ msg = "'%s' skipped on DB %s version '%s': %s" % (
+ fn.__name__, config.db.name, _server_version(), reason)
+ print msg
+ if carp:
+ print >> sys.stderr, msg
+ return True
+ else:
+ return fn(*args, **kw)
return decorate
def emits_warning(*messages):
# and may work on non-CPython if they keep to the spirit of
# warnings.showwarning's docstring.
# - update: jython looks ok, it uses cpython's module
- def decorate(fn):
- def safe(*args, **kw):
- # todo: should probably be strict about this, too
- filters = [dict(action='ignore',
- category=sa_exc.SAPendingDeprecationWarning)]
- if not messages:
- filters.append(dict(action='ignore',
- category=sa_exc.SAWarning))
- else:
- filters.extend(dict(action='ignore',
- message=message,
- category=sa_exc.SAWarning)
- for message in messages)
- for f in filters:
- warnings.filterwarnings(**f)
- try:
- return fn(*args, **kw)
- finally:
- resetwarnings()
- return function_named(safe, fn.__name__)
+
+ @decorator
+ def decorate(fn, *args, **kw):
+ # todo: should probably be strict about this, too
+ filters = [dict(action='ignore',
+ category=sa_exc.SAPendingDeprecationWarning)]
+ if not messages:
+ filters.append(dict(action='ignore',
+ category=sa_exc.SAWarning))
+ else:
+ filters.extend(dict(action='ignore',
+ message=message,
+ category=sa_exc.SAWarning)
+ for message in messages)
+ for f in filters:
+ warnings.filterwarnings(**f)
+ try:
+ return fn(*args, **kw)
+ finally:
+ resetwarnings()
return decorate
def emits_warning_on(db, *warnings):
"""
spec = db_spec(db)
- def decorate(fn):
- def maybe(*args, **kw):
- if isinstance(db, basestring):
- if not spec(config.db):
- return fn(*args, **kw)
- else:
- wrapped = emits_warning(*warnings)(fn)
- return wrapped(*args, **kw)
+ @decorator
+ def decorate(fn, *args, **kw):
+ if isinstance(db, basestring):
+ if not spec(config.db):
+ return fn(*args, **kw)
else:
- if not _is_excluded(*db):
- return fn(*args, **kw)
- else:
- wrapped = emits_warning(*warnings)(fn)
- return wrapped(*args, **kw)
- return function_named(maybe, fn.__name__)
+ wrapped = emits_warning(*warnings)(fn)
+ return wrapped(*args, **kw)
+ else:
+ if not _is_excluded(*db):
+ return fn(*args, **kw)
+ else:
+ wrapped = emits_warning(*warnings)(fn)
+ return wrapped(*args, **kw)
return decorate
def assert_warnings(fn, warnings):
verbiage emitted by the sqlalchemy.util.deprecated decorator.
"""
-
- def decorate(fn):
- def safe(*args, **kw):
- # todo: should probably be strict about this, too
- filters = [dict(action='ignore',
- category=sa_exc.SAPendingDeprecationWarning)]
- if not messages:
- filters.append(dict(action='ignore',
- category=sa_exc.SADeprecationWarning))
- else:
- filters.extend(
- [dict(action='ignore',
- message=message,
- category=sa_exc.SADeprecationWarning)
- for message in
- [ (m.startswith('//') and
- ('Call to deprecated function ' + m[2:]) or m)
- for m in messages] ])
-
- for f in filters:
- warnings.filterwarnings(**f)
- try:
- return fn(*args, **kw)
- finally:
- resetwarnings()
- return function_named(safe, fn.__name__)
+ @decorator
+ def decorate(fn, *args, **kw):
+ # todo: should probably be strict about this, too
+ filters = [dict(action='ignore',
+ category=sa_exc.SAPendingDeprecationWarning)]
+ if not messages:
+ filters.append(dict(action='ignore',
+ category=sa_exc.SADeprecationWarning))
+ else:
+ filters.extend(
+ [dict(action='ignore',
+ message=message,
+ category=sa_exc.SADeprecationWarning)
+ for message in
+ [ (m.startswith('//') and
+ ('Call to deprecated function ' + m[2:]) or m)
+ for m in messages] ])
+
+ for f in filters:
+ warnings.filterwarnings(**f)
+ try:
+ return fn(*args, **kw)
+ finally:
+ resetwarnings()
return decorate
def testing_warn(msg, stacklevel=3):
for column_values in rows])
table.append_ddl_listener('after-create', onload)
-def provide_metadata(fn):
+@decorator
+def provide_metadata(fn, *args, **kw):
"""Provides a bound MetaData object for a single test,
drops it afterwards."""
- def maybe(*args, **kw):
- metadata = schema.MetaData(db)
- context = dict(fn.func_globals)
- context['metadata'] = metadata
- # jython bug #1034
- rebound = types.FunctionType(
- fn.func_code, context, fn.func_name, fn.func_defaults,
- fn.func_closure)
- try:
- return rebound(*args, **kw)
- finally:
- metadata.drop_all()
- return function_named(maybe, fn.__name__)
-
-def resolve_artifact_names(fn):
+ metadata = schema.MetaData(db)
+ context = dict(fn.func_globals)
+ context['metadata'] = metadata
+ # jython bug #1034
+ rebound = types.FunctionType(
+ fn.func_code, context, fn.func_name, fn.func_defaults,
+ fn.func_closure)
+ try:
+ return rebound(*args, **kw)
+ finally:
+ metadata.drop_all()
+
+@decorator
+def resolve_artifact_names(fn, *args, **kw):
"""Decorator, augment function globals with tables and classes.
Swaps out the function's globals at execution time. The 'global' statement
# Also: it's lame that CPython accepts a dict-subclass for globals, but
# only calls dict methods. That would allow 'global' to pass through to
# the func_globals.
- def resolved(*args, **kwargs):
- self = args[0]
- context = dict(fn.func_globals)
- for source in self._artifact_registries:
- context.update(getattr(self, source))
- # jython bug #1034
- rebound = types.FunctionType(
- fn.func_code, context, fn.func_name, fn.func_defaults,
- fn.func_closure)
- return rebound(*args, **kwargs)
- return function_named(resolved, fn.func_name)
+ self = args[0]
+ context = dict(fn.func_globals)
+ for source in self._artifact_registries:
+ context.update(getattr(self, source))
+ # jython bug #1034
+ rebound = types.FunctionType(
+ fn.func_code, context, fn.func_name, fn.func_defaults,
+ fn.func_closure)
+ return rebound(*args, **kw)
class adict(dict):
"""Dict keys available as attributes. Shadows."""
-from sqlalchemy.util import jython, function_named, defaultdict
+from sqlalchemy.util import jython, defaultdict, decorator
import gc
import time
return iter(_all_orderings(elements))
+
+def function_named(fn, name):
+ """Return a function with a given __name__.
+
+ Will assign to __name__ and return the original function if possible on
+ the Python implementation, otherwise a new function will be constructed.
+
+ This function should be phased out as much as possible
+ in favor of @decorator. Tests that "generate" many named tests
+ should be modernized.
+
+ """
+ try:
+ fn.__name__ = name
+ except TypeError:
+ fn = types.FunctionType(fn.func_code, fn.func_globals, name,
+ fn.func_defaults, fn.func_closure)
+ return fn
+
from test.lib import config, testing
from test.lib.testing import resolve_artifact_names, adict
from test.lib.engines import drop_all_tables
-from sqlalchemy.util import function_named
from test.lib.entities import BasicEntity, ComparableEntity
Entity = BasicEntity
from sqlalchemy import util
from sqlalchemy.orm import *
-from sqlalchemy.util import function_named
+from test.lib.util import function_named
from test.orm import _base, _fixtures
from test.lib.schema import Table, Column
from test.lib.assertsql import AllOf, CompiledSQL
from test.lib import testing, engines
-from sqlalchemy.util import function_named
from test.orm import _base, _fixtures
from test.lib.schema import Table, Column
from sqlalchemy.orm import *
from test.lib import testing
-from sqlalchemy.util import function_named
+from test.lib.util import function_named
from test.orm import _base
from test.lib.schema import Table, Column
from sqlalchemy.orm import exc as orm_exc
from sqlalchemy import exc as sa_exc
from test.lib import Column, testing
-from sqlalchemy.util import function_named
+from test.lib.util import function_named
from test.orm import _fixtures, _base
class Person(_fixtures.Base):
from sqlalchemy.orm import *
from test.lib import TestBase, AssertsExecutionResults, testing
-from sqlalchemy.util import function_named
+from test.lib.util import function_named
from test.orm import _base, _fixtures
from test.lib.testing import eq_
from test.lib.schema import Table, Column
from test.lib.testing import eq_, ne_, assert_raises
from test.orm import _base
from test.lib.util import gc_collect, all_partial_orderings
-from sqlalchemy.util import cmp, jython
-from sqlalchemy import event, topological
+from sqlalchemy.util import cmp, jython, topological
+from sqlalchemy import event
# global for pickling tests
MyTest = None
from sqlalchemy.orm import mapper, relationship, create_session, Query, attributes
from sqlalchemy.orm.dynamic import AppenderMixin
from test.lib.testing import eq_, AssertsCompiledSQL, assert_raises_message
-from sqlalchemy.util import function_named
from test.orm import _base, _fixtures
from test.lib.schema import Table
from test.lib.schema import Column
from test.lib.testing import eq_, ne_
-from sqlalchemy.util import function_named
+from test.lib.util import decorator
from test.orm import _base
+@decorator
+def modifies_instrumentation_finders(fn, *args, **kw):
+ pristine = instrumentation.instrumentation_finders[:]
+ try:
+ fn(*args, **kw)
+ finally:
+ del instrumentation.instrumentation_finders[:]
+ instrumentation.instrumentation_finders.extend(pristine)
-def modifies_instrumentation_finders(fn):
- def decorated(*args, **kw):
- pristine = instrumentation.instrumentation_finders[:]
+def with_lookup_strategy(strategy):
+ @decorator
+ def decorate(fn, *args, **kw):
try:
- fn(*args, **kw)
+ instrumentation._install_lookup_strategy(strategy)
+ return fn(*args, **kw)
finally:
- del instrumentation.instrumentation_finders[:]
- instrumentation.instrumentation_finders.extend(pristine)
- return function_named(decorated, fn.func_name)
-
-def with_lookup_strategy(strategy):
- def decorate(fn):
- def wrapped(*args, **kw):
- try:
- instrumentation._install_lookup_strategy(strategy)
- return fn(*args, **kw)
- finally:
- instrumentation._install_lookup_strategy(sa.util.symbol('native'))
- return function_named(wrapped, fn.func_name)
+ instrumentation._install_lookup_strategy(sa.util.symbol('native'))
return decorate
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
