import pickle
import sqlalchemy.orm.attributes as attributes
from sqlalchemy.orm.collections import collection
from sqlalchemy.orm.interfaces import AttributeExtension
from sqlalchemy import exc as sa_exc
from sqlalchemy.test import *
from sqlalchemy.test.testing import eq_, ne_, assert_raises
from test.orm import _base
from sqlalchemy.test.util import gc_collect
from sqlalchemy.util import cmp, jython

# global for pickling tests
MyTest = None
MyTest2 = None


class AttributesTest(_base.ORMTest):
    def setup(self):
        global MyTest, MyTest2
        class MyTest(object): pass
        class MyTest2(object): pass

    def teardown(self):
        global MyTest, MyTest2
        MyTest, MyTest2 = None, None

    def test_basic(self):
        class User(object):pass

        attributes.register_class(User)
        attributes.register_attribute(User, 'user_id', uselist=False, useobject=False)
        attributes.register_attribute(User, 'user_name', uselist=False, useobject=False)
        attributes.register_attribute(User, 'email_address', uselist=False, useobject=False)

        u = User()
        u.user_id = 7
        u.user_name = 'john'
        u.email_address = 'lala@123.com'

        self.assert_(u.user_id == 7 and u.user_name == 'john' and u.email_address == 'lala@123.com')
        attributes.instance_state(u).commit_all(attributes.instance_dict(u))
        self.assert_(u.user_id == 7 and u.user_name == 'john' and u.email_address == 'lala@123.com')

        u.user_name = 'heythere'
        u.email_address = 'foo@bar.com'
        self.assert_(u.user_id == 7 and u.user_name == 'heythere' and u.email_address == 'foo@bar.com')

    def test_pickleness(self):
        attributes.register_class(MyTest)
        attributes.register_class(MyTest2)
        attributes.register_attribute(MyTest, 'user_id', uselist=False, useobject=False)
        attributes.register_attribute(MyTest, 'user_name', uselist=False, useobject=False)
        attributes.register_attribute(MyTest, 'email_address', uselist=False, useobject=False)
        attributes.register_attribute(MyTest, 'some_mutable_data', mutable_scalars=True, copy_function=list, compare_function=cmp, uselist=False, useobject=False)
        attributes.register_attribute(MyTest2, 'a', uselist=False, useobject=False)
        attributes.register_attribute(MyTest2, 'b', uselist=False, useobject=False)
        # shouldnt be pickling callables at the class level
        def somecallable(*args, **kw):
            return None
        attributes.register_attribute(MyTest, "mt2", uselist = True, trackparent=True, callable_=somecallable, useobject=True)

        o = MyTest()
        o.mt2.append(MyTest2())
        o.user_id=7
        o.some_mutable_data = [1,2,3]
        o.mt2[0].a = 'abcde'
        pk_o = pickle.dumps(o)

        o2 = pickle.loads(pk_o)
        pk_o2 = pickle.dumps(o2)

        # so... pickle is creating a new 'mt2' string after a roundtrip here,
        # so we'll brute-force set it to be id-equal to the original string
        if False:
            o_mt2_str = [ k for k in o.__dict__ if k == 'mt2'][0]
            o2_mt2_str = [ k for k in o2.__dict__ if k == 'mt2'][0]
            self.assert_(o_mt2_str == o2_mt2_str)
            self.assert_(o_mt2_str is not o2_mt2_str)
            # change the id of o2.__dict__['mt2']
            former = o2.__dict__['mt2']
            del o2.__dict__['mt2']
            o2.__dict__[o_mt2_str] = former

            # Relies on dict ordering
            if not jython:
                self.assert_(pk_o == pk_o2)

        # the above is kind of distrurbing, so let's do it again a little
        # differently.  the string-id in serialization thing is just an
        # artifact of pickling that comes up in the first round-trip.
        # a -> b differs in pickle memoization of 'mt2', but b -> c will
        # serialize identically.

        o3 = pickle.loads(pk_o2)
        pk_o3 = pickle.dumps(o3)
        o4 = pickle.loads(pk_o3)
        pk_o4 = pickle.dumps(o4)

        # Relies on dict ordering
        if not jython:
            self.assert_(pk_o3 == pk_o4)

        # and lastly make sure we still have our data after all that.
        # identical serialzation is great, *if* it's complete :)
        self.assert_(o4.user_id == 7)
        self.assert_(o4.user_name is None)
        self.assert_(o4.email_address is None)
        self.assert_(o4.some_mutable_data == [1,2,3])
        self.assert_(len(o4.mt2) == 1)
        self.assert_(o4.mt2[0].a == 'abcde')
        self.assert_(o4.mt2[0].b is None)
    
    def test_state_gc(self):
        """test that InstanceState always has a dict, even after host object gc'ed."""
        
        class Foo(object):
            pass
        
        attributes.register_class(Foo)
        f = Foo()
        state = attributes.instance_state(f)
        f.bar = "foo"
        assert state.dict == {'bar':'foo', state.manager.STATE_ATTR:state}
        del f
        gc_collect()
        assert state.obj() is None
        assert state.dict == {}
        
    def test_deferred(self):
        class Foo(object):pass

        data = {'a':'this is a', 'b':12}
        def loader(state, keys):
            for k in keys:
                state.dict[k] = data[k]
            return attributes.ATTR_WAS_SET

        attributes.register_class(Foo)
        manager = attributes.manager_of_class(Foo)
        manager.deferred_scalar_loader = loader
        attributes.register_attribute(Foo, 'a', uselist=False, useobject=False)
        attributes.register_attribute(Foo, 'b', uselist=False, useobject=False)

        f = Foo()
        attributes.instance_state(f).expire_attributes(attributes.instance_dict(f), None)
        eq_(f.a, "this is a")
        eq_(f.b, 12)

        f.a = "this is some new a"
        attributes.instance_state(f).expire_attributes(attributes.instance_dict(f), None)
        eq_(f.a, "this is a")
        eq_(f.b, 12)

        attributes.instance_state(f).expire_attributes(attributes.instance_dict(f), None)
        f.a = "this is another new a"
        eq_(f.a, "this is another new a")
        eq_(f.b, 12)

        attributes.instance_state(f).expire_attributes(attributes.instance_dict(f), None)
        eq_(f.a, "this is a")
        eq_(f.b, 12)

        del f.a
        eq_(f.a, None)
        eq_(f.b, 12)

        attributes.instance_state(f).commit_all(attributes.instance_dict(f))
        eq_(f.a, None)
        eq_(f.b, 12)

    def test_deferred_pickleable(self):
        data = {'a':'this is a', 'b':12}
        def loader(state, keys):
            for k in keys:
                state.dict[k] = data[k]
            return attributes.ATTR_WAS_SET

        attributes.register_class(MyTest)
        manager = attributes.manager_of_class(MyTest)
        manager.deferred_scalar_loader=loader
        attributes.register_attribute(MyTest, 'a', uselist=False, useobject=False)
        attributes.register_attribute(MyTest, 'b', uselist=False, useobject=False)

        m = MyTest()
        attributes.instance_state(m).expire_attributes(attributes.instance_dict(m), None)
        assert 'a' not in m.__dict__
        m2 = pickle.loads(pickle.dumps(m))
        assert 'a' not in m2.__dict__
        eq_(m2.a, "this is a")
        eq_(m2.b, 12)

    def test_list(self):
        class User(object):pass
        class Address(object):pass

        attributes.register_class(User)
        attributes.register_class(Address)
        attributes.register_attribute(User, 'user_id', uselist=False, useobject=False)
        attributes.register_attribute(User, 'user_name', uselist=False, useobject=False)
        attributes.register_attribute(User, 'addresses', uselist = True, useobject=True)
        attributes.register_attribute(Address, 'address_id', uselist=False, useobject=False)
        attributes.register_attribute(Address, 'email_address', uselist=False, useobject=False)

        u = User()
        u.user_id = 7
        u.user_name = 'john'
        u.addresses = []
        a = Address()
        a.address_id = 10
        a.email_address = 'lala@123.com'
        u.addresses.append(a)

        self.assert_(u.user_id == 7 and u.user_name == 'john' and u.addresses[0].email_address == 'lala@123.com')
        u, attributes.instance_state(a).commit_all(attributes.instance_dict(a))
        self.assert_(u.user_id == 7 and u.user_name == 'john' and u.addresses[0].email_address == 'lala@123.com')

        u.user_name = 'heythere'
        a = Address()
        a.address_id = 11
        a.email_address = 'foo@bar.com'
        u.addresses.append(a)
        self.assert_(u.user_id == 7 and u.user_name == 'heythere' and u.addresses[0].email_address == 'lala@123.com' and u.addresses[1].email_address == 'foo@bar.com')
    
    def test_extension_commit_attr(self):
        """test that an extension which commits attribute history
        maintains the end-result history.
        
        This won't work in conjunction with some unitofwork extensions.
        
        """
        
        class Foo(_base.ComparableEntity):
            pass
        class Bar(_base.ComparableEntity):
            pass
        
        class ReceiveEvents(AttributeExtension):
            def __init__(self, key):
                self.key = key
                
            def append(self, state, child, initiator):
                if commit:
                    state.commit_all(state.dict)
                return child

            def remove(self, state, child, initiator):
                if commit:
                    state.commit_all(state.dict)
                return child

            def set(self, state, child, oldchild, initiator):
                if commit:
                    state.commit_all(state.dict)
                return child

        attributes.register_class(Foo)
        attributes.register_class(Bar)

        b1, b2, b3, b4 = Bar(id='b1'), Bar(id='b2'), Bar(id='b3'), Bar(id='b4')
        
        def loadcollection(**kw):
            if kw.get('passive') is attributes.PASSIVE_NO_FETCH:
                return attributes.PASSIVE_NO_RESULT
            return [b1, b2]
        
        def loadscalar(**kw):
            if kw.get('passive') is attributes.PASSIVE_NO_FETCH:
                return attributes.PASSIVE_NO_RESULT
            return b2
            
        attributes.register_attribute(Foo, 'bars', 
                               uselist=True, 
                               useobject=True, 
                               callable_=lambda o:loadcollection,
                               extension=[ReceiveEvents('bars')])
                               
        attributes.register_attribute(Foo, 'bar', 
                              uselist=False, 
                              useobject=True, 
                              callable_=lambda o:loadscalar,
                              extension=[ReceiveEvents('bar')])
                              
        attributes.register_attribute(Foo, 'scalar', 
                            uselist=False, 
                            useobject=False, extension=[ReceiveEvents('scalar')])
        
            
        def create_hist():
            def hist(key, shouldmatch, fn, *arg):
                attributes.instance_state(f1).commit_all(attributes.instance_dict(f1))
                fn(*arg)
                histories.append((shouldmatch, attributes.get_history(f1, key)))

            f1 = Foo()
            hist('bars', True, f1.bars.append, b3)
            hist('bars', True, f1.bars.append, b4)
            hist('bars', False, f1.bars.remove, b2)
            hist('bar', True, setattr, f1, 'bar', b3)
            hist('bar', True, setattr, f1, 'bar', None)
            hist('bar', True, setattr, f1, 'bar', b4)
            hist('scalar', True, setattr, f1, 'scalar', 5)
            hist('scalar', True, setattr, f1, 'scalar', None)
            hist('scalar', True, setattr, f1, 'scalar', 4)
        
        histories = []
        commit = False
        create_hist()
        without_commit = list(histories)
        histories[:] = []
        commit = True
        create_hist()
        with_commit = histories
        for without, with_ in zip(without_commit, with_commit):
            shouldmatch, woc = without
            shouldmatch, wic = with_
            if shouldmatch:
                eq_(woc, wic)
            else:
                ne_(woc, wic)
        
    def test_extension_lazyload_assertion(self):
        class Foo(_base.BasicEntity):
            pass
        class Bar(_base.BasicEntity):
            pass

        class ReceiveEvents(AttributeExtension):
            def append(self, state, child, initiator):
                state.obj().bars
                return child

            def remove(self, state, child, initiator):
                state.obj().bars
                return child

            def set(self, state, child, oldchild, initiator):
                return child

        attributes.register_class(Foo)
        attributes.register_class(Bar)

        bar1, bar2, bar3 = [Bar(id=1), Bar(id=2), Bar(id=3)]
        def func1(**kw):
            if kw.get('passive') is attributes.PASSIVE_NO_FETCH:
                return attributes.PASSIVE_NO_RESULT
            
            return [bar1, bar2, bar3]

        attributes.register_attribute(Foo, 'bars', uselist=True, callable_=lambda o:func1, useobject=True, extension=[ReceiveEvents()])
        attributes.register_attribute(Bar, 'foos', uselist=True, useobject=True, extension=[attributes.GenericBackrefExtension('bars')])

        x = Foo()
        assert_raises(AssertionError, Bar(id=4).foos.append, x)
        
        x.bars
        b = Bar(id=4)
        b.foos.append(x)
        attributes.instance_state(x).expire_attributes(attributes.instance_dict(x), ['bars'])
        assert_raises(AssertionError, b.foos.remove, x)
        
        
    def test_scalar_listener(self):
        # listeners on ScalarAttributeImpl and MutableScalarAttributeImpl aren't used normally.
        # test that they work for the benefit of user extensions
        class Foo(object):
            pass
        
        results = []
        class ReceiveEvents(AttributeExtension):
            def append(self, state, child, initiator):
                assert False

            def remove(self, state, child, initiator):
                results.append(("remove", state.obj(), child))

            def set(self, state, child, oldchild, initiator):
                results.append(("set", state.obj(), child, oldchild))
                return child
        
        attributes.register_class(Foo)
        attributes.register_attribute(Foo, 'x', uselist=False, mutable_scalars=False, useobject=False, extension=ReceiveEvents())
        attributes.register_attribute(Foo, 'y', uselist=False, mutable_scalars=True, useobject=False, copy_function=lambda x:x, extension=ReceiveEvents())
        
        f = Foo()
        f.x = 5
        f.x = 17
        del f.x
        f.y = [1,2,3]
        f.y = [4,5,6]
        del f.y
        
        eq_(results, [
            ('set', f, 5, None),
            ('set', f, 17, 5),
            ('remove', f, 17),
            ('set', f, [1,2,3], None),
            ('set', f, [4,5,6], [1,2,3]),
            ('remove', f, [4,5,6])
        ])
        
        
    def test_lazytrackparent(self):
        """test that the "hasparent" flag works properly 
           when lazy loaders and backrefs are used
           
        """

        class Post(object):pass
        class Blog(object):pass
        attributes.register_class(Post)
        attributes.register_class(Blog)

        # set up instrumented attributes with backrefs
        attributes.register_attribute(Post, 'blog', uselist=False,
                                        extension=attributes.GenericBackrefExtension('posts'),
                                        trackparent=True, useobject=True)
        attributes.register_attribute(Blog, 'posts', uselist=True,
                                        extension=attributes.GenericBackrefExtension('blog'),
                                        trackparent=True, useobject=True)

        # create objects as if they'd been freshly loaded from the database (without history)
        b = Blog()
        p1 = Post()
        attributes.instance_state(b).set_callable(attributes.instance_dict(b), 
                                                    'posts', lambda **kw:[p1])
        attributes.instance_state(p1).set_callable(attributes.instance_dict(p1), 
                                                    'blog', lambda **kw:b)
        p1, attributes.instance_state(b).commit_all(attributes.instance_dict(b))

        # no orphans (called before the lazy loaders fire off)
        assert attributes.has_parent(Blog, p1, 'posts', optimistic=True)
        assert attributes.has_parent(Post, b, 'blog', optimistic=True)

        # assert connections
        assert p1.blog is b
        assert p1 in b.posts

        # manual connections
        b2 = Blog()
        p2 = Post()
        b2.posts.append(p2)
        assert attributes.has_parent(Blog, p2, 'posts')
        assert attributes.has_parent(Post, b2, 'blog')

    def test_inheritance(self):
        """tests that attributes are polymorphic"""
        class Foo(object):pass
        class Bar(Foo):pass


        attributes.register_class(Foo)
        attributes.register_class(Bar)

        def func1(**kw):
            return "this is the foo attr"
        def func2(**kw):
            return "this is the bar attr"
        def func3(**kw):
            return "this is the shared attr"
        attributes.register_attribute(Foo, 'element', uselist=False, 
                                            callable_=lambda o:func1, useobject=True)
        attributes.register_attribute(Foo, 'element2', uselist=False, 
                                            callable_=lambda o:func3, useobject=True)
        attributes.register_attribute(Bar, 'element', uselist=False, 
                                            callable_=lambda o:func2, useobject=True)

        x = Foo()
        y = Bar()
        assert x.element == 'this is the foo attr'
        assert y.element == 'this is the bar attr'
        assert x.element2 == 'this is the shared attr'
        assert y.element2 == 'this is the shared attr'

    def test_no_double_state(self):
        states = set()
        class Foo(object):
            def __init__(self):
                states.add(attributes.instance_state(self))
        class Bar(Foo):
            def __init__(self):
                states.add(attributes.instance_state(self))
                Foo.__init__(self)


        attributes.register_class(Foo)
        attributes.register_class(Bar)

        b = Bar()
        eq_(len(states), 1)
        eq_(list(states)[0].obj(), b)


    def test_inheritance2(self):
        """test that the attribute manager can properly traverse the managed attributes of an object,
        if the object is of a descendant class with managed attributes in the parent class"""
        class Foo(object):pass
        class Bar(Foo):pass

        class Element(object):
            _state = True

        attributes.register_class(Foo)
        attributes.register_class(Bar)
        attributes.register_attribute(Foo, 'element', uselist=False, useobject=True)
        el = Element()
        x = Bar()
        x.element = el
        eq_(attributes.get_state_history(attributes.instance_state(x), 'element'), ([el], (), ()))
        attributes.instance_state(x).commit_all(attributes.instance_dict(x))

        (added, unchanged, deleted) = attributes.get_state_history(attributes.instance_state(x), 'element')
        assert added == ()
        assert unchanged == [el]

    def test_lazyhistory(self):
        """tests that history functions work with lazy-loading attributes"""

        class Foo(_base.BasicEntity):
            pass
        class Bar(_base.BasicEntity):
            pass

        attributes.register_class(Foo)
        attributes.register_class(Bar)

        bar1, bar2, bar3, bar4 = [Bar(id=1), Bar(id=2), Bar(id=3), Bar(id=4)]
        def func1(**kw):
            return "this is func 1"
        def func2(**kw):
            return [bar1, bar2, bar3]

        attributes.register_attribute(Foo, 'col1', uselist=False, callable_=lambda o:func1, useobject=True)
        attributes.register_attribute(Foo, 'col2', uselist=True, callable_=lambda o:func2, useobject=True)
        attributes.register_attribute(Bar, 'id', uselist=False, useobject=True)

        x = Foo()
        attributes.instance_state(x).commit_all(attributes.instance_dict(x))
        x.col2.append(bar4)
        eq_(attributes.get_state_history(attributes.instance_state(x), 'col2'), ([bar4], [bar1, bar2, bar3], []))

    def test_parenttrack(self):
        class Foo(object):pass
        class Bar(object):pass

        attributes.register_class(Foo)
        attributes.register_class(Bar)

        attributes.register_attribute(Foo, 'element', uselist=False, trackparent=True, useobject=True)
        attributes.register_attribute(Bar, 'element', uselist=False, trackparent=True, useobject=True)

        f1 = Foo()
        f2 = Foo()
        b1 = Bar()
        b2 = Bar()

        f1.element = b1
        b2.element = f2

        assert attributes.has_parent(Foo, b1, 'element')
        assert not attributes.has_parent(Foo, b2, 'element')
        assert not attributes.has_parent(Foo, f2, 'element')
        assert attributes.has_parent(Bar, f2, 'element')

        b2.element = None
        assert not attributes.has_parent(Bar, f2, 'element')

        # test that double assignment doesn't accidentally reset the 'parent' flag.
        b3 = Bar()
        f4 = Foo()
        b3.element = f4
        assert attributes.has_parent(Bar, f4, 'element')
        b3.element = f4
        assert attributes.has_parent(Bar, f4, 'element')

    def test_mutablescalars(self):
        """test detection of changes on mutable scalar items"""
        class Foo(object):pass

        attributes.register_class(Foo)
        attributes.register_attribute(Foo, 'element', uselist=False, copy_function=lambda x:[y for y in x], mutable_scalars=True, useobject=False)
        x = Foo()
        x.element = ['one', 'two', 'three']
        attributes.instance_state(x).commit_all(attributes.instance_dict(x))
        x.element[1] = 'five'
        assert attributes.instance_state(x).modified

        attributes.unregister_class(Foo)

        attributes.register_class(Foo)
        attributes.register_attribute(Foo, 'element', uselist=False, useobject=False)
        x = Foo()
        x.element = ['one', 'two', 'three']
        attributes.instance_state(x).commit_all(attributes.instance_dict(x))
        x.element[1] = 'five'
        assert not attributes.instance_state(x).modified

    def test_descriptorattributes(self):
        """changeset: 1633 broke ability to use ORM to map classes with unusual
        descriptor attributes (for example, classes that inherit from ones
        implementing zope.interface.Interface).
        This is a simple regression test to prevent that defect.
        """
        class des(object):
            def __get__(self, instance, owner):
                raise AttributeError('fake attribute')

        class Foo(object):
            A = des()

        attributes.register_class(Foo)
        attributes.unregister_class(Foo)

    def test_collectionclasses(self):

        class Foo(object):pass
        attributes.register_class(Foo)

        attributes.register_attribute(Foo, "collection", uselist=True, typecallable=set, useobject=True)
        assert attributes.manager_of_class(Foo).is_instrumented("collection")
        assert isinstance(Foo().collection, set)

        attributes.unregister_attribute(Foo, "collection")
        assert not attributes.manager_of_class(Foo).is_instrumented("collection")
        
        try:
            attributes.register_attribute(Foo, "collection", uselist=True, typecallable=dict, useobject=True)
            assert False
        except sa_exc.ArgumentError, e:
            assert str(e) == "Type InstrumentedDict must elect an appender method to be a collection class"

        class MyDict(dict):
            @collection.appender
            def append(self, item):
                self[item.foo] = item
            @collection.remover
            def remove(self, item):
                del self[item.foo]
        attributes.register_attribute(Foo, "collection", uselist=True, typecallable=MyDict, useobject=True)
        assert isinstance(Foo().collection, MyDict)

        attributes.unregister_attribute(Foo, "collection")

        class MyColl(object):pass
        try:
            attributes.register_attribute(Foo, "collection", uselist=True, typecallable=MyColl, useobject=True)
            assert False
        except sa_exc.ArgumentError, e:
            assert str(e) == "Type MyColl must elect an appender method to be a collection class"

        class MyColl(object):
            @collection.iterator
            def __iter__(self):
                return iter([])
            @collection.appender
            def append(self, item):
                pass
            @collection.remover
            def remove(self, item):
                pass
        attributes.register_attribute(Foo, "collection", uselist=True, typecallable=MyColl, useobject=True)
        try:
            Foo().collection
            assert True
        except sa_exc.ArgumentError, e:
            assert False

