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__)

    def make_test(fetchtype):
        def test_roundtrip(self):
            class A(_fixtures.Base):pass
            class B(A):pass
            class C(B):pass

            if fetchtype == 'union':
                abc = a.outerjoin(b).outerjoin(c)
                bc = a.join(b).outerjoin(c)
            else:
                abc = bc = None

            mapper(A, a, with_polymorphic=('*', abc), polymorphic_on=a.c.type, polymorphic_identity='a')
            mapper(B, b, with_polymorphic=('*', bc), inherits=A, polymorphic_identity='b')
            mapper(C, c, inherits=B, polymorphic_identity='c')

            a1 = A(adata='a1')
            b1 = B(bdata='b1', adata='b1')
            b2 = B(bdata='b2', adata='b2')
            b3 = B(bdata='b3', adata='b3')
            c1 = C(cdata='c1', bdata='c1', adata='c1')
            c2 = C(cdata='c2', bdata='c2', adata='c2')
            c3 = C(cdata='c2', bdata='c2', adata='c2')

            sess = create_session()
            for x in (a1, b1, b2, b3, c1, c2, c3):
                sess.add(x)
            sess.flush()
            sess.expunge_all()

            #for obj in sess.query(A).all():
            #    print obj
            assert [
                A(adata='a1'),
                B(bdata='b1', adata='b1'),
                B(bdata='b2', adata='b2'),
                B(bdata='b3', adata='b3'),
                C(cdata='c1', bdata='c1', adata='c1'),
                C(cdata='c2', bdata='c2', adata='c2'),
                C(cdata='c2', bdata='c2', adata='c2'),
            ] == sess.query(A).order_by(A.id).all()

            assert [
                B(bdata='b1', adata='b1'),
                B(bdata='b2', adata='b2'),
                B(bdata='b3', adata='b3'),
                C(cdata='c1', bdata='c1', adata='c1'),
                C(cdata='c2', bdata='c2', adata='c2'),
                C(cdata='c2', bdata='c2', adata='c2'),
            ] == sess.query(B).all()

            assert [
                C(cdata='c1', bdata='c1', adata='c1'),
                C(cdata='c2', bdata='c2', adata='c2'),
                C(cdata='c2', bdata='c2', adata='c2'),
            ] == sess.query(C).all()

        test_roundtrip = function_named(
            test_roundtrip, 'test_%s' % fetchtype)
        return test_roundtrip

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 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__)

def resolve_artifact_names(fn):
    """Decorator, augment function globals with tables and classes.

    Swaps out the function's globals at execution time. The 'global' statement
    will not work as expected inside a decorated function.

    """
    # This could be automatically applied to framework and test_ methods in
    # the MappedTest-derived test suites but... *some* explicitness for this
    # magic is probably good.  Especially as 'global' won't work- these
    # rebound functions aren't regular Python..
    #
    # 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)

 def decorate(fn):
     def wrapped(*args, **kw):
         try:
             attributes._install_lookup_strategy(strategy)
             return fn(*args, **kw)
         finally:
             attributes._install_lookup_strategy(sa.util.symbol('native'))
     return function_named(wrapped, fn.func_name)

def close_first(fn):
    """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__)

def rollback_open_connections(fn):
    """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__)

def modifies_instrumentation_finders(fn):
    def decorated(*args, **kw):
        pristine = attributes.instrumentation_finders[:]
        try:
            fn(*args, **kw)
        finally:
            del attributes.instrumentation_finders[:]
            attributes.instrumentation_finders.extend(pristine)
    return function_named(decorated, fn.func_name)

def close_open_connections(fn):
    """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__)

 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)
             raise SkipTest(msg)
         else:
             return fn(*args, **kw)
     return function_named(maybe, fn_name)

 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)

def provide_metadata(fn):
    """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 _create_backref_test(autoflush, saveuser):

    @testing.resolve_artifact_names
    def test_backref(self):
        mapper(User, users, properties={
            'addresses':dynamic_loader(mapper(Address, addresses), backref='user')
        })
        sess = create_session(autoflush=autoflush)

        u = User(name='buffy')

        a = Address(email_address='[email protected]')
        a.user = u

        if saveuser:
            sess.add(u)
        else:
            sess.add(a)

        if not autoflush:
            sess.flush()

        assert u in sess
        assert a in sess

        self.assert_(list(u.addresses) == [a])

        a.user = None
        if not autoflush:
            self.assert_(list(u.addresses) == [a])

        if not autoflush:
            sess.flush()
        self.assert_(list(u.addresses) == [])

    test_backref = function_named(
        test_backref, "test%s%s" % ((autoflush and "_autoflush" or ""),
                                    (saveuser and "_saveuser" or "_savead")))
    setattr(SessionTest, test_backref.__name__, test_backref)

