diff options
Diffstat (limited to 'test/orm/test_attributes.py')
| -rw-r--r-- | test/orm/test_attributes.py | 201 |
1 files changed, 0 insertions, 201 deletions
diff --git a/test/orm/test_attributes.py b/test/orm/test_attributes.py index 2690c7442..d99fcc77b 100644 --- a/test/orm/test_attributes.py +++ b/test/orm/test_attributes.py @@ -7,7 +7,6 @@ from sqlalchemy.orm import attributes from sqlalchemy.orm import exc as orm_exc from sqlalchemy.orm import instrumentation from sqlalchemy.orm.collections import collection -from sqlalchemy.orm.interfaces import AttributeExtension from sqlalchemy.orm.state import InstanceState from sqlalchemy.testing import assert_raises from sqlalchemy.testing import assert_raises_message @@ -574,206 +573,6 @@ class AttributesTest(fixtures.ORMTest): eq_(u.addresses[0].email_address, "lala@123.com") eq_(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(fixtures.BasicEntity): - pass - - class Bar(fixtures.BasicEntity): - 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 - - instrumentation.register_class(Foo) - instrumentation.register_class(Bar) - - b1, b2, b3, b4 = Bar(id="b1"), Bar(id="b2"), Bar(id="b3"), Bar(id="b4") - - def loadcollection(state, passive): - if passive is attributes.PASSIVE_NO_FETCH: - return attributes.PASSIVE_NO_RESULT - return [b1, b2] - - def loadscalar(state, passive): - if passive is attributes.PASSIVE_NO_FETCH: - return attributes.PASSIVE_NO_RESULT - return b2 - - attributes.register_attribute( - Foo, - "bars", - uselist=True, - useobject=True, - callable_=loadcollection, - extension=[ReceiveEvents("bars")], - ) - - attributes.register_attribute( - Foo, - "bar", - uselist=False, - useobject=True, - callable_=loadscalar, - extension=[ReceiveEvents("bar")], - ) - - attributes.register_attribute( - Foo, - "scalar", - uselist=False, - useobject=False, - extension=[ReceiveEvents("scalar")], - ) - - def create_hist(): - def hist(key, fn, *arg): - attributes.instance_state(f1)._commit_all( - attributes.instance_dict(f1) - ) - fn(*arg) - histories.append(attributes.get_history(f1, key)) - - f1 = Foo() - hist("bars", f1.bars.append, b3) - hist("bars", f1.bars.append, b4) - hist("bars", f1.bars.remove, b2) - hist("bar", setattr, f1, "bar", b3) - hist("bar", setattr, f1, "bar", None) - hist("bar", setattr, f1, "bar", b4) - hist("scalar", setattr, f1, "scalar", 5) - hist("scalar", setattr, f1, "scalar", None) - hist("scalar", 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): - woc = without - wic = with_ - eq_(woc, wic) - - def test_extension_lazyload_assertion(self): - class Foo(fixtures.BasicEntity): - pass - - class Bar(fixtures.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 - - instrumentation.register_class(Foo) - instrumentation.register_class(Bar) - - bar1, bar2, bar3 = [Bar(id=1), Bar(id=2), Bar(id=3)] - - def func1(state, passive): - if passive is attributes.PASSIVE_NO_FETCH: - return attributes.PASSIVE_NO_RESULT - - return [bar1, bar2, bar3] - - attributes.register_attribute( - Foo, - "bars", - uselist=True, - callable_=func1, - useobject=True, - extension=[ReceiveEvents()], - ) - attributes.register_attribute( - Bar, "foos", uselist=True, useobject=True, backref="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 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 - - instrumentation.register_class(Foo) - attributes.register_attribute( - Foo, "x", uselist=False, useobject=False, extension=ReceiveEvents() - ) - - f = Foo() - f.x = 5 - f.x = 17 - del f.x - - eq_( - results, - [ - ("set", f, 5, attributes.NEVER_SET), - ("set", f, 17, 5), - ("remove", f, 17), - ], - ) - def test_lazytrackparent(self): """test that the "hasparent" flag works properly when lazy loaders and backrefs are used |