class UtilTest(_base.ORMTest):
    def test_helpers(self):
        class Foo(object):
            pass

        class Bar(object):
            pass
        
        attributes.register_class(Foo)
        attributes.register_class(Bar)
        attributes.register_attribute(Foo, "coll", uselist=True, useobject=True)
    
        f1 = Foo()
        b1 = Bar()
        b2 = Bar()
        coll = attributes.init_collection(f1, "coll")
        assert coll.data is f1.coll
        assert attributes.get_attribute(f1, "coll") is f1.coll
        attributes.set_attribute(f1, "coll", [b1])
        assert f1.coll == [b1]
        eq_(attributes.get_history(f1, "coll"), ([b1], [], []))
        attributes.set_committed_value(f1, "coll", [b2])
        eq_(attributes.get_history(f1, "coll"), ((), [b2], ()))
        
        attributes.del_attribute(f1, "coll")
        assert "coll" not in f1.__dict__

class BackrefTest(_base.ORMTest):

    def test_manytomany(self):
        class Student(object):pass
        class Course(object):pass

        attributes.register_class(Student)
        attributes.register_class(Course)
        attributes.register_attribute(Student, 'courses', uselist=True, extension=attributes.GenericBackrefExtension('students'), useobject=True)
        attributes.register_attribute(Course, 'students', uselist=True, extension=attributes.GenericBackrefExtension('courses'), useobject=True)

        s = Student()
        c = Course()
        s.courses.append(c)
        self.assert_(c.students == [s])
        s.courses.remove(c)
        self.assert_(c.students == [])

        (s1, s2, s3) = (Student(), Student(), Student())

        c.students = [s1, s2, s3]
        self.assert_(s2.courses == [c])
        self.assert_(s1.courses == [c])
        s1.courses.remove(c)
        self.assert_(c.students == [s2,s3])

    def test_onetomany(self):
        class Post(object):pass
        class Blog(object):pass

        attributes.register_class(Post)
        attributes.register_class(Blog)
        attributes.register_attribute(Post, 'blog', uselist=False, extension=attributes.GenericBackrefExtension('posts'), trackparent=True, useobject=True)
        attributes.register_attribute(Blog, 'posts', uselist=True, extension=attributes.GenericBackrefExtension('blog'), trackparent=True, useobject=True)
        b = Blog()
        (p1, p2, p3) = (Post(), Post(), Post())
        b.posts.append(p1)
        b.posts.append(p2)
        b.posts.append(p3)
        self.assert_(b.posts == [p1, p2, p3])
        self.assert_(p2.blog is b)

        p3.blog = None
        self.assert_(b.posts == [p1, p2])
        p4 = Post()
        p4.blog = b
        self.assert_(b.posts == [p1, p2, p4])

        p4.blog = b
        p4.blog = b
        self.assert_(b.posts == [p1, p2, p4])

        # assert no failure removing None
        p5 = Post()
        p5.blog = None
        del p5.blog

    def test_onetoone(self):
        class Port(object):pass
        class Jack(object):pass
        attributes.register_class(Port)
        attributes.register_class(Jack)
        attributes.register_attribute(Port, 'jack', uselist=False, extension=attributes.GenericBackrefExtension('port'), useobject=True)
        attributes.register_attribute(Jack, 'port', uselist=False, extension=attributes.GenericBackrefExtension('jack'), useobject=True)
        p = Port()
        j = Jack()
        p.jack = j
        self.assert_(j.port is p)
        self.assert_(p.jack is not None)

        j.port = None
        self.assert_(p.jack is None)