#.........这里部分代码省略.........
            Manager(status='ABA', manager_name='manager2', **{person_attribute_name:'jsmith'})
        ]
        
        pointy = employees[0]
        jsmith = employees[-1]
        dilbert = employees[1]
        
        session = create_session()
        c = Company(name='company1')
        c.employees = employees
        session.add(c)

        session.flush()
        session.expunge_all()
        
        eq_(session.query(Person).get(dilbert.person_id), dilbert)
        session.expunge_all()

        eq_(session.query(Person).filter(Person.person_id==dilbert.person_id).one(), dilbert)
        session.expunge_all()

        def go():
            cc = session.query(Company).get(c.company_id)
            eq_(cc.employees, employees)
            
        if not lazy_relation:
            if with_polymorphic != 'none':
                self.assert_sql_count(testing.db, go, 1)
            else:
                self.assert_sql_count(testing.db, go, 5)

        else:
            if with_polymorphic != 'none':
                self.assert_sql_count(testing.db, go, 2)
            else:
                self.assert_sql_count(testing.db, go, 6)
        
        # test selecting from the query, using the base mapped table (people) as the selection criterion.
        # in the case of the polymorphic Person query, the "people" selectable should be adapted to be "person_join"
        eq_(
            session.query(Person).filter(getattr(Person, person_attribute_name)=='dilbert').first(),
            dilbert
        )

        assert session.query(Person).filter(getattr(Person, person_attribute_name)=='dilbert').first().person_id

        eq_(
            session.query(Engineer).filter(getattr(Person, person_attribute_name)=='dilbert').first(),
            dilbert
        )
        
        # test selecting from the query, joining against an alias of the base "people" table.  test that
        # the "palias" alias does *not* get sucked up into the "person_join" conversion.
        palias = people.alias("palias")
        dilbert = session.query(Person).get(dilbert.person_id)
        assert dilbert is session.query(Person).filter((palias.c.name=='dilbert') & (palias.c.person_id==Person.person_id)).first()
        assert dilbert is session.query(Engineer).filter((palias.c.name=='dilbert') & (palias.c.person_id==Person.person_id)).first()
        assert dilbert is session.query(Person).filter((Engineer.engineer_name=="engineer1") & (engineers.c.person_id==people.c.person_id)).first()
        assert dilbert is session.query(Engineer).filter(Engineer.engineer_name=="engineer1")[0]
        
        dilbert.engineer_name = 'hes dibert!'

        session.flush()
        session.expunge_all()
        
        def go():
            session.query(Person).filter(getattr(Person, person_attribute_name)=='dilbert').first()
        self.assert_sql_count(testing.db, go, 1)
        session.expunge_all()
        dilbert = session.query(Person).filter(getattr(Person, person_attribute_name)=='dilbert').first()
        def go():
            # assert that only primary table is queried for already-present-in-session
            d = session.query(Person).filter(getattr(Person, person_attribute_name)=='dilbert').first()
        self.assert_sql_count(testing.db, go, 1)

        # test standalone orphans
        daboss = Boss(status='BBB', manager_name='boss', golf_swing='fore', **{person_attribute_name:'daboss'})
        session.add(daboss)
        assert_raises(orm_exc.FlushError, session.flush)
        c = session.query(Company).first()
        daboss.company = c
        manager_list = [e for e in c.employees if isinstance(e, Manager)]
        session.flush()
        session.expunge_all()

        eq_(session.query(Manager).order_by(Manager.person_id).all(), manager_list)
        c = session.query(Company).first()
        
        session.delete(c)
        session.flush()
        
        eq_(people.count().scalar(), 0)
        
    test_roundtrip = function_named(
        test_roundtrip, "test_%s%s%s_%s" % (
          (lazy_relation and "lazy" or "eager"),
          (include_base and "_inclbase" or ""),
          (redefine_colprop and "_redefcol" or ""),
          with_polymorphic))
    setattr(RoundTripTest, test_roundtrip.__name__, test_roundtrip)

        fn_name = fn.__name__
        def maybe(*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:
                    raise AssertionError(
                        "Unexpected success for '%s' (condition: %s)" %
                        (fn_name, description))
        return function_named(maybe, fn_name)
    return decorate


def future(fn):
    """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 " % (

def _generate_test(jointype="join1", usedata=False):
    def do_test(self):
        class Person(AttrSettable):
            pass
        class Manager(Person):
            pass

        if usedata:
            class Data(object):
                def __init__(self, data):
                    self.data = data

        if jointype == "join1":
            poly_union = polymorphic_union({
                'manager':managers.join(people, people.c.person_id==managers.c.person_id),
                'person':people.select(people.c.type=='person')
            }, None)
        elif jointype =="join2":
            poly_union = polymorphic_union({
                'manager':join(people, managers, people.c.person_id==managers.c.person_id),
                'person':people.select(people.c.type=='person')
            }, None)
        elif jointype == 'join3':
            poly_union = people.outerjoin(managers)
        elif jointype == "join4":
            poly_union=None

        if usedata:
            mapper(Data, data)

        if usedata:
            mapper(Person, people, with_polymorphic=('*', poly_union), polymorphic_identity='person', polymorphic_on=people.c.type,
                  properties={
                    'colleagues':relation(Person, primaryjoin=people.c.colleague_id==people.c.person_id, remote_side=people.c.colleague_id, uselist=True),
                    'data':relation(Data, uselist=False)
                    }
            )
        else:
            mapper(Person, people, with_polymorphic=('*', poly_union), polymorphic_identity='person', polymorphic_on=people.c.type,
                  properties={
                    'colleagues':relation(Person, primaryjoin=people.c.colleague_id==people.c.person_id,
                        remote_side=people.c.colleague_id, uselist=True)
                    }
            )

        mapper(Manager, managers, inherits=Person, inherit_condition=people.c.person_id==managers.c.person_id, polymorphic_identity='manager')

        sess = create_session()
        p = Person(name='person1')
        p2 = Person(name='person2')
        p3 = Person(name='person3')
        m = Manager(name='manager1')
        p.colleagues.append(p2)
        m.colleagues.append(p3)
        if usedata:
            p.data = Data('ps data')
            m.data = Data('ms data')

        sess.add(m)
        sess.add(p)
        sess.flush()

        sess.expunge_all()
        p = sess.query(Person).get(p.person_id)
        p2 = sess.query(Person).get(p2.person_id)
        p3 = sess.query(Person).get(p3.person_id)
        m = sess.query(Person).get(m.person_id)
        print p, p2, p.colleagues, m.colleagues
        assert len(p.colleagues) == 1
        assert p.colleagues == [p2]
        assert m.colleagues == [p3]
        if usedata:
            assert p.data.data == 'ps data'
            assert m.data.data == 'ms data'

    do_test = function_named(
        do_test, 'test_relation_on_base_class_%s_%s' % (
        jointype, data and "nodata" or "data"))
    return do_test

        location = Location(ref='ABC',name='London',issue=issue)

        page_size = PageSize(name='A4',width=210,height=297)

        magazine = Magazine(location=location,size=page_size)
        page = ClassifiedPage(magazine=magazine,page_no=1)
        page2 = MagazinePage(magazine=magazine,page_no=2)
        page3 = ClassifiedPage(magazine=magazine,page_no=3)
        session.add(pub)

        session.flush()
        print [x for x in session]
        session.expunge_all()

        session.flush()
        session.expunge_all()
        p = session.query(Publication).filter(Publication.name=="Test").one()

        print p.issues[0].locations[0].magazine.pages
        print [page, page2, page3]
        assert repr(p.issues[0].locations[0].magazine.pages) == repr([page, page2, page3]), repr(p.issues[0].locations[0].magazine.pages)

    test_roundtrip = function_named(
        test_roundtrip, "test_%s" % (not use_union and (use_joins and "joins" or "select") or "unions"))
    setattr(MagazineTest, test_roundtrip.__name__, test_roundtrip)

for (use_union, use_join) in [(True, False), (False, True), (False, False)]:
    generate_round_trip_test(use_union, use_join)

    def _create_test(polymorphic, name):
        def test_get(self):
            class Foo(object):
                pass

            class Bar(Foo):
                pass

            class Blub(Bar):
                pass

            if polymorphic:
                mapper(Foo, foo, polymorphic_on=foo.c.type, polymorphic_identity="foo")
                mapper(Bar, bar, inherits=Foo, polymorphic_identity="bar")
                mapper(Blub, blub, inherits=Bar, polymorphic_identity="blub")
            else:
                mapper(Foo, foo)
                mapper(Bar, bar, inherits=Foo)
                mapper(Blub, blub, inherits=Bar)

            sess = create_session()
            f = Foo()
            b = Bar()
            bl = Blub()
            sess.add(f)
            sess.add(b)
            sess.add(bl)
            sess.flush()

            if polymorphic:

                def go():
                    assert sess.query(Foo).get(f.id) == f
                    assert sess.query(Foo).get(b.id) == b
                    assert sess.query(Foo).get(bl.id) == bl
                    assert sess.query(Bar).get(b.id) == b
                    assert sess.query(Bar).get(bl.id) == bl
                    assert sess.query(Blub).get(bl.id) == bl

                self.assert_sql_count(testing.db, go, 0)
            else:
                # this is testing the 'wrong' behavior of using get()
                # polymorphically with mappers that are not configured to be
                # polymorphic.  the important part being that get() always
                # returns an instance of the query's type.
                def go():
                    assert sess.query(Foo).get(f.id) == f

                    bb = sess.query(Foo).get(b.id)
                    assert isinstance(b, Foo) and bb.id == b.id

                    bll = sess.query(Foo).get(bl.id)
                    assert isinstance(bll, Foo) and bll.id == bl.id

                    assert sess.query(Bar).get(b.id) == b

                    bll = sess.query(Bar).get(bl.id)
                    assert isinstance(bll, Bar) and bll.id == bl.id

                    assert sess.query(Blub).get(bl.id) == bl

                self.assert_sql_count(testing.db, go, 3)

        test_get = function_named(test_get, name)
        return test_get