path: root/test/orm/attributes.py

import testenv; testenv.configure_for_tests()
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 testlib import *
from testlib.testing import eq_
from orm import _base

# 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()
        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(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):
            return None
        attr_name = 'mt2'
        attributes.register_attribute(MyTest, attr_name, uselist = True, trackparent=True, callable_=somecallable, useobject=True)

        o = MyTest()
        o.mt2.append(MyTest2())
        o.user_id=7
        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

            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)

        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_(len(o4.mt2) == 1)
        self.assert_(o4.mt2[0].a == 'abcde')
        self.assert_(o4.mt2[0].b is None)

    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(None)
        eq_(f.a, "this is a")
        eq_(f.b, 12)

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

        attributes.instance_state(f).expire_attributes(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(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()
        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(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()
        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_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))
        
        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('posts', lambda:[p1])
        attributes.instance_state(p1).set_callable('blog', lambda:b)
        p1, attributes.instance_state(b).commit_all()

        # 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():
            print "func1"
            return "this is the foo attr"
        def func2():
            print "func2"
            return "this is the bar attr"
        def func3():
            print "func3"
            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_history(attributes.instance_state(x), 'element'), ([el],[], []))
        attributes.instance_state(x).commit_all()

        (added, unchanged, deleted) = attributes.get_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():
            return "this is func 1"
        def func2():
            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()
        x.col2.append(bar4)
        eq_(attributes.get_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()
        x.element[1] = 'five'
        assert attributes.instance_state(x).check_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()
        x.element[1] = 'five'
        assert not attributes.instance_state(x).check_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 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
            def __eq__(self, other):
                return other.name == self.name

        class Blog(object):
            def __init__(self, name):
                self.name = name
            def __eq__(self, other):
                return other.name == self.name

        called = [0]

        lazy_load = []
        def lazy_posts(instance):
            def load():
                called[0] += 1
                return lazy_load
            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_state(p1).commit_all()
        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)), None)

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

        f = Foo()
        f.someattr = 3
        eq_(Foo.someattr.impl.get_committed_value(attributes.instance_state(f)), None)
        
        attributes.instance_state(f).commit(['someattr'])
        eq_(Foo.someattr.impl.get_committed_value(attributes.instance_state(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_history(attributes.instance_state(f), 'someattr'), ([], [], []))

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

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

        f.someattr = 'there'

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

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

        del f.someattr
        eq_(attributes.get_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_history(attributes.instance_state(f), 'someattr'), ([], ['new'], []))

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

        attributes.instance_state(f).commit(['someattr'])
        eq_(attributes.get_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_history(attributes.instance_state(f), 'someattr'), ([], [], []))
        f.someattr = None
        eq_(attributes.get_history(attributes.instance_state(f), 'someattr'), ([None], [], []))

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

        # set same value twice
        f = Foo()
        attributes.instance_state(f).commit(['someattr'])
        f.someattr = 'one'
        eq_(attributes.get_history(attributes.instance_state(f), 'someattr'), (['one'], [], []))
        f.someattr = 'two'
        eq_(attributes.get_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_history(attributes.instance_state(f), 'someattr'), ([], [], []))

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

        attributes.instance_state(f).commit(['someattr'])
        eq_(attributes.get_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_history(attributes.instance_state(f), 'someattr'), ([{'foo':'there'}], [], [{'foo':'hi'}]))
        attributes.instance_state(f).commit(['someattr'])

        eq_(attributes.get_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_history(attributes.instance_state(f), 'someattr'), ([], [{'foo':'new'}], []))

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

        attributes.instance_state(f).commit(['someattr'])
        eq_(attributes.get_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_history(attributes.instance_state(f), 'someattr'), ([], [None], []))

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

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

        f.someattr = there

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

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

        del f.someattr
        eq_(attributes.get_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_history(attributes.instance_state(f), 'someattr'), ([], ['new'], []))

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

        attributes.instance_state(f).commit(['someattr'])
        eq_(attributes.get_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_history(attributes.instance_state(f), 'someattr'), ([], [None], []))
        f.someattr = None
        eq_(attributes.get_history(attributes.instance_state(f), 'someattr'), ([], [None], []))

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

        # set same value twice
        f = Foo()
        attributes.instance_state(f).commit(['someattr'])
        f.someattr = 'one'
        eq_(attributes.get_history(attributes.instance_state(f), 'someattr'), (['one'], [], []))
        f.someattr = 'two'
        eq_(attributes.get_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_history(attributes.instance_state(f), 'someattr'), ([], [], []))

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

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

        f.someattr = [there]

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

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

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

        f.someattr = [old, new]
        eq_(attributes.get_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(attributes.instance_state(f), 'someattr')
        collection.append_without_event(new)
        attributes.instance_state(f).commit_all()
        eq_(attributes.get_history(attributes.instance_state(f), 'someattr'), ([], [new], []))

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

        attributes.instance_state(f).commit(['someattr'])
        eq_(attributes.get_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_history(attributes.instance_state(f), 'someattr'), ([], [], []))

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

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

        attributes.instance_state(f).commit(['someattr'])
        eq_(tuple([set(x) for x in attributes.get_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_history(attributes.instance_state(f), 'someattr'), ([], [], []))

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

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

        f.someattr.append(there)

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

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

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

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

        f.someattr.pop(0)
        eq_(attributes.get_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(attributes.instance_state(f), 'someattr')
        collection.append_without_event(new)
        attributes.instance_state(f).commit_all()
        eq_(attributes.get_history(attributes.instance_state(f), 'someattr'), ([], [new], []))

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

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

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

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

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

    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_history(attributes.instance_state(f1), 'bars'), ([], [], []))
        eq_(attributes.get_history(attributes.instance_state(b1), 'foo'), ([], [None], []))

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

        b2 = Bar()
        f1.bars.append(b2)
        eq_(attributes.get_history(attributes.instance_state(f1), 'bars'), ([b1, b2], [], []))
        eq_(attributes.get_history(attributes.instance_state(b1), 'foo'), ([f1], [], []))
        eq_(attributes.get_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():
                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_history(attributes.instance_state(f), 'bars'), ([bar4], [bar1, bar2, bar3], []))

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

        lazy_load = [bar1, bar2, bar3]
        attributes.instance_state(f).expire_attributes(['bars'])
        eq_(attributes.get_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():
                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_history(attributes.instance_state(f), 'bars'), ([], [], [bar1, bar2, bar3]))

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

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

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

        lazy_load = None
        f = Foo()
        f.bars.append(bar2)
        eq_(attributes.get_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():
                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, the lazy callable is only executed on gets, not history
        # operations

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

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

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

        f = Foo()
        eq_(f.bar, "hi")
        del f.bar
        eq_(attributes.get_history(attributes.instance_state(f), 'bar'), ([], [], ["hi"]))
        assert f.bar is None
        eq_(attributes.get_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():
                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_history(attributes.instance_state(f), 'bar'), ([], [bar1], []))

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

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

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

    
if __name__ == "__main__":
    testenv.main()