class PendingBackrefTest(_base.ORMTest):
    def setup(self):
        global Post, Blog, called, lazy_load

        class Post(object):
            def __init__(self, name):
                self.name = name
            __hash__ = None
            def __eq__(self, other):
                return other is not None and other.name == self.name

        class Blog(object):
            def __init__(self, name):
                self.name = name
            __hash__ = None
            def __eq__(self, other):
                return other is not None and other.name == self.name

        called = [0]

        lazy_load = []
        def lazy_posts(instance):
            def load(**kw):
                if kw['passive'] is not attributes.PASSIVE_NO_FETCH:
                    called[0] += 1
                    return lazy_load
                else:
                    return attributes.PASSIVE_NO_RESULT
            return load

        attributes.register_class(Post)
        attributes.register_class(Blog)
        attributes.register_attribute(Post, 'blog', uselist=False, extension=attributes.GenericBackrefExtension('posts'), trackparent=True, useobject=True)
        attributes.register_attribute(Blog, 'posts', uselist=True, extension=attributes.GenericBackrefExtension('blog'), callable_=lazy_posts, trackparent=True, useobject=True)

    def test_lazy_add(self):
        global lazy_load

        p1, p2, p3 = Post("post 1"), Post("post 2"), Post("post 3")
        lazy_load = [p1, p2, p3]

        b = Blog("blog 1")
        p = Post("post 4")
        
        p.blog = b
        p = Post("post 5")
        p.blog = b
        # setting blog doesnt call 'posts' callable
        assert called[0] == 0

        # calling backref calls the callable, populates extra posts
        assert b.posts == [p1, p2, p3, Post("post 4"), Post("post 5")]
        assert called[0] == 1
    
    def test_lazy_history(self):
        global lazy_load

        p1, p2, p3 = Post("post 1"), Post("post 2"), Post("post 3")
        lazy_load = [p1, p2, p3]
        
        b = Blog("blog 1")
        p = Post("post 4")
        p.blog = b
        
        p4 = Post("post 5")
        p4.blog = b
        assert called[0] == 0
        eq_(attributes.instance_state(b).get_history('posts'), ([p, p4], [p1, p2, p3], []))
        assert called[0] == 1

    def test_lazy_remove(self):
        global lazy_load
        called[0] = 0
        lazy_load = []

        b = Blog("blog 1")
        p = Post("post 1")
        p.blog = b
        assert called[0] == 0

        lazy_load = [p]

        p.blog = None
        p2 = Post("post 2")
        p2.blog = b
        assert called[0] == 0
        assert b.posts == [p2]
        assert called[0] == 1

    def test_normal_load(self):
        global lazy_load
        lazy_load = (p1, p2, p3) = [Post("post 1"), Post("post 2"), Post("post 3")]
        called[0] = 0

        b = Blog("blog 1")

        # assign without using backref system
        p2.__dict__['blog'] = b

        assert b.posts == [Post("post 1"), Post("post 2"), Post("post 3")]
        assert called[0] == 1
        p2.blog = None
        p4 = Post("post 4")
        p4.blog = b
        assert b.posts == [Post("post 1"), Post("post 3"), Post("post 4")]
        assert called[0] == 1

        called[0] = 0
        lazy_load = (p1, p2, p3) = [Post("post 1"), Post("post 2"), Post("post 3")]

    def test_commit_removes_pending(self):
        global lazy_load
        lazy_load = (p1, ) = [Post("post 1"), ]
        called[0] = 0

        b = Blog("blog 1")
        p1.blog = b
        attributes.instance_state(b).commit_all(attributes.instance_dict(b))
        attributes.instance_state(p1).commit_all(attributes.instance_dict(p1))
        assert b.posts == [Post("post 1")]

