def test_transient_to_pending_collection(self):
        mapper(User, users, properties={
            'addresses': relation(Address, backref='user',
                                  collection_class=OrderedSet)})
        mapper(Address, addresses)
        on_load = self.on_load_tracker(User)
        self.on_load_tracker(Address, on_load)

        u = User(id=7, name='fred', addresses=OrderedSet([
            Address(id=1, email_address='fred1'),
            Address(id=2, email_address='fred2'),
            ]))
        eq_(on_load.called, 0)

        sess = create_session()
        sess.merge(u)
        eq_(on_load.called, 3)

        merged_users = [e for e in sess if isinstance(e, User)]
        eq_(len(merged_users), 1)
        assert merged_users[0] is not u

        sess.flush()
        sess.clear()

        eq_(sess.query(User).one(),
            User(id=7, name='fred', addresses=OrderedSet([
                Address(id=1, email_address='fred1'),
                Address(id=2, email_address='fred2'),
            ]))
        )

    def test_dontload_with_eager(self):
        """

        This test illustrates that with dont_load=True, we can't just copy the
        committed_state of the merged instance over; since it references
        collection objects which themselves are to be merged.  This
        committed_state would instead need to be piecemeal 'converted' to
        represent the correct objects.  However, at the moment I'd rather not
        support this use case; if you are merging with dont_load=True, you're
        typically dealing with caching and the merged objects shouldnt be
        'dirty'.

        """
        mapper(User, users, properties={
            'addresses':relation(mapper(Address, addresses))
        })
        sess = create_session()
        u = User()
        u.id = 7
        u.name = "fred"
        a1 = Address()
        a1.email_address='[email protected]'
        u.addresses.append(a1)

        sess.add(u)
        sess.flush()

        sess2 = create_session()
        u2 = sess2.query(User).options(sa.orm.eagerload('addresses')).get(7)

        sess3 = create_session()
        u3 = sess3.merge(u2, dont_load=True)
        def go():
            sess3.flush()
        self.assert_sql_count(testing.db, go, 0)

    def test_expired_eager(self):
        mapper(User, users, properties={
            'addresses':relation(Address, backref='user', lazy=False),
            })
        mapper(Address, addresses)

        sess = create_session()
        u = sess.query(User).get(7)

        sess.expire(u)
        assert 'name' not in u.__dict__
        assert 'addresses' not in u.__dict__

        def go():
            assert u.addresses[0].email_address == '[email protected]'
            assert u.name == 'jack'
        # two loads, since relation() + scalar are
        # separate right now on per-attribute load
        self.assert_sql_count(testing.db, go, 2)
        assert 'name' in u.__dict__
        assert 'addresses' in u.__dict__

        sess.expire(u, ['name', 'addresses'])
        assert 'name' not in u.__dict__
        assert 'addresses' not in u.__dict__

        def go():
            sess.query(User).filter_by(id=7).one()
            assert u.addresses[0].email_address == '[email protected]'
            assert u.name == 'jack'
        # one load, since relation() + scalar are
        # together when eager load used with Query
        self.assert_sql_count(testing.db, go, 1)

    def test_detached_to_persistent_collection(self):
        mapper(User, users, properties={
            'addresses':relation(Address,
                                 backref='user',
                                 collection_class=OrderedSet)})
        mapper(Address, addresses)
        on_load = self.on_load_tracker(User)
        self.on_load_tracker(Address, on_load)

        a = Address(id=1, email_address='fred1')
        u = User(id=7, name='fred', addresses=OrderedSet([
            a,
            Address(id=2, email_address='fred2'),
        ]))
        sess = create_session()
        sess.add(u)
        sess.flush()
        sess.clear()

        u.name='fred jones'
        u.addresses.add(Address(id=3, email_address='fred3'))
        u.addresses.remove(a)

        eq_(on_load.called, 0)
        u = sess.merge(u)
        eq_(on_load.called, 4)
        sess.flush()
        sess.clear()

        eq_(sess.query(User).first(),
            User(id=7, name='fred jones', addresses=OrderedSet([
                Address(id=2, email_address='fred2'),
                Address(id=3, email_address='fred3')])))

    def test_uses_get(self):
        """test that a simple many-to-one lazyload optimizes to use query.get()."""

        for pj in (
            None,
            users.c.id==addresses.c.user_id,
            addresses.c.user_id==users.c.id
        ):
            mapper(Address, addresses, properties = dict(
                user = relation(mapper(User, users), lazy=True, primaryjoin=pj)
            ))

            sess = create_session()

            # load address
            a1 = sess.query(Address).filter_by(email_address="[email protected]").one()

            # load user that is attached to the address
            u1 = sess.query(User).get(8)

            def go():
                # lazy load of a1.user should get it from the session
                assert a1.user is u1
            self.assert_sql_count(testing.db, go, 0)
            sa.orm.clear_mappers()

    def test_basic(self):
        mapper(Employee, employees)
        mapper(Department, departments, properties=dict(employees=relation(Employee, lazy=False, backref="department")))

        d1 = Department(name="One")
        for e in "Jim", "Jack", "John", "Susan":
            d1.employees.append(Employee(name=e))

        d2 = Department(name="Two")
        for e in "Joe", "Bob", "Mary", "Wally":
            d2.employees.append(Employee(name=e))

        sess = create_session()
        sess.add_all((d1, d2))
        sess.flush()

        q = (
            sess.query(Department)
            .join("employees")
            .filter(Employee.name.startswith("J"))
            .distinct()
            .order_by([sa.desc(Department.name)])
        )

        eq_(q.count(), 2)
        assert q[0] is d2

 def test_one_to_many_scalar(self):
     mapper(User, users, properties = dict(
         address = relation(mapper(Address, addresses), lazy=True, uselist=False)
     ))
     q = create_session().query(User)
     l = q.filter(users.c.id == 7).all()
     assert [User(id=7, address=Address(id=1))] == l

    def test_basic(self):
        mapper(User, users, properties={
            'addresses':dynamic_loader(mapper(Address, addresses))
        })
        sess = create_session()
        u1 = User(name='jack')
        u2 = User(name='ed')
        u2.addresses.append(Address(email_address='[email protected]'))
        u1.addresses.append(Address(email_address='[email protected]'))
        sess.add_all((u1, u2))
        sess.flush()
        
        from sqlalchemy.orm import attributes
        self.assertEquals(attributes.get_history(attributes.instance_state(u1), 'addresses'), ([], [Address(email_address='[email protected]')], []))
        
        sess.clear()

        # test the test fixture a little bit
        assert User(name='jack', addresses=[Address(email_address='wrong')]) != sess.query(User).first()
        assert User(name='jack', addresses=[Address(email_address='[email protected]')]) == sess.query(User).first()

        assert [
            User(name='jack', addresses=[Address(email_address='[email protected]')]),
            User(name='ed', addresses=[Address(email_address='[email protected]')])
        ] == sess.query(User).all()

 def test_basic(self):
     mapper(User, users, properties={
         'addresses':relation(mapper(Address, addresses), lazy=True)
     })
     sess = create_session()
     q = sess.query(User)
     assert [User(id=7, addresses=[Address(id=1, email_address='[email protected]')])] == q.filter(users.c.id == 7).all()

    def test_delete_cascade(self):
        mapper(User, users, properties={
            'addresses':dynamic_loader(mapper(Address, addresses), order_by=Address.id, backref='user', cascade="all, delete-orphan")
        })
        sess = create_session(autoflush=True)
        u = User(name='ed')
        u.addresses.append(Address(email_address='a'))
        u.addresses.append(Address(email_address='b'))
        u.addresses.append(Address(email_address='c'))
        u.addresses.append(Address(email_address='d'))
        u.addresses.append(Address(email_address='e'))
        u.addresses.append(Address(email_address='f'))
        sess.add(u)

        assert Address(email_address='c') == u.addresses[2]
        sess.delete(u.addresses[2])
        sess.delete(u.addresses[4])
        sess.delete(u.addresses[3])
        assert [Address(email_address='a'), Address(email_address='b'), Address(email_address='d')] == list(u.addresses)

        sess.clear()
        u = sess.query(User).get(u.id)

        sess.delete(u)

        # u.addresses relation will have to force the load
        # of all addresses so that they can be updated
        sess.flush()
        sess.close()

        assert testing.db.scalar(addresses.count()) ==0

    def test_backrefs_dont_lazyload(self):
        mapper(User, users, properties={
            'addresses':relation(Address, backref='user')
        })
        mapper(Address, addresses)
        sess = create_session()
        ad = sess.query(Address).filter_by(id=1).one()
        assert ad.user.id == 7
        def go():
            ad.user = None
            assert ad.user is None
        self.assert_sql_count(testing.db, go, 0)

        u1 = sess.query(User).filter_by(id=7).one()
        def go():
            assert ad not in u1.addresses
        self.assert_sql_count(testing.db, go, 1)

        sess.expire(u1, ['addresses'])
        def go():
            assert ad in u1.addresses
        self.assert_sql_count(testing.db, go, 1)

        sess.expire(u1, ['addresses'])
        ad2 = Address()
        def go():
            ad2.user = u1
            assert ad2.user is u1
        self.assert_sql_count(testing.db, go, 0)

        def go():
            assert ad2 in u1.addresses
        self.assert_sql_count(testing.db, go, 1)

    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) == [])

    def test_compileonattr_rel_backref_b(self):
        m = MetaData()
        t1 = Table("t1", m, Column("id", Integer, primary_key=True), Column("x", Integer))
        t2 = Table("t2", m, Column("id", Integer, primary_key=True), Column("t1_id", Integer, ForeignKey("t1.id")))

        class Base(object):
            def __init__(self):
                pass

        class Base_AKW(object):
            def __init__(self, *args, **kwargs):
                pass

        for base in object, Base, Base_AKW:

            class A(base):
                pass

            class B(base):
                pass

            mapper(A, t1)
            mapper(B, t2, properties=dict(a=relation(A, backref="bs")))

            a = A()
            b = B()
            b.a = a

            session = create_session()
            session.add(a)
            assert b in session, "base: %s" % base

 def test_configured_order_by(self):
     mapper(User, users, properties={
         'addresses':dynamic_loader(mapper(Address, addresses), order_by=desc(Address.email_address))
     })
     sess = create_session()
     u = sess.query(User).get(8)
     eq_(list(u.addresses), [Address(email_address=u'[email protected]'), Address(email_address=u'[email protected]'), Address(email_address=u'[email protected]')])

    def test_persistence_check(self):
        mapper(User, users)
        s = create_session()
        u = s.query(User).get(7)
        s.clear()

        self.assertRaisesMessage(sa.exc.InvalidRequestError, r"is not persistent within this Session", s.expire, u)

 def test_count(self):
     mapper(User, users, properties={
         'addresses':dynamic_loader(mapper(Address, addresses))
     })
     sess = create_session()
     u = sess.query(User).first()
     eq_(u.addresses.count(), 1)

    def test_one_to_one_cascade(self):

        mapper(User, users, properties={
            'address':relation(mapper(Address, addresses),uselist = False)
        })
        on_load = self.on_load_tracker(User)
        self.on_load_tracker(Address, on_load)
        sess = create_session()

        u = User()
        u.id = 7
        u.name = "fred"
        a1 = Address()
        a1.email_address='[email protected]'
        u.address = a1

        sess.add(u)
        sess.flush()

        eq_(on_load.called, 0)

        sess2 = create_session()
        u2 = sess2.query(User).get(7)
        eq_(on_load.called, 1)
        u2.name = 'fred2'
        u2.address.email_address = '[email protected]'
        eq_(on_load.called, 2)

        u3 = sess.merge(u2)
        eq_(on_load.called, 2)
        assert u3 is u

    def test_dontload_with_backrefs(self):
        """dontload populates relations in both directions without requiring a load"""
        mapper(User, users, properties={
            'addresses':relation(mapper(Address, addresses), backref='user')
        })

        u = User(id=7, name='fred', addresses=[
            Address(email_address='ad1'),
            Address(email_address='ad2')])
        sess = create_session()
        sess.add(u)
        sess.flush()
        sess.close()
        assert 'user' in u.addresses[1].__dict__

        sess = create_session()
        u2 = sess.merge(u, dont_load=True)
        assert 'user' in u2.addresses[1].__dict__
        eq_(u2.addresses[1].user, User(id=7, name='fred'))

        sess.expire(u2.addresses[1], ['user'])
        assert 'user' not in u2.addresses[1].__dict__
        sess.close()

        sess = create_session()
        u = sess.merge(u2, dont_load=True)
        assert 'user' not in u.addresses[1].__dict__
        eq_(u.addresses[1].user, User(id=7, name='fred'))

    def test_synonym_comparable(self):
        class User(object):

           class Comparator(PropComparator):
               pass

           def _getValue(self):
               return self._value

           def _setValue(self, value):
               setattr(self, '_value', value)

           value = property(_getValue, _setValue)

        mapper(User, users, properties={
            'uid':synonym('id'),
            'foobar':comparable_property(User.Comparator,User.value),
        })
        
        sess = create_session()
        u = User()
        u.name = 'ed'
        sess.save(u)
        sess.flush()
        sess.expunge(u)
        sess.merge(u)