"""
a series of tests which assert the behavior of moving objects between collections
and scalar attributes resulting in the expected state w.r.t. backrefs, add/remove
events, etc.
there's a particular focus on collections that have "uselist=False", since in these
cases the re-assignment of an attribute means the previous owner needs an
UPDATE in the database.
"""
from sqlalchemy.test.testing import assert_raises,assert_raises_message
from sqlalchemy import Integer,String,ForeignKey,Sequence,exc
from sqlalchemy.test.schema import Table
from sqlalchemy.test.schema import Column
from sqlalchemy.orm import mapper,relationship,create_session,class_mapper,backref,sessionmaker
from sqlalchemy.orm import attributes,exc
from sqlalchemy.test import testing
from sqlalchemy.test.testing import eq_
from test.orm import _base,_fixtures
class O2MCollectionTest(_fixtures.FixtureTest):
run_inserts = None
@classmethod
@testing.resolve_artifact_names
def setup_mappers(cls):
mapper(Address, addresses)
mapper(User, users, properties = dict(
addresses = relationship(Address, backref="user"),
))
@testing.resolve_artifact_names
def test_collection_move_hitslazy(self):
sess = sessionmaker()()
a1 = Address(email_address="address1")
a2 = Address(email_address="address2")
a3 = Address(email_address="address3")
u1= User(name='jack', addresses=[a1, a2, a3])
u2= User(name='ed')
sess.add_all([u1, a1, a2, a3])
sess.commit()
#u1.addresses
def go():
u2.addresses.append(a1)
u2.addresses.append(a2)
u2.addresses.append(a3)
self.assert_sql_count(testing.db, go, 0)
@testing.resolve_artifact_names
def test_collection_move_preloaded(self):
sess = sessionmaker()()
a1 = Address(email_address="address1")
u1 = User(name='jack', addresses=[a1])
u2 = User(name='ed')
sess.add_all([u1, u2])
sess.commit() # everything is expired
# load u1.addresses collection
u1.addresses
u2.addresses.append(a1)
# backref fires
assert a1.user is u2
# doesn't extend to the previous collection tho,
# which was already loaded.
# flushing at this point means its anyone's guess.
assert a1 in u1.addresses
assert a1 in u2.addresses
@testing.resolve_artifact_names
def test_collection_move_notloaded(self):
sess = sessionmaker()()
a1 = Address(email_address="address1")
u1 = User(name='jack', addresses=[a1])
u2 = User(name='ed')
sess.add_all([u1, u2])
sess.commit() # everything is expired
u2.addresses.append(a1)
# backref fires
assert a1.user is u2
# u1.addresses wasn't loaded,
# so when it loads its correct
assert a1 not in u1.addresses
assert a1 in u2.addresses
@testing.resolve_artifact_names
def test_collection_move_commitfirst(self):
sess = sessionmaker()()
a1 = Address(email_address="address1")
u1 = User(name='jack', addresses=[a1])
u2 = User(name='ed')
sess.add_all([u1, u2])
sess.commit() # everything is expired
# load u1.addresses collection
u1.addresses
u2.addresses.append(a1)
# backref fires
assert a1.user is u2
# everything expires, no changes in
# u1.addresses, so all is fine
sess.commit()
assert a1 not in u1.addresses
assert a1 in u2.addresses
@testing.resolve_artifact_names
def test_scalar_move_preloaded(self):
sess = sessionmaker()()
u1 = User(name='jack')
u2 = User(name='ed')
a1 = Address(email_address='a1')
a1.user = u1
sess.add_all([u1, u2, a1])
sess.commit()
# u1.addresses is loaded
u1.addresses
# direct set - the "old" is "fetched",
# but only from the local session - not the
# database, due to the PASSIVE_NO_FETCH flag.
# this is a more fine grained behavior introduced
# in 0.6
a1.user = u2
assert a1 not in u1.addresses
assert a1 in u2.addresses
@testing.resolve_artifact_names
def test_plain_load_passive(self):
"""test that many-to-one set doesn't load the old value."""
sess = sessionmaker()()
u1 = User(name='jack')
u2 = User(name='ed')
a1 = Address(email_address='a1')
a1.user = u1
sess.add_all([u1, u2, a1])
sess.commit()
# in this case, a lazyload would
# ordinarily occur except for the
# PASSIVE_NO_FETCH flag.
def go():
a1.user = u2
self.assert_sql_count(testing.db, go, 0)
assert a1 not in u1.addresses
assert a1 in u2.addresses
@testing.resolve_artifact_names
def test_set_none(self):
sess = sessionmaker()()
u1 = User(name='jack')
a1 = Address(email_address='a1')
a1.user = u1
sess.add_all([u1, a1])
sess.commit()
# works for None too
def go():
a1.user = None
self.assert_sql_count(testing.db, go, 0)
assert a1 not in u1.addresses
@testing.resolve_artifact_names
def test_scalar_move_notloaded(self):
sess = sessionmaker()()
u1 = User(name='jack')
u2 = User(name='ed')
a1 = Address(email_address='a1')
a1.user = u1
sess.add_all([u1, u2, a1])
sess.commit()
# direct set - the fetching of the
# "old" u1 here allows the backref
# to remove it from the addresses collection
a1.user = u2
assert a1 not in u1.addresses
assert a1 in u2.addresses
@testing.resolve_artifact_names
def test_scalar_move_commitfirst(self):
sess = sessionmaker()()
u1 = User(name='jack')
u2 = User(name='ed')
a1 = Address(email_address='a1')
a1.user = u1
sess.add_all([u1, u2, a1])
sess.commit()
# u1.addresses is loaded
u1.addresses
# direct set - the fetching of the
# "old" u1 here allows the backref
# to remove it from the addresses collection
a1.user = u2
sess.commit()
assert a1 not in u1.addresses
assert a1 in u2.addresses
class O2OScalarBackrefMoveTest(_fixtures.FixtureTest):
run_inserts = None
@classmethod
@testing.resolve_artifact_names
def setup_mappers(cls):
mapper(Address, addresses)
mapper(User, users, properties = {
'address':relationship(Address, backref=backref("user"), uselist=False)
})
@testing.resolve_artifact_names
def test_collection_move_preloaded(self):
sess = sessionmaker()()
a1 = Address(email_address="address1")
u1 = User(name='jack', address=a1)
u2 = User(name='ed')
sess.add_all([u1, u2])
sess.commit() # everything is expired
# load u1.address
u1.address
# reassign
u2.address = a1
assert u2.address is a1
# backref fires
assert a1.user is u2
# doesn't extend to the previous attribute tho.
# flushing at this point means its anyone's guess.
assert u1.address is a1
assert u2.address is a1
@testing.resolve_artifact_names
def test_scalar_move_preloaded(self):
sess = sessionmaker()()
a1 = Address(email_address="address1")
a2 = Address(email_address="address1")
u1 = User(name='jack', address=a1)
sess.add_all([u1, a1, a2])
sess.commit() # everything is expired
# load a1.user
a1.user
# reassign
a2.user = u1
# backref fires
assert u1.address is a2
# stays on both sides
assert a1.user is u1
assert a2.user is u1
@testing.resolve_artifact_names
def test_collection_move_notloaded(self):
sess = sessionmaker()()
a1 = Address(email_address="address1")
u1 = User(name='jack', address=a1)
u2 = User(name='ed')
sess.add_all([u1, u2])
sess.commit() # everything is expired
# reassign
u2.address = a1
assert u2.address is a1
# backref fires
assert a1.user is u2
# u1.address loads now after a flush
assert u1.address is None
assert u2.address is a1
@testing.resolve_artifact_names
def test_scalar_move_notloaded(self):
sess = sessionmaker()()
a1 = Address(email_address="address1")
a2 = Address(email_address="address1")
u1 = User(name='jack', address=a1)
sess.add_all([u1, a1, a2])
sess.commit() # everything is expired
# reassign
a2.user = u1
# backref fires
assert u1.address is a2
# stays on both sides
assert a1.user is u1
assert a2.user is u1
@testing.resolve_artifact_names
def test_collection_move_commitfirst(self):
sess = sessionmaker()()
a1 = Address(email_address="address1")
u1 = User(name='jack', address=a1)
u2 = User(name='ed')
sess.add_all([u1, u2])
sess.commit() # everything is expired
# load u1.address
u1.address
# reassign
u2.address = a1
assert u2.address is a1
# backref fires
assert a1.user is u2
# the commit cancels out u1.addresses
# being loaded, on next access its fine.
sess.commit()
assert u1.address is None
assert u2.address is a1
@testing.resolve_artifact_names
def test_scalar_move_commitfirst(self):
sess = sessionmaker()()
a1 = Address(email_address="address1")
a2 = Address(email_address="address2")
u1 = User(name='jack', address=a1)
sess.add_all([u1, a1, a2])
sess.commit() # everything is expired
# load
assert a1.user is u1
# reassign
a2.user = u1
# backref fires
assert u1.address is a2
# didnt work this way tho
assert a1.user is u1
# moves appropriately after commit
sess.commit()
assert u1.address is a2
assert a1.user is None
assert a2.user is u1
class O2OScalarMoveTest(_fixtures.FixtureTest):
run_inserts = None
@classmethod
@testing.resolve_artifact_names
def setup_mappers(cls):
mapper(Address, addresses)
mapper(User, users, properties = {
'address':relationship(Address, uselist=False)
})
@testing.resolve_artifact_names
def test_collection_move_commitfirst(self):
sess = sessionmaker()()
a1 = Address(email_address="address1")
u1 = User(name='jack', address=a1)
u2 = User(name='ed')
sess.add_all([u1, u2])
sess.commit() # everything is expired
# load u1.address
u1.address
# reassign
u2.address = a1
assert u2.address is a1
# the commit cancels out u1.addresses
# being loaded, on next access its fine.
sess.commit()
assert u1.address is None
assert u2.address is a1
class O2OScalarOrphanTest(_fixtures.FixtureTest):
run_inserts = None
@classmethod
@testing.resolve_artifact_names
def setup_mappers(cls):
mapper(Address, addresses)
mapper(User, users, properties = {
'address':relationship(Address, uselist=False,
backref=backref('user', single_parent=True, cascade="all, delete-orphan"))
})
@testing.resolve_artifact_names
def test_m2o_event(self):
sess = sessionmaker()()
a1 = Address(email_address="address1")
u1 = User(name='jack', address=a1)
sess.add(u1)
sess.commit()
sess.expunge(u1)
u2= User(name='ed')
# the _SingleParent extension sets the backref get to "active" !
# u1 gets loaded and deleted
u2.address = a1
sess.commit()
assert sess.query(User).count() == 1
class M2MScalarMoveTest(_fixtures.FixtureTest):
run_inserts = None
@classmethod
@testing.resolve_artifact_names
def setup_mappers(cls):
mapper(Item, items, properties={
'keyword':relationship(Keyword, secondary=item_keywords, uselist=False, backref=backref("item", uselist=False))
})
mapper(Keyword, keywords)
@testing.resolve_artifact_names
def test_collection_move_preloaded(self):
sess = sessionmaker()()
k1 = Keyword(name='k1')
i1 = Item(description='i1', keyword=k1)
i2 = Item(description='i2')
sess.add_all([i1, i2, k1])
sess.commit() # everything is expired
# load i1.keyword
assert i1.keyword is k1
i2.keyword = k1
assert k1.item is i2
# nothing happens.
assert i1.keyword is k1
assert i2.keyword is k1
@testing.resolve_artifact_names
def test_collection_move_notloaded(self):
sess = sessionmaker()()
k1 = Keyword(name='k1')
i1 = Item(description='i1', keyword=k1)
i2 = Item(description='i2')
sess.add_all([i1, i2, k1])
sess.commit() # everything is expired
i2.keyword = k1
assert k1.item is i2
assert i1.keyword is None
assert i2.keyword is k1
@testing.resolve_artifact_names
def test_collection_move_commit(self):
sess = sessionmaker()()
k1 = Keyword(name='k1')
i1 = Item(description='i1', keyword=k1)
i2 = Item(description='i2')
sess.add_all([i1, i2, k1])
sess.commit() # everything is expired
# load i1.keyword
assert i1.keyword is k1
i2.keyword = k1
assert k1.item is i2
sess.commit()
assert i1.keyword is None
assert i2.keyword is k1
|