class HistoryTest(_base.ORMTest):

    def test_get_committed_value(self):
        class Foo(_base.BasicEntity):
            pass

        attributes.register_class(Foo)
        attributes.register_attribute(Foo, 'someattr', uselist=False, useobject=False)

        f = Foo()
        eq_(Foo.someattr.impl.get_committed_value(attributes.instance_state(f), attributes.instance_dict(f)), None)

        f.someattr = 3
        eq_(Foo.someattr.impl.get_committed_value(attributes.instance_state(f), attributes.instance_dict(f)), None)

        f = Foo()
        f.someattr = 3
        eq_(Foo.someattr.impl.get_committed_value(attributes.instance_state(f), attributes.instance_dict(f)), None)
        
        attributes.instance_state(f).commit(attributes.instance_dict(f), ['someattr'])
        eq_(Foo.someattr.impl.get_committed_value(attributes.instance_state(f), attributes.instance_dict(f)), 3)

    def test_scalar(self):
        class Foo(_base.BasicEntity):
            pass

        attributes.register_class(Foo)
        attributes.register_attribute(Foo, 'someattr', uselist=False, useobject=False)

        # case 1.  new object
        f = Foo()
        eq_(attributes.get_state_history(attributes.instance_state(f), 'someattr'), ((), (), ()))

        f.someattr = "hi"
        eq_(attributes.get_state_history(attributes.instance_state(f), 'someattr'), (['hi'], (), ()))

        attributes.instance_state(f).commit(attributes.instance_dict(f), ['someattr'])
        eq_(attributes.get_state_history(attributes.instance_state(f), 'someattr'), ((), ['hi'], ()))

        f.someattr = 'there'

        eq_(attributes.get_state_history(attributes.instance_state(f), 'someattr'), (['there'], (), ['hi']))
        attributes.instance_state(f).commit(attributes.instance_dict(f), ['someattr'])

        eq_(attributes.get_state_history(attributes.instance_state(f), 'someattr'), ((), ['there'], ()))

        del f.someattr
        eq_(attributes.get_state_history(attributes.instance_state(f), 'someattr'), ((), (), ['there']))

        # case 2.  object with direct dictionary settings (similar to a load operation)
        f = Foo()
        f.__dict__['someattr'] = 'new'
        eq_(attributes.get_state_history(attributes.instance_state(f), 'someattr'), ((), ['new'], ()))

        f.someattr = 'old'
        eq_(attributes.get_state_history(attributes.instance_state(f), 'someattr'), (['old'], (), ['new']))

        attributes.instance_state(f).commit(attributes.instance_dict(f), ['someattr'])
        eq_(attributes.get_state_history(attributes.instance_state(f), 'someattr'), ((), ['old'], ()))

        # setting None on uninitialized is currently a change for a scalar attribute
        # no lazyload occurs so this allows overwrite operation to proceed
        f = Foo()
        eq_(attributes.get_state_history(attributes.instance_state(f), 'someattr'), ((), (), ()))
        f.someattr = None
        eq_(attributes.get_state_history(attributes.instance_state(f), 'someattr'), ([None], (), ()))

        f = Foo()
        f.__dict__['someattr'] = 'new'
        eq_(attributes.get_state_history(attributes.instance_state(f), 'someattr'), ((), ['new'], ()))
        f.someattr = None
        eq_(attributes.get_state_history(attributes.instance_state(f), 'someattr'), ([None], (), ['new']))

        # set same value twice
        f = Foo()
        attributes.instance_state(f).commit(attributes.instance_dict(f), ['someattr'])
        f.someattr = 'one'
        eq_(attributes.get_state_history(attributes.instance_state(f), 'someattr'), (['one'], (), ()))
        f.someattr = 'two'
        eq_(attributes.get_state_history(attributes.instance_state(f), 'someattr'), (['two'], (), ()))
        
        
    def test_mutable_scalar(self):
        class Foo(_base.BasicEntity):
            pass

        attributes.register_class(Foo)
        attributes.register_attribute(Foo, 'someattr', uselist=False, useobject=False, mutable_scalars=True, copy_function=dict)

        # case 1.  new object
        f = Foo()
        eq_(attributes.get_state_history(attributes.instance_state(f), 'someattr'), ((), (), ()))

        f.someattr = {'foo':'hi'}
        eq_(attributes.get_state_history(attributes.instance_state(f), 'someattr'), ([{'foo':'hi'}], (), ()))

        attributes.instance_state(f).commit(attributes.instance_dict(f), ['someattr'])
        eq_(attributes.get_state_history(attributes.instance_state(f), 'someattr'), ((), [{'foo':'hi'}], ()))
        eq_(attributes.instance_state(f).committed_state['someattr'], {'foo':'hi'})

        f.someattr['foo'] = 'there'
        eq_(attributes.instance_state(f).committed_state['someattr'], {'foo':'hi'})

        eq_(attributes.get_state_history(attributes.instance_state(f), 'someattr'), ([{'foo':'there'}], (), [{'foo':'hi'}]))
        attributes.instance_state(f).commit(attributes.instance_dict(f), ['someattr'])

        eq_(attributes.get_state_history(attributes.instance_state(f), 'someattr'), ((), [{'foo':'there'}], ()))

        # case 2.  object with direct dictionary settings (similar to a load operation)
        f = Foo()
        f.__dict__['someattr'] = {'foo':'new'}
        eq_(attributes.get_state_history(attributes.instance_state(f), 'someattr'), ((), [{'foo':'new'}], ()))

        f.someattr = {'foo':'old'}
        eq_(attributes.get_state_history(attributes.instance_state(f), 'someattr'), ([{'foo':'old'}], (), [{'foo':'new'}]))

        attributes.instance_state(f).commit(attributes.instance_dict(f), ['someattr'])
        eq_(attributes.get_state_history(attributes.instance_state(f), 'someattr'), ((), [{'foo':'old'}], ()))


    def test_use_object(self):
        class Foo(_base.BasicEntity):
            pass

        class Bar(_base.BasicEntity):
            _state = None
            def __nonzero__(self):
                assert False

        hi = Bar(name='hi')
        there = Bar(name='there')
        new = Bar(name='new')
        old = Bar(name='old')

        attributes.register_class(Foo)
        attributes.register_attribute(Foo, 'someattr', uselist=False, useobject=True)

        # case 1.  new object
        f = Foo()
        eq_(attributes.get_state_history(attributes.instance_state(f), 'someattr'), ((), [None], ()))

        f.someattr = hi
        eq_(attributes.get_state_history(attributes.instance_state(f), 'someattr'), ([hi], (), ()))

        attributes.instance_state(f).commit(attributes.instance_dict(f), ['someattr'])
        eq_(attributes.get_state_history(attributes.instance_state(f), 'someattr'), ((), [hi], ()))

        f.someattr = there

        eq_(attributes.get_state_history(attributes.instance_state(f), 'someattr'), ([there], (), [hi]))
        attributes.instance_state(f).commit(attributes.instance_dict(f), ['someattr'])

        eq_(attributes.get_state_history(attributes.instance_state(f), 'someattr'), ((), [there], ()))

        del f.someattr
        eq_(attributes.get_state_history(attributes.instance_state(f), 'someattr'), ([None], (), [there]))

        # case 2.  object with direct dictionary settings (similar to a load operation)
        f = Foo()
        f.__dict__['someattr'] = 'new'
        eq_(attributes.get_state_history(attributes.instance_state(f), 'someattr'), ((), ['new'], ()))

        f.someattr = old
        eq_(attributes.get_state_history(attributes.instance_state(f), 'someattr'), ([old], (), ['new']))

        attributes.instance_state(f).commit(attributes.instance_dict(f), ['someattr'])
        eq_(attributes.get_state_history(attributes.instance_state(f), 'someattr'), ((), [old], ()))

        # setting None on uninitialized is currently not a change for an object attribute
        # (this is different than scalar attribute).  a lazyload has occured so if its
        # None, its really None
        f = Foo()
        eq_(attributes.get_state_history(attributes.instance_state(f), 'someattr'), ((), [None], ()))
        f.someattr = None
        eq_(attributes.get_state_history(attributes.instance_state(f), 'someattr'), ((), [None], ()))

        f = Foo()
        f.__dict__['someattr'] = 'new'
        eq_(attributes.get_state_history(attributes.instance_state(f), 'someattr'), ((), ['new'], ()))
        f.someattr = None
        eq_(attributes.get_state_history(attributes.instance_state(f), 'someattr'), ([None], (), ['new']))

        # set same value twice
        f = Foo()
        attributes.instance_state(f).commit(attributes.instance_dict(f), ['someattr'])
        f.someattr = 'one'
        eq_(attributes.get_state_history(attributes.instance_state(f), 'someattr'), (['one'], (), ()))
        f.someattr = 'two'
        eq_(attributes.get_state_history(attributes.instance_state(f), 'someattr'), (['two'], (), ()))

    def test_object_collections_set(self):
        class Foo(_base.BasicEntity):
            pass
        class Bar(_base.BasicEntity):
            def __nonzero__(self):
                assert False

        attributes.register_class(Foo)
        attributes.register_attribute(Foo, 'someattr', uselist=True, useobject=True)

        hi = Bar(name='hi')
        there = Bar(name='there')
        old = Bar(name='old')
        new = Bar(name='new')

        # case 1.  new object
        f = Foo()
        eq_(attributes.get_state_history(attributes.instance_state(f), 'someattr'), ((), [], ()))

        f.someattr = [hi]
        eq_(attributes.get_state_history(attributes.instance_state(f), 'someattr'), ([hi], [], []))

        attributes.instance_state(f).commit(attributes.instance_dict(f), ['someattr'])
        eq_(attributes.get_state_history(attributes.instance_state(f), 'someattr'), ((), [hi], ()))

        f.someattr = [there]

        eq_(attributes.get_state_history(attributes.instance_state(f), 'someattr'), ([there], [], [hi]))
        attributes.instance_state(f).commit(attributes.instance_dict(f), ['someattr'])

        eq_(attributes.get_state_history(attributes.instance_state(f), 'someattr'), ((), [there], ()))

        f.someattr = [hi]
        eq_(attributes.get_state_history(attributes.instance_state(f), 'someattr'), ([hi], [], [there]))

        f.someattr = [old, new]
        eq_(attributes.get_state_history(attributes.instance_state(f), 'someattr'), ([old, new], [], [there]))

        # case 2.  object with direct settings (similar to a load operation)
        f = Foo()
        collection = attributes.init_collection(f, 'someattr')
        collection.append_without_event(new)
        attributes.instance_state(f).commit_all(attributes.instance_dict(f))
        eq_(attributes.get_state_history(attributes.instance_state(f), 'someattr'), ((), [new], ()))

        f.someattr = [old]
        eq_(attributes.get_state_history(attributes.instance_state(f), 'someattr'), ([old], [], [new]))

        attributes.instance_state(f).commit(attributes.instance_dict(f), ['someattr'])
        eq_(attributes.get_state_history(attributes.instance_state(f), 'someattr'), ((), [old], ()))

    def test_dict_collections(self):
        class Foo(_base.BasicEntity):
            pass
        class Bar(_base.BasicEntity):
            pass

        from sqlalchemy.orm.collections import attribute_mapped_collection

        attributes.register_class(Foo)
        attributes.register_attribute(Foo, 'someattr', uselist=True, useobject=True, typecallable=attribute_mapped_collection('name'))

        hi = Bar(name='hi')
        there = Bar(name='there')
        old = Bar(name='old')
        new = Bar(name='new')

        f = Foo()
        eq_(attributes.get_state_history(attributes.instance_state(f), 'someattr'), ((), [], ()))

        f.someattr['hi'] = hi
        eq_(attributes.get_state_history(attributes.instance_state(f), 'someattr'), ([hi], [], []))

        f.someattr['there'] = there
        eq_(tuple([set(x) for x in attributes.get_state_history(attributes.instance_state(f), 'someattr')]), (set([hi, there]), set(), set()))

        attributes.instance_state(f).commit(attributes.instance_dict(f), ['someattr'])
        eq_(tuple([set(x) for x in attributes.get_state_history(attributes.instance_state(f), 'someattr')]), (set(), set([hi, there]), set()))

    def test_object_collections_mutate(self):
        class Foo(_base.BasicEntity):
            pass
        class Bar(_base.BasicEntity):
            pass

        attributes.register_class(Foo)
        attributes.register_attribute(Foo, 'someattr', uselist=True, useobject=True)
        attributes.register_attribute(Foo, 'id', uselist=False, useobject=False)

        hi = Bar(name='hi')
        there = Bar(name='there')
        old = Bar(name='old')
        new = Bar(name='new')

        # case 1.  new object
        f = Foo(id=1)
        eq_(attributes.get_state_history(attributes.instance_state(f), 'someattr'), ((), [], ()))

        f.someattr.append(hi)
        eq_(attributes.get_state_history(attributes.instance_state(f), 'someattr'), ([hi], [], []))

        attributes.instance_state(f).commit(attributes.instance_dict(f), ['someattr'])
        eq_(attributes.get_state_history(attributes.instance_state(f), 'someattr'), ((), [hi], ()))

        f.someattr.append(there)

        eq_(attributes.get_state_history(attributes.instance_state(f), 'someattr'), ([there], [hi], []))
        attributes.instance_state(f).commit(attributes.instance_dict(f), ['someattr'])

        eq_(attributes.get_state_history(attributes.instance_state(f), 'someattr'), ((), [hi, there], ()))

        f.someattr.remove(there)
        eq_(attributes.get_state_history(attributes.instance_state(f), 'someattr'), ([], [hi], [there]))

        f.someattr.append(old)
        f.someattr.append(new)
        eq_(attributes.get_state_history(attributes.instance_state(f), 'someattr'), ([old, new], [hi], [there]))
        attributes.instance_state(f).commit(attributes.instance_dict(f), ['someattr'])
        eq_(attributes.get_state_history(attributes.instance_state(f), 'someattr'), ((), [hi, old, new], ()))

        f.someattr.pop(0)
        eq_(attributes.get_state_history(attributes.instance_state(f), 'someattr'), ([], [old, new], [hi]))

        # case 2.  object with direct settings (similar to a load operation)
        f = Foo()
        f.__dict__['id'] = 1
        collection = attributes.init_collection(f, 'someattr')
        collection.append_without_event(new)
        attributes.instance_state(f).commit_all(attributes.instance_dict(f))
        eq_(attributes.get_state_history(attributes.instance_state(f), 'someattr'), ((), [new], ()))

        f.someattr.append(old)
        eq_(attributes.get_state_history(attributes.instance_state(f), 'someattr'), ([old], [new], []))

        attributes.instance_state(f).commit(attributes.instance_dict(f), ['someattr'])
        eq_(attributes.get_state_history(attributes.instance_state(f), 'someattr'), ((), [new, old], ()))

        f = Foo()
        collection = attributes.init_collection(f, 'someattr')
        collection.append_without_event(new)
        attributes.instance_state(f).commit_all(attributes.instance_dict(f))
        eq_(attributes.get_state_history(attributes.instance_state(f), 'someattr'), ((), [new], ()))

        f.id = 1
        f.someattr.remove(new)
        eq_(attributes.get_state_history(attributes.instance_state(f), 'someattr'), ([], [], [new]))

        # case 3.  mixing appends with sets
        f = Foo()
        f.someattr.append(hi)
        eq_(attributes.get_state_history(attributes.instance_state(f), 'someattr'), ([hi], [], []))
        f.someattr.append(there)
        eq_(attributes.get_state_history(attributes.instance_state(f), 'someattr'), ([hi, there], [], []))
        f.someattr = [there]
        eq_(attributes.get_state_history(attributes.instance_state(f), 'someattr'), ([there], [], []))

        # case 4.  ensure duplicates show up, order is maintained
        f = Foo()
        f.someattr.append(hi)
        f.someattr.append(there)
        f.someattr.append(hi)
        eq_(attributes.get_state_history(attributes.instance_state(f), 'someattr'), ([hi, there, hi], [], []))

        attributes.instance_state(f).commit_all(attributes.instance_dict(f))
        eq_(attributes.get_state_history(attributes.instance_state(f), 'someattr'), ((), [hi, there, hi], ()))
        
        f.someattr = []
        eq_(attributes.get_state_history(attributes.instance_state(f), 'someattr'), ([], [], [hi, there, hi]))
        
    def test_collections_via_backref(self):
        class Foo(_base.BasicEntity):
            pass
        class Bar(_base.BasicEntity):
            pass

        attributes.register_class(Foo)
        attributes.register_class(Bar)
        attributes.register_attribute(Foo, 'bars', uselist=True, extension=attributes.GenericBackrefExtension('foo'), trackparent=True, useobject=True)
        attributes.register_attribute(Bar, 'foo', uselist=False, extension=attributes.GenericBackrefExtension('bars'), trackparent=True, useobject=True)

        f1 = Foo()
        b1 = Bar()
        eq_(attributes.get_state_history(attributes.instance_state(f1), 'bars'), ((), [], ()))
        eq_(attributes.get_state_history(attributes.instance_state(b1), 'foo'), ((), [None], ()))

        #b1.foo = f1
        f1.bars.append(b1)
        eq_(attributes.get_state_history(attributes.instance_state(f1), 'bars'), ([b1], [], []))
        eq_(attributes.get_state_history(attributes.instance_state(b1), 'foo'), ([f1], (), ()))

        b2 = Bar()
        f1.bars.append(b2)
        eq_(attributes.get_state_history(attributes.instance_state(f1), 'bars'), ([b1, b2], [], []))
        eq_(attributes.get_state_history(attributes.instance_state(b1), 'foo'), ([f1], (), ()))
        eq_(attributes.get_state_history(attributes.instance_state(b2), 'foo'), ([f1], (), ()))

    def test_lazy_backref_collections(self):
        class Foo(_base.BasicEntity):
            pass
        class Bar(_base.BasicEntity):
            pass

        lazy_load = []
        def lazyload(instance):
            def load(**kw):
                return lazy_load
            return load

        attributes.register_class(Foo)
        attributes.register_class(Bar)
        attributes.register_attribute(Foo, 'bars', uselist=True, extension=attributes.GenericBackrefExtension('foo'), trackparent=True, callable_=lazyload, useobject=True)
        attributes.register_attribute(Bar, 'foo', uselist=False, extension=attributes.GenericBackrefExtension('bars'), trackparent=True, useobject=True)

        bar1, bar2, bar3, bar4 = [Bar(id=1), Bar(id=2), Bar(id=3), Bar(id=4)]
        lazy_load = [bar1, bar2, bar3]

        f = Foo()
        bar4 = Bar()
        bar4.foo = f
        eq_(attributes.get_state_history(attributes.instance_state(f), 'bars'), ([bar4], [bar1, bar2, bar3], []))

        lazy_load = None
        f = Foo()
        bar4 = Bar()
        bar4.foo = f
        eq_(attributes.get_state_history(attributes.instance_state(f), 'bars'), ([bar4], [], []))

        lazy_load = [bar1, bar2, bar3]
        attributes.instance_state(f).expire_attributes(attributes.instance_dict(f), ['bars'])
        eq_(attributes.get_state_history(attributes.instance_state(f), 'bars'), ((), [bar1, bar2, bar3], ()))

    def test_collections_via_lazyload(self):
        class Foo(_base.BasicEntity):
            pass
        class Bar(_base.BasicEntity):
            pass

        lazy_load = []
        def lazyload(instance):
            def load(**kw):
                return lazy_load
            return load

        attributes.register_class(Foo)
        attributes.register_class(Bar)
        attributes.register_attribute(Foo, 'bars', uselist=True, callable_=lazyload, trackparent=True, useobject=True)

        bar1, bar2, bar3, bar4 = [Bar(id=1), Bar(id=2), Bar(id=3), Bar(id=4)]
        lazy_load = [bar1, bar2, bar3]

        f = Foo()
        f.bars = []
        eq_(attributes.get_state_history(attributes.instance_state(f), 'bars'), ([], [], [bar1, bar2, bar3]))

        f = Foo()
        f.bars.append(bar4)
        eq_(attributes.get_state_history(attributes.instance_state(f), 'bars'), ([bar4], [bar1, bar2, bar3], []) )

        f = Foo()
        f.bars.remove(bar2)
        eq_(attributes.get_state_history(attributes.instance_state(f), 'bars'), ([], [bar1, bar3], [bar2]))
        f.bars.append(bar4)
        eq_(attributes.get_state_history(attributes.instance_state(f), 'bars'), ([bar4], [bar1, bar3], [bar2]))

        f = Foo()
        del f.bars[1]
        eq_(attributes.get_state_history(attributes.instance_state(f), 'bars'), ([], [bar1, bar3], [bar2]))

        lazy_load = None
        f = Foo()
        f.bars.append(bar2)
        eq_(attributes.get_state_history(attributes.instance_state(f), 'bars'), ([bar2], [], []))

    def test_scalar_via_lazyload(self):
        class Foo(_base.BasicEntity):
            pass

        lazy_load = None
        def lazyload(instance):
            def load(**kw):
                return lazy_load
            return load

        attributes.register_class(Foo)
        attributes.register_attribute(Foo, 'bar', uselist=False, callable_=lazyload, useobject=False)
        lazy_load = "hi"

        # with scalar non-object and active_history=False, the lazy callable is only executed on gets, not history
        # operations

        f = Foo()
        eq_(f.bar, "hi")
        eq_(attributes.get_state_history(attributes.instance_state(f), 'bar'), ((), ["hi"], ()))

        f = Foo()
        f.bar = None
        eq_(attributes.get_state_history(attributes.instance_state(f), 'bar'), ([None], (), ()))

        f = Foo()
        f.bar = "there"
        eq_(attributes.get_state_history(attributes.instance_state(f), 'bar'), (["there"], (), ()))
        f.bar = "hi"
        eq_(attributes.get_state_history(attributes.instance_state(f), 'bar'), (["hi"], (), ()))

        f = Foo()
        eq_(f.bar, "hi")
        del f.bar
        eq_(attributes.get_state_history(attributes.instance_state(f), 'bar'), ((), (), ["hi"]))
        assert f.bar is None
        eq_(attributes.get_state_history(attributes.instance_state(f), 'bar'), ([None], (), ["hi"]))

    def test_scalar_via_lazyload_with_active(self):
        class Foo(_base.BasicEntity):
            pass

        lazy_load = None
        def lazyload(instance):
            def load(**kw):
                return lazy_load
            return load

        attributes.register_class(Foo)
        attributes.register_attribute(Foo, 'bar', uselist=False, callable_=lazyload, useobject=False, active_history=True)
        lazy_load = "hi"

        # active_history=True means the lazy callable is executed on set as well as get,
        # causing the old value to appear in the history

        f = Foo()
        eq_(f.bar, "hi")
        eq_(attributes.get_state_history(attributes.instance_state(f), 'bar'), ((), ["hi"], ()))

        f = Foo()
        f.bar = None
        eq_(attributes.get_state_history(attributes.instance_state(f), 'bar'), ([None], (), ['hi']))

        f = Foo()
        f.bar = "there"
        eq_(attributes.get_state_history(attributes.instance_state(f), 'bar'), (["there"], (), ['hi']))
        f.bar = "hi"
        eq_(attributes.get_state_history(attributes.instance_state(f), 'bar'), ((), ["hi"], ()))

        f = Foo()
        eq_(f.bar, "hi")
        del f.bar
        eq_(attributes.get_state_history(attributes.instance_state(f), 'bar'), ((), (), ["hi"]))
        assert f.bar is None
        eq_(attributes.get_state_history(attributes.instance_state(f), 'bar'), ([None], (), ["hi"]))

    def test_scalar_object_via_lazyload(self):
        class Foo(_base.BasicEntity):
            pass
        class Bar(_base.BasicEntity):
            pass

        lazy_load = None
        def lazyload(instance):
            def load(**kw):
                return lazy_load
            return load

        attributes.register_class(Foo)
        attributes.register_class(Bar)
        attributes.register_attribute(Foo, 'bar', uselist=False, callable_=lazyload, trackparent=True, useobject=True)
        bar1, bar2 = [Bar(id=1), Bar(id=2)]
        lazy_load = bar1

        # with scalar object, the lazy callable is only executed on gets and history
        # operations

        f = Foo()
        eq_(attributes.get_state_history(attributes.instance_state(f), 'bar'), ((), [bar1], ()))

        f = Foo()
        f.bar = None
        eq_(attributes.get_state_history(attributes.instance_state(f), 'bar'), ([None], (), [bar1]))

        f = Foo()
        f.bar = bar2
        eq_(attributes.get_state_history(attributes.instance_state(f), 'bar'), ([bar2], (), [bar1]))
        f.bar = bar1
        eq_(attributes.get_state_history(attributes.instance_state(f), 'bar'), ((), [bar1], ()))

        f = Foo()
        eq_(f.bar, bar1)
        del f.bar
        eq_(attributes.get_state_history(attributes.instance_state(f), 'bar'), ([None], (), [bar1]))
        assert f.bar is None
        eq_(attributes.get_state_history(attributes.instance_state(f), 'bar'), ([None], (), [bar1]))

