diff options
Diffstat (limited to 'test/orm/attributes.py')
| -rw-r--r-- | test/orm/attributes.py | 426 |
1 files changed, 258 insertions, 168 deletions
diff --git a/test/orm/attributes.py b/test/orm/attributes.py index caa129e5e..3883cdcd1 100644 --- a/test/orm/attributes.py +++ b/test/orm/attributes.py @@ -2,18 +2,24 @@ import testenv; testenv.configure_for_tests() import pickle import sqlalchemy.orm.attributes as attributes from sqlalchemy.orm.collections import collection -from sqlalchemy import exceptions +from sqlalchemy.orm.interfaces import AttributeExtension +from sqlalchemy import exc as sa_exc from testlib import * from testlib import fixtures -ROLLBACK_SUPPORTED=False - -# these test classes defined at the module -# level to support pickling -class MyTest(object):pass -class MyTest2(object):pass +# global for pickling tests +MyTest = None +MyTest2 = None class AttributesTest(TestBase): + 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 @@ -29,7 +35,7 @@ class AttributesTest(TestBase): u.email_address = 'lala@123.com' self.assert_(u.user_id == 7 and u.user_name == 'john' and u.email_address == 'lala@123.com') - u._state.commit_all() + 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' @@ -99,31 +105,33 @@ class AttributesTest(TestBase): class Foo(object):pass data = {'a':'this is a', 'b':12} - def loader(instance, keys): + def loader(state, keys): for k in keys: - instance.__dict__[k] = data[k] + state.dict[k] = data[k] return attributes.ATTR_WAS_SET - attributes.register_class(Foo, deferred_scalar_loader=loader) + 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() - f._state.expire_attributes(None) + attributes.instance_state(f).expire_attributes(None) self.assertEquals(f.a, "this is a") self.assertEquals(f.b, 12) f.a = "this is some new a" - f._state.expire_attributes(None) + attributes.instance_state(f).expire_attributes(None) self.assertEquals(f.a, "this is a") self.assertEquals(f.b, 12) - f._state.expire_attributes(None) + attributes.instance_state(f).expire_attributes(None) f.a = "this is another new a" self.assertEquals(f.a, "this is another new a") self.assertEquals(f.b, 12) - f._state.expire_attributes(None) + attributes.instance_state(f).expire_attributes(None) self.assertEquals(f.a, "this is a") self.assertEquals(f.b, 12) @@ -131,23 +139,25 @@ class AttributesTest(TestBase): self.assertEquals(f.a, None) self.assertEquals(f.b, 12) - f._state.commit_all() + attributes.instance_state(f).commit_all() self.assertEquals(f.a, None) self.assertEquals(f.b, 12) def test_deferred_pickleable(self): data = {'a':'this is a', 'b':12} - def loader(instance, keys): + def loader(state, keys): for k in keys: - instance.__dict__[k] = data[k] + state.dict[k] = data[k] return attributes.ATTR_WAS_SET - attributes.register_class(MyTest, deferred_scalar_loader=loader) + 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() - m._state.expire_attributes(None) + 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__ @@ -176,7 +186,7 @@ class AttributesTest(TestBase): 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, a._state.commit_all() + 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' @@ -186,6 +196,45 @@ class AttributesTest(TestBase): 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 + + self.assertEquals(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""" @@ -201,9 +250,9 @@ class AttributesTest(TestBase): # create objects as if they'd been freshly loaded from the database (without history) b = Blog() p1 = Post() - b._state.set_callable('posts', lambda:[p1]) - p1._state.set_callable('blog', lambda:b) - p1, b._state.commit_all() + 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) @@ -253,10 +302,10 @@ class AttributesTest(TestBase): states = set() class Foo(object): def __init__(self): - states.add(self._state) + states.add(attributes.instance_state(self)) class Bar(Foo): def __init__(self): - states.add(self._state) + states.add(attributes.instance_state(self)) Foo.__init__(self) @@ -283,10 +332,10 @@ class AttributesTest(TestBase): el = Element() x = Bar() x.element = el - self.assertEquals(attributes.get_history(x._state, 'element'), ([el],[], [])) - x._state.commit_all() + self.assertEquals(attributes.get_history(attributes.instance_state(x), 'element'), ([el],[], [])) + attributes.instance_state(x).commit_all() - (added, unchanged, deleted) = attributes.get_history(x._state, 'element') + (added, unchanged, deleted) = attributes.get_history(attributes.instance_state(x), 'element') assert added == [] assert unchanged == [el] @@ -312,9 +361,9 @@ class AttributesTest(TestBase): attributes.register_attribute(Bar, 'id', uselist=False, useobject=True) x = Foo() - x._state.commit_all() + attributes.instance_state(x).commit_all() x.col2.append(bar4) - self.assertEquals(attributes.get_history(x._state, 'col2'), ([bar4], [bar1, bar2, bar3], [])) + self.assertEquals(attributes.get_history(attributes.instance_state(x), 'col2'), ([bar4], [bar1, bar2, bar3], [])) def test_parenttrack(self): class Foo(object):pass @@ -358,9 +407,9 @@ class AttributesTest(TestBase): 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'] - x._state.commit_all() + attributes.instance_state(x).commit_all() x.element[1] = 'five' - assert x._state.is_modified() + assert attributes.instance_state(x).check_modified() attributes.unregister_class(Foo) @@ -368,9 +417,9 @@ class AttributesTest(TestBase): attributes.register_attribute(Foo, 'element', uselist=False, useobject=False) x = Foo() x.element = ['one', 'two', 'three'] - x._state.commit_all() + attributes.instance_state(x).commit_all() x.element[1] = 'five' - assert not x._state.is_modified() + assert not attributes.instance_state(x).check_modified() def test_descriptorattributes(self): """changeset: 1633 broke ability to use ORM to map classes with unusual @@ -379,27 +428,31 @@ class AttributesTest(TestBase): This is a simple regression test to prevent that defect. """ class des(object): - def __get__(self, instance, owner): raise AttributeError('fake attribute') + 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 exceptions.ArgumentError, e: + except sa_exc.ArgumentError, e: assert str(e) == "Type InstrumentedDict must elect an appender method to be a collection class" class MyDict(dict): @@ -418,7 +471,7 @@ class AttributesTest(TestBase): try: attributes.register_attribute(Foo, "collection", uselist=True, typecallable=MyColl, useobject=True) assert False - except exceptions.ArgumentError, e: + except sa_exc.ArgumentError, e: assert str(e) == "Type MyColl must elect an appender method to be a collection class" class MyColl(object): @@ -435,7 +488,7 @@ class AttributesTest(TestBase): try: Foo().collection assert True - except exceptions.ArgumentError, e: + except sa_exc.ArgumentError, e: assert False @@ -512,7 +565,7 @@ class BackrefTest(TestBase): j.port = None self.assert_(p.jack is None) -class DeferredBackrefTest(TestBase): +class PendingBackrefTest(TestBase): def setUp(self): global Post, Blog, called, lazy_load @@ -550,6 +603,7 @@ class DeferredBackrefTest(TestBase): b = Blog("blog 1") p = Post("post 4") + p.blog = b p = Post("post 5") p.blog = b @@ -559,6 +613,22 @@ class DeferredBackrefTest(TestBase): # 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 + self.assertEquals(attributes.instance_state(b).get_history('posts'), ([p, p4], [p1, p2, p3], [])) + assert called[0] == 1 def test_lazy_remove(self): global lazy_load @@ -609,17 +679,17 @@ class HistoryTest(TestBase): attributes.register_attribute(Foo, 'someattr', uselist=False, useobject=False) f = Foo() - self.assertEquals(Foo.someattr.impl.get_committed_value(f._state), None) + self.assertEquals(Foo.someattr.impl.get_committed_value(attributes.instance_state(f)), None) f.someattr = 3 - self.assertEquals(Foo.someattr.impl.get_committed_value(f._state), None) + self.assertEquals(Foo.someattr.impl.get_committed_value(attributes.instance_state(f)), None) f = Foo() f.someattr = 3 - self.assertEquals(Foo.someattr.impl.get_committed_value(f._state), None) + self.assertEquals(Foo.someattr.impl.get_committed_value(attributes.instance_state(f)), None) - f._state.commit(['someattr']) - self.assertEquals(Foo.someattr.impl.get_committed_value(f._state), 3) + attributes.instance_state(f).commit(['someattr']) + self.assertEquals(Foo.someattr.impl.get_committed_value(attributes.instance_state(f)), 3) def test_scalar(self): class Foo(fixtures.Base): @@ -630,48 +700,59 @@ class HistoryTest(TestBase): # case 1. new object f = Foo() - self.assertEquals(attributes.get_history(f._state, 'someattr'), ([], [], [])) + self.assertEquals(attributes.get_history(attributes.instance_state(f), 'someattr'), ([], [], [])) f.someattr = "hi" - self.assertEquals(attributes.get_history(f._state, 'someattr'), (['hi'], [], [])) + self.assertEquals(attributes.get_history(attributes.instance_state(f), 'someattr'), (['hi'], [], [])) - f._state.commit(['someattr']) - self.assertEquals(attributes.get_history(f._state, 'someattr'), ([], ['hi'], [])) + attributes.instance_state(f).commit(['someattr']) + self.assertEquals(attributes.get_history(attributes.instance_state(f), 'someattr'), ([], ['hi'], [])) f.someattr = 'there' - self.assertEquals(attributes.get_history(f._state, 'someattr'), (['there'], [], ['hi'])) - f._state.commit(['someattr']) + self.assertEquals(attributes.get_history(attributes.instance_state(f), 'someattr'), (['there'], [], ['hi'])) + attributes.instance_state(f).commit(['someattr']) - self.assertEquals(attributes.get_history(f._state, 'someattr'), ([], ['there'], [])) + self.assertEquals(attributes.get_history(attributes.instance_state(f), 'someattr'), ([], ['there'], [])) del f.someattr - self.assertEquals(attributes.get_history(f._state, 'someattr'), ([], [], ['there'])) + self.assertEquals(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' - self.assertEquals(attributes.get_history(f._state, 'someattr'), ([], ['new'], [])) + self.assertEquals(attributes.get_history(attributes.instance_state(f), 'someattr'), ([], ['new'], [])) f.someattr = 'old' - self.assertEquals(attributes.get_history(f._state, 'someattr'), (['old'], [], ['new'])) + self.assertEquals(attributes.get_history(attributes.instance_state(f), 'someattr'), (['old'], [], ['new'])) - f._state.commit(['someattr']) - self.assertEquals(attributes.get_history(f._state, 'someattr'), ([], ['old'], [])) + attributes.instance_state(f).commit(['someattr']) + self.assertEquals(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() - self.assertEquals(attributes.get_history(f._state, 'someattr'), ([], [], [])) + self.assertEquals(attributes.get_history(attributes.instance_state(f), 'someattr'), ([], [], [])) + print f._foostate.committed_state f.someattr = None - self.assertEquals(attributes.get_history(f._state, 'someattr'), ([None], [], [])) + print f._foostate.committed_state, f._foostate.dict + self.assertEquals(attributes.get_history(attributes.instance_state(f), 'someattr'), ([None], [], [])) f = Foo() f.__dict__['someattr'] = 'new' - self.assertEquals(attributes.get_history(f._state, 'someattr'), ([], ['new'], [])) + self.assertEquals(attributes.get_history(attributes.instance_state(f), 'someattr'), ([], ['new'], [])) f.someattr = None - self.assertEquals(attributes.get_history(f._state, 'someattr'), ([None], [], ['new'])) + self.assertEquals(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' + self.assertEquals(attributes.get_history(attributes.instance_state(f), 'someattr'), (['one'], [], [])) + f.someattr = 'two' + self.assertEquals(attributes.get_history(attributes.instance_state(f), 'someattr'), (['two'], [], [])) + + def test_mutable_scalar(self): class Foo(fixtures.Base): pass @@ -681,33 +762,33 @@ class HistoryTest(TestBase): # case 1. new object f = Foo() - self.assertEquals(attributes.get_history(f._state, 'someattr'), ([], [], [])) + self.assertEquals(attributes.get_history(attributes.instance_state(f), 'someattr'), ([], [], [])) f.someattr = {'foo':'hi'} - self.assertEquals(attributes.get_history(f._state, 'someattr'), ([{'foo':'hi'}], [], [])) + self.assertEquals(attributes.get_history(attributes.instance_state(f), 'someattr'), ([{'foo':'hi'}], [], [])) - f._state.commit(['someattr']) - self.assertEquals(attributes.get_history(f._state, 'someattr'), ([], [{'foo':'hi'}], [])) - self.assertEquals(f._state.committed_state['someattr'], {'foo':'hi'}) + attributes.instance_state(f).commit(['someattr']) + self.assertEquals(attributes.get_history(attributes.instance_state(f), 'someattr'), ([], [{'foo':'hi'}], [])) + self.assertEquals(attributes.instance_state(f).committed_state['someattr'], {'foo':'hi'}) f.someattr['foo'] = 'there' - self.assertEquals(f._state.committed_state['someattr'], {'foo':'hi'}) + self.assertEquals(attributes.instance_state(f).committed_state['someattr'], {'foo':'hi'}) - self.assertEquals(attributes.get_history(f._state, 'someattr'), ([{'foo':'there'}], [], [{'foo':'hi'}])) - f._state.commit(['someattr']) + self.assertEquals(attributes.get_history(attributes.instance_state(f), 'someattr'), ([{'foo':'there'}], [], [{'foo':'hi'}])) + attributes.instance_state(f).commit(['someattr']) - self.assertEquals(attributes.get_history(f._state, 'someattr'), ([], [{'foo':'there'}], [])) + self.assertEquals(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'} - self.assertEquals(attributes.get_history(f._state, 'someattr'), ([], [{'foo':'new'}], [])) + self.assertEquals(attributes.get_history(attributes.instance_state(f), 'someattr'), ([], [{'foo':'new'}], [])) f.someattr = {'foo':'old'} - self.assertEquals(attributes.get_history(f._state, 'someattr'), ([{'foo':'old'}], [], [{'foo':'new'}])) + self.assertEquals(attributes.get_history(attributes.instance_state(f), 'someattr'), ([{'foo':'old'}], [], [{'foo':'new'}])) - f._state.commit(['someattr']) - self.assertEquals(attributes.get_history(f._state, 'someattr'), ([], [{'foo':'old'}], [])) + attributes.instance_state(f).commit(['someattr']) + self.assertEquals(attributes.get_history(attributes.instance_state(f), 'someattr'), ([], [{'foo':'old'}], [])) def test_use_object(self): @@ -729,48 +810,56 @@ class HistoryTest(TestBase): # case 1. new object f = Foo() - self.assertEquals(attributes.get_history(f._state, 'someattr'), ([], [None], [])) + self.assertEquals(attributes.get_history(attributes.instance_state(f), 'someattr'), ([], [None], [])) f.someattr = hi - self.assertEquals(attributes.get_history(f._state, 'someattr'), ([hi], [], [])) + self.assertEquals(attributes.get_history(attributes.instance_state(f), 'someattr'), ([hi], [], [])) - f._state.commit(['someattr']) - self.assertEquals(attributes.get_history(f._state, 'someattr'), ([], [hi], [])) + attributes.instance_state(f).commit(['someattr']) + self.assertEquals(attributes.get_history(attributes.instance_state(f), 'someattr'), ([], [hi], [])) f.someattr = there - self.assertEquals(attributes.get_history(f._state, 'someattr'), ([there], [], [hi])) - f._state.commit(['someattr']) + self.assertEquals(attributes.get_history(attributes.instance_state(f), 'someattr'), ([there], [], [hi])) + attributes.instance_state(f).commit(['someattr']) - self.assertEquals(attributes.get_history(f._state, 'someattr'), ([], [there], [])) + self.assertEquals(attributes.get_history(attributes.instance_state(f), 'someattr'), ([], [there], [])) del f.someattr - self.assertEquals(attributes.get_history(f._state, 'someattr'), ([None], [], [there])) + self.assertEquals(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 - self.assertEquals(attributes.get_history(f._state, 'someattr'), ([], [new], [])) + f.__dict__['someattr'] = 'new' + self.assertEquals(attributes.get_history(attributes.instance_state(f), 'someattr'), ([], ['new'], [])) f.someattr = old - self.assertEquals(attributes.get_history(f._state, 'someattr'), ([old], [], [new])) + self.assertEquals(attributes.get_history(attributes.instance_state(f), 'someattr'), ([old], [], ['new'])) - f._state.commit(['someattr']) - self.assertEquals(attributes.get_history(f._state, 'someattr'), ([], [old], [])) + attributes.instance_state(f).commit(['someattr']) + self.assertEquals(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() - self.assertEquals(attributes.get_history(f._state, 'someattr'), ([], [None], [])) + self.assertEquals(attributes.get_history(attributes.instance_state(f), 'someattr'), ([], [None], [])) f.someattr = None - self.assertEquals(attributes.get_history(f._state, 'someattr'), ([], [None], [])) + self.assertEquals(attributes.get_history(attributes.instance_state(f), 'someattr'), ([], [None], [])) f = Foo() - f.__dict__['someattr'] = new - self.assertEquals(attributes.get_history(f._state, 'someattr'), ([], [new], [])) + f.__dict__['someattr'] = 'new' + self.assertEquals(attributes.get_history(attributes.instance_state(f), 'someattr'), ([], ['new'], [])) f.someattr = None - self.assertEquals(attributes.get_history(f._state, 'someattr'), ([None], [], [new])) + self.assertEquals(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' + self.assertEquals(attributes.get_history(attributes.instance_state(f), 'someattr'), (['one'], [], [])) + f.someattr = 'two' + self.assertEquals(attributes.get_history(attributes.instance_state(f), 'someattr'), (['two'], [], [])) def test_object_collections_set(self): class Foo(fixtures.Base): @@ -789,39 +878,39 @@ class HistoryTest(TestBase): # case 1. new object f = Foo() - self.assertEquals(attributes.get_history(f._state, 'someattr'), ([], [], [])) + self.assertEquals(attributes.get_history(attributes.instance_state(f), 'someattr'), ([], [], [])) f.someattr = [hi] - self.assertEquals(attributes.get_history(f._state, 'someattr'), ([hi], [], [])) + self.assertEquals(attributes.get_history(attributes.instance_state(f), 'someattr'), ([hi], [], [])) - f._state.commit(['someattr']) - self.assertEquals(attributes.get_history(f._state, 'someattr'), ([], [hi], [])) + attributes.instance_state(f).commit(['someattr']) + self.assertEquals(attributes.get_history(attributes.instance_state(f), 'someattr'), ([], [hi], [])) f.someattr = [there] - self.assertEquals(attributes.get_history(f._state, 'someattr'), ([there], [], [hi])) - f._state.commit(['someattr']) + self.assertEquals(attributes.get_history(attributes.instance_state(f), 'someattr'), ([there], [], [hi])) + attributes.instance_state(f).commit(['someattr']) - self.assertEquals(attributes.get_history(f._state, 'someattr'), ([], [there], [])) + self.assertEquals(attributes.get_history(attributes.instance_state(f), 'someattr'), ([], [there], [])) f.someattr = [hi] - self.assertEquals(attributes.get_history(f._state, 'someattr'), ([hi], [], [there])) + self.assertEquals(attributes.get_history(attributes.instance_state(f), 'someattr'), ([hi], [], [there])) f.someattr = [old, new] - self.assertEquals(attributes.get_history(f._state, 'someattr'), ([old, new], [], [there])) + self.assertEquals(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(f, 'someattr') + collection = attributes.init_collection(attributes.instance_state(f), 'someattr') collection.append_without_event(new) - f._state.commit_all() - self.assertEquals(attributes.get_history(f._state, 'someattr'), ([], [new], [])) + attributes.instance_state(f).commit_all() + self.assertEquals(attributes.get_history(attributes.instance_state(f), 'someattr'), ([], [new], [])) f.someattr = [old] - self.assertEquals(attributes.get_history(f._state, 'someattr'), ([old], [], [new])) + self.assertEquals(attributes.get_history(attributes.instance_state(f), 'someattr'), ([old], [], [new])) - f._state.commit(['someattr']) - self.assertEquals(attributes.get_history(f._state, 'someattr'), ([], [old], [])) + attributes.instance_state(f).commit(['someattr']) + self.assertEquals(attributes.get_history(attributes.instance_state(f), 'someattr'), ([], [old], [])) def test_dict_collections(self): class Foo(fixtures.Base): @@ -840,16 +929,16 @@ class HistoryTest(TestBase): new = Bar(name='new') f = Foo() - self.assertEquals(attributes.get_history(f._state, 'someattr'), ([], [], [])) + self.assertEquals(attributes.get_history(attributes.instance_state(f), 'someattr'), ([], [], [])) f.someattr['hi'] = hi - self.assertEquals(attributes.get_history(f._state, 'someattr'), ([hi], [], [])) + self.assertEquals(attributes.get_history(attributes.instance_state(f), 'someattr'), ([hi], [], [])) f.someattr['there'] = there - self.assertEquals(tuple([set(x) for x in attributes.get_history(f._state, 'someattr')]), (set([hi, there]), set([]), set([]))) + self.assertEquals(tuple([set(x) for x in attributes.get_history(attributes.instance_state(f), 'someattr')]), (set([hi, there]), set([]), set([]))) - f._state.commit(['someattr']) - self.assertEquals(tuple([set(x) for x in attributes.get_history(f._state, 'someattr')]), (set([]), set([hi, there]), set([]))) + attributes.instance_state(f).commit(['someattr']) + self.assertEquals(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(fixtures.Base): @@ -868,65 +957,65 @@ class HistoryTest(TestBase): # case 1. new object f = Foo(id=1) - self.assertEquals(attributes.get_history(f._state, 'someattr'), ([], [], [])) + self.assertEquals(attributes.get_history(attributes.instance_state(f), 'someattr'), ([], [], [])) f.someattr.append(hi) - self.assertEquals(attributes.get_history(f._state, 'someattr'), ([hi], [], [])) + self.assertEquals(attributes.get_history(attributes.instance_state(f), 'someattr'), ([hi], [], [])) - f._state.commit(['someattr']) - self.assertEquals(attributes.get_history(f._state, 'someattr'), ([], [hi], [])) + attributes.instance_state(f).commit(['someattr']) + self.assertEquals(attributes.get_history(attributes.instance_state(f), 'someattr'), ([], [hi], [])) f.someattr.append(there) - self.assertEquals(attributes.get_history(f._state, 'someattr'), ([there], [hi], [])) - f._state.commit(['someattr']) + self.assertEquals(attributes.get_history(attributes.instance_state(f), 'someattr'), ([there], [hi], [])) + attributes.instance_state(f).commit(['someattr']) - self.assertEquals(attributes.get_history(f._state, 'someattr'), ([], [hi, there], [])) + self.assertEquals(attributes.get_history(attributes.instance_state(f), 'someattr'), ([], [hi, there], [])) f.someattr.remove(there) - self.assertEquals(attributes.get_history(f._state, 'someattr'), ([], [hi], [there])) + self.assertEquals(attributes.get_history(attributes.instance_state(f), 'someattr'), ([], [hi], [there])) f.someattr.append(old) f.someattr.append(new) - self.assertEquals(attributes.get_history(f._state, 'someattr'), ([old, new], [hi], [there])) - f._state.commit(['someattr']) - self.assertEquals(attributes.get_history(f._state, 'someattr'), ([], [hi, old, new], [])) + self.assertEquals(attributes.get_history(attributes.instance_state(f), 'someattr'), ([old, new], [hi], [there])) + attributes.instance_state(f).commit(['someattr']) + self.assertEquals(attributes.get_history(attributes.instance_state(f), 'someattr'), ([], [hi, old, new], [])) f.someattr.pop(0) - self.assertEquals(attributes.get_history(f._state, 'someattr'), ([], [old, new], [hi])) + self.assertEquals(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(f, 'someattr') + collection = attributes.init_collection(attributes.instance_state(f), 'someattr') collection.append_without_event(new) - f._state.commit_all() - self.assertEquals(attributes.get_history(f._state, 'someattr'), ([], [new], [])) + attributes.instance_state(f).commit_all() + self.assertEquals(attributes.get_history(attributes.instance_state(f), 'someattr'), ([], [new], [])) f.someattr.append(old) - self.assertEquals(attributes.get_history(f._state, 'someattr'), ([old], [new], [])) + self.assertEquals(attributes.get_history(attributes.instance_state(f), 'someattr'), ([old], [new], [])) - f._state.commit(['someattr']) - self.assertEquals(attributes.get_history(f._state, 'someattr'), ([], [new, old], [])) + attributes.instance_state(f).commit(['someattr']) + self.assertEquals(attributes.get_history(attributes.instance_state(f), 'someattr'), ([], [new, old], [])) f = Foo() - collection = attributes.init_collection(f, 'someattr') + collection = attributes.init_collection(attributes.instance_state(f), 'someattr') collection.append_without_event(new) - f._state.commit_all() - self.assertEquals(attributes.get_history(f._state, 'someattr'), ([], [new], [])) + attributes.instance_state(f).commit_all() + self.assertEquals(attributes.get_history(attributes.instance_state(f), 'someattr'), ([], [new], [])) f.id = 1 f.someattr.remove(new) - self.assertEquals(attributes.get_history(f._state, 'someattr'), ([], [], [new])) + self.assertEquals(attributes.get_history(attributes.instance_state(f), 'someattr'), ([], [], [new])) # case 3. mixing appends with sets f = Foo() f.someattr.append(hi) - self.assertEquals(attributes.get_history(f._state, 'someattr'), ([hi], [], [])) + self.assertEquals(attributes.get_history(attributes.instance_state(f), 'someattr'), ([hi], [], [])) f.someattr.append(there) - self.assertEquals(attributes.get_history(f._state, 'someattr'), ([hi, there], [], [])) + self.assertEquals(attributes.get_history(attributes.instance_state(f), 'someattr'), ([hi, there], [], [])) f.someattr = [there] - self.assertEquals(attributes.get_history(f._state, 'someattr'), ([there], [], [])) + self.assertEquals(attributes.get_history(attributes.instance_state(f), 'someattr'), ([there], [], [])) def test_collections_via_backref(self): class Foo(fixtures.Base): @@ -941,19 +1030,19 @@ class HistoryTest(TestBase): f1 = Foo() b1 = Bar() - self.assertEquals(attributes.get_history(f1._state, 'bars'), ([], [], [])) - self.assertEquals(attributes.get_history(b1._state, 'foo'), ([], [None], [])) + self.assertEquals(attributes.get_history(attributes.instance_state(f1), 'bars'), ([], [], [])) + self.assertEquals(attributes.get_history(attributes.instance_state(b1), 'foo'), ([], [None], [])) #b1.foo = f1 f1.bars.append(b1) - self.assertEquals(attributes.get_history(f1._state, 'bars'), ([b1], [], [])) - self.assertEquals(attributes.get_history(b1._state, 'foo'), ([f1], [], [])) + self.assertEquals(attributes.get_history(attributes.instance_state(f1), 'bars'), ([b1], [], [])) + self.assertEquals(attributes.get_history(attributes.instance_state(b1), 'foo'), ([f1], [], [])) b2 = Bar() f1.bars.append(b2) - self.assertEquals(attributes.get_history(f1._state, 'bars'), ([b1, b2], [], [])) - self.assertEquals(attributes.get_history(b1._state, 'foo'), ([f1], [], [])) - self.assertEquals(attributes.get_history(b2._state, 'foo'), ([f1], [], [])) + self.assertEquals(attributes.get_history(attributes.instance_state(f1), 'bars'), ([b1, b2], [], [])) + self.assertEquals(attributes.get_history(attributes.instance_state(b1), 'foo'), ([f1], [], [])) + self.assertEquals(attributes.get_history(attributes.instance_state(b2), 'foo'), ([f1], [], [])) def test_lazy_backref_collections(self): class Foo(fixtures.Base): @@ -978,17 +1067,17 @@ class HistoryTest(TestBase): f = Foo() bar4 = Bar() bar4.foo = f - self.assertEquals(attributes.get_history(f._state, 'bars'), ([bar4], [bar1, bar2, bar3], [])) + self.assertEquals(attributes.get_history(attributes.instance_state(f), 'bars'), ([bar4], [bar1, bar2, bar3], [])) lazy_load = None f = Foo() bar4 = Bar() bar4.foo = f - self.assertEquals(attributes.get_history(f._state, 'bars'), ([bar4], [], [])) + self.assertEquals(attributes.get_history(attributes.instance_state(f), 'bars'), ([bar4], [], [])) lazy_load = [bar1, bar2, bar3] - f._state.expire_attributes(['bars']) - self.assertEquals(attributes.get_history(f._state, 'bars'), ([], [bar1, bar2, bar3], [])) + attributes.instance_state(f).expire_attributes(['bars']) + self.assertEquals(attributes.get_history(attributes.instance_state(f), 'bars'), ([], [bar1, bar2, bar3], [])) def test_collections_via_lazyload(self): class Foo(fixtures.Base): @@ -1011,26 +1100,26 @@ class HistoryTest(TestBase): f = Foo() f.bars = [] - self.assertEquals(attributes.get_history(f._state, 'bars'), ([], [], [bar1, bar2, bar3])) + self.assertEquals(attributes.get_history(attributes.instance_state(f), 'bars'), ([], [], [bar1, bar2, bar3])) f = Foo() f.bars.append(bar4) - self.assertEquals(attributes.get_history(f._state, 'bars'), ([bar4], [bar1, bar2, bar3], []) ) + self.assertEquals(attributes.get_history(attributes.instance_state(f), 'bars'), ([bar4], [bar1, bar2, bar3], []) ) f = Foo() f.bars.remove(bar2) - self.assertEquals(attributes.get_history(f._state, 'bars'), ([], [bar1, bar3], [bar2])) + self.assertEquals(attributes.get_history(attributes.instance_state(f), 'bars'), ([], [bar1, bar3], [bar2])) f.bars.append(bar4) - self.assertEquals(attributes.get_history(f._state, 'bars'), ([bar4], [bar1, bar3], [bar2])) + self.assertEquals(attributes.get_history(attributes.instance_state(f), 'bars'), ([bar4], [bar1, bar3], [bar2])) f = Foo() del f.bars[1] - self.assertEquals(attributes.get_history(f._state, 'bars'), ([], [bar1, bar3], [bar2])) + self.assertEquals(attributes.get_history(attributes.instance_state(f), 'bars'), ([], [bar1, bar3], [bar2])) lazy_load = None f = Foo() f.bars.append(bar2) - self.assertEquals(attributes.get_history(f._state, 'bars'), ([bar2], [], [])) + self.assertEquals(attributes.get_history(attributes.instance_state(f), 'bars'), ([bar2], [], [])) def test_scalar_via_lazyload(self): class Foo(fixtures.Base): @@ -1051,24 +1140,24 @@ class HistoryTest(TestBase): f = Foo() self.assertEquals(f.bar, "hi") - self.assertEquals(attributes.get_history(f._state, 'bar'), ([], ["hi"], [])) + self.assertEquals(attributes.get_history(attributes.instance_state(f), 'bar'), ([], ["hi"], [])) f = Foo() f.bar = None - self.assertEquals(attributes.get_history(f._state, 'bar'), ([None], [], [])) + self.assertEquals(attributes.get_history(attributes.instance_state(f), 'bar'), ([None], [], [])) f = Foo() f.bar = "there" - self.assertEquals(attributes.get_history(f._state, 'bar'), (["there"], [], [])) + self.assertEquals(attributes.get_history(attributes.instance_state(f), 'bar'), (["there"], [], [])) f.bar = "hi" - self.assertEquals(attributes.get_history(f._state, 'bar'), (["hi"], [], [])) + self.assertEquals(attributes.get_history(attributes.instance_state(f), 'bar'), (["hi"], [], [])) f = Foo() self.assertEquals(f.bar, "hi") del f.bar - self.assertEquals(attributes.get_history(f._state, 'bar'), ([], [], ["hi"])) + self.assertEquals(attributes.get_history(attributes.instance_state(f), 'bar'), ([], [], ["hi"])) assert f.bar is None - self.assertEquals(attributes.get_history(f._state, 'bar'), ([None], [], ["hi"])) + self.assertEquals(attributes.get_history(attributes.instance_state(f), 'bar'), ([None], [], ["hi"])) def test_scalar_object_via_lazyload(self): class Foo(fixtures.Base): @@ -1092,24 +1181,25 @@ class HistoryTest(TestBase): # operations f = Foo() - self.assertEquals(attributes.get_history(f._state, 'bar'), ([], [bar1], [])) + self.assertEquals(attributes.get_history(attributes.instance_state(f), 'bar'), ([], [bar1], [])) f = Foo() f.bar = None - self.assertEquals(attributes.get_history(f._state, 'bar'), ([None], [], [bar1])) + self.assertEquals(attributes.get_history(attributes.instance_state(f), 'bar'), ([None], [], [bar1])) f = Foo() f.bar = bar2 - self.assertEquals(attributes.get_history(f._state, 'bar'), ([bar2], [], [bar1])) + self.assertEquals(attributes.get_history(attributes.instance_state(f), 'bar'), ([bar2], [], [bar1])) f.bar = bar1 - self.assertEquals(attributes.get_history(f._state, 'bar'), ([], [bar1], [])) + self.assertEquals(attributes.get_history(attributes.instance_state(f), 'bar'), ([], [bar1], [])) f = Foo() self.assertEquals(f.bar, bar1) del f.bar - self.assertEquals(attributes.get_history(f._state, 'bar'), ([None], [], [bar1])) + self.assertEquals(attributes.get_history(attributes.instance_state(f), 'bar'), ([None], [], [bar1])) assert f.bar is None - self.assertEquals(attributes.get_history(f._state, 'bar'), ([None], [], [bar1])) + self.assertEquals(attributes.get_history(attributes.instance_state(f), 'bar'), ([None], [], [bar1])) + if __name__ == "__main__": testenv.main() |