class ListenerTest(_base.ORMTest):
    def test_receive_changes(self):
        """test that Listeners can mutate the given value.
        
        This is a rudimentary test which would be better suited by a full-blown inclusion
        into collection.py.
        
        """
        class Foo(object):
            pass
        class Bar(object):
            pass

        class AlteringListener(AttributeExtension):
            def append(self, state, child, initiator):
                b2 = Bar()
                b2.data = b1.data + " appended"
                return b2

            def set(self, state, value, oldvalue, initiator):
                return value + " modified"

        attributes.register_class(Foo)
        attributes.register_class(Bar)
        attributes.register_attribute(Foo, 'data', uselist=False, useobject=False, extension=AlteringListener())
        attributes.register_attribute(Foo, 'barlist', uselist=True, useobject=True, extension=AlteringListener())
        attributes.register_attribute(Foo, 'barset', typecallable=set, uselist=True, useobject=True, extension=AlteringListener())
        attributes.register_attribute(Bar, 'data', uselist=False, useobject=False)
        
        f1 = Foo()
        f1.data = "some data"
        eq_(f1.data, "some data modified")
        b1 = Bar()
        b1.data = "some bar"
        f1.barlist.append(b1)
        assert b1.data == "some bar"
        assert f1.barlist[0].data == "some bar appended"
        
        f1.barset.add(b1)
        assert f1.barset.pop().data == "some bar appended"