mirror of
https://github.com/sqlalchemy/sqlalchemy.git
synced 2026-05-28 03:26:01 -04:00
- cleanup on these tests
- add "extend()" to AppenderQuery
This commit is contained in:
Mike Bayer
@@ -286,6 +286,12 @@ class AppenderMixin(object):
|
||||
|
||||
return query
|
||||
|
||||
def extend(self, iterator):
|
||||
for item in iterator:
|
||||
self.attr.append(
|
||||
attributes.instance_state(self.instance),
|
||||
attributes.instance_dict(self.instance), item, None)
|
||||
|
||||
def append(self, item):
|
||||
self.attr.append(
|
||||
attributes.instance_state(self.instance),
|
||||
|
||||
+217
-373
@@ -1,69 +1,60 @@
|
||||
from sqlalchemy.testing import eq_, ne_
|
||||
import operator
|
||||
from sqlalchemy.orm import dynamic_loader, backref, configure_mappers
|
||||
from sqlalchemy.testing import eq_
|
||||
from sqlalchemy.orm import backref, configure_mappers
|
||||
from sqlalchemy import testing
|
||||
from sqlalchemy import Integer, String, ForeignKey, desc, select, func, exc
|
||||
from sqlalchemy.testing.schema import Table, Column
|
||||
from sqlalchemy.orm import mapper, relationship, create_session, Query, attributes, exc as orm_exc
|
||||
from sqlalchemy import desc, select, func, exc
|
||||
from sqlalchemy.orm import mapper, relationship, create_session, Query, \
|
||||
attributes, exc as orm_exc
|
||||
from sqlalchemy.orm.dynamic import AppenderMixin
|
||||
from sqlalchemy.testing import eq_, AssertsCompiledSQL, assert_raises_message, assert_raises
|
||||
from sqlalchemy.testing import fixtures
|
||||
from sqlalchemy.testing import AssertsCompiledSQL, \
|
||||
assert_raises_message, assert_raises
|
||||
from test.orm import _fixtures
|
||||
|
||||
|
||||
class DynamicTest(_fixtures.FixtureTest, AssertsCompiledSQL):
|
||||
def test_basic(self):
|
||||
class _DynamicFixture(object):
|
||||
def _user_address_fixture(self, addresses_args={}):
|
||||
users, Address, addresses, User = (self.tables.users,
|
||||
self.classes.Address,
|
||||
self.tables.addresses,
|
||||
self.classes.User)
|
||||
|
||||
mapper(User, users, properties={
|
||||
'addresses':dynamic_loader(mapper(Address, addresses))
|
||||
'addresses': relationship(Address, lazy="dynamic",
|
||||
**addresses_args)
|
||||
})
|
||||
mapper(Address, addresses)
|
||||
return User, Address
|
||||
|
||||
class DynamicTest(_DynamicFixture, _fixtures.FixtureTest, AssertsCompiledSQL):
|
||||
|
||||
def test_basic(self):
|
||||
User, Address = self._user_address_fixture()
|
||||
sess = create_session()
|
||||
q = sess.query(User)
|
||||
|
||||
u = q.filter(User.id==7).first()
|
||||
|
||||
eq_([User(id=7,
|
||||
addresses=[Address(id=1, email_address='jack@bean.com')])],
|
||||
q.filter(User.id==7).all())
|
||||
q.filter(User.id == 7).all())
|
||||
eq_(self.static.user_address_result, q.all())
|
||||
|
||||
def test_statement(self):
|
||||
"""test that the .statement accessor returns the actual statement that
|
||||
would render, without any _clones called."""
|
||||
|
||||
users, Address, addresses, User = (self.tables.users,
|
||||
self.classes.Address,
|
||||
self.tables.addresses,
|
||||
self.classes.User)
|
||||
|
||||
|
||||
mapper(User, users, properties={
|
||||
'addresses':dynamic_loader(mapper(Address, addresses))
|
||||
})
|
||||
User, Address = self._user_address_fixture()
|
||||
sess = create_session()
|
||||
q = sess.query(User)
|
||||
|
||||
u = q.filter(User.id==7).first()
|
||||
u = q.filter(User.id == 7).first()
|
||||
self.assert_compile(
|
||||
u.addresses.statement,
|
||||
"SELECT addresses.id, addresses.user_id, addresses.email_address FROM "
|
||||
"SELECT addresses.id, addresses.user_id, addresses.email_address "
|
||||
"FROM "
|
||||
"addresses WHERE :param_1 = addresses.user_id",
|
||||
use_default_dialect=True
|
||||
)
|
||||
|
||||
def test_detached_raise(self):
|
||||
users, Address, addresses, User = (self.tables.users,
|
||||
self.classes.Address,
|
||||
self.tables.addresses,
|
||||
self.classes.User)
|
||||
|
||||
mapper(User, users, properties={
|
||||
'addresses': dynamic_loader(mapper(Address, addresses))
|
||||
})
|
||||
User, Address = self._user_address_fixture()
|
||||
sess = create_session()
|
||||
u = sess.query(User).get(8)
|
||||
sess.expunge(u)
|
||||
@@ -74,14 +65,8 @@ class DynamicTest(_fixtures.FixtureTest, AssertsCompiledSQL):
|
||||
)
|
||||
|
||||
def test_no_uselist_false(self):
|
||||
users, Address, addresses, User = (self.tables.users,
|
||||
self.classes.Address,
|
||||
self.tables.addresses,
|
||||
self.classes.User)
|
||||
mapper(Address, addresses)
|
||||
mapper(User, users, properties={
|
||||
"addresses": relationship(Address, lazy='dynamic', uselist=False)
|
||||
})
|
||||
User, Address = self._user_address_fixture(
|
||||
addresses_args={"uselist": False})
|
||||
assert_raises_message(
|
||||
exc.InvalidRequestError,
|
||||
"On relationship User.addresses, 'dynamic' loaders cannot be "
|
||||
@@ -108,68 +93,71 @@ class DynamicTest(_fixtures.FixtureTest, AssertsCompiledSQL):
|
||||
)
|
||||
|
||||
def test_order_by(self):
|
||||
users, Address, addresses, User = (self.tables.users,
|
||||
self.classes.Address,
|
||||
self.tables.addresses,
|
||||
self.classes.User)
|
||||
|
||||
mapper(User, users, properties={
|
||||
'addresses':dynamic_loader(mapper(Address, addresses))
|
||||
})
|
||||
User, Address = self._user_address_fixture()
|
||||
sess = create_session()
|
||||
u = sess.query(User).get(8)
|
||||
eq_(
|
||||
list(u.addresses.order_by(desc(Address.email_address))),
|
||||
[Address(email_address=u'ed@wood.com'), Address(email_address=u'ed@lala.com'),
|
||||
Address(email_address=u'ed@bettyboop.com')]
|
||||
)
|
||||
[
|
||||
Address(email_address=u'ed@wood.com'),
|
||||
Address(email_address=u'ed@lala.com'),
|
||||
Address(email_address=u'ed@bettyboop.com')
|
||||
]
|
||||
)
|
||||
|
||||
def test_configured_order_by(self):
|
||||
users, Address, addresses, User = (self.tables.users,
|
||||
self.classes.Address,
|
||||
self.tables.addresses,
|
||||
self.classes.User)
|
||||
addresses = self.tables.addresses
|
||||
User, Address = self._user_address_fixture(
|
||||
addresses_args={
|
||||
"order_by":
|
||||
addresses.c.email_address.desc()})
|
||||
|
||||
mapper(User, users, properties={
|
||||
'addresses':dynamic_loader(mapper(Address, addresses), order_by=desc(Address.email_address))
|
||||
})
|
||||
sess = create_session()
|
||||
u = sess.query(User).get(8)
|
||||
eq_(list(u.addresses), [Address(email_address=u'ed@wood.com'), Address(email_address=u'ed@lala.com'), Address(email_address=u'ed@bettyboop.com')])
|
||||
eq_(
|
||||
list(u.addresses),
|
||||
[
|
||||
Address(email_address=u'ed@wood.com'),
|
||||
Address(email_address=u'ed@lala.com'),
|
||||
Address(email_address=u'ed@bettyboop.com')
|
||||
]
|
||||
)
|
||||
|
||||
# test cancellation of None, replacement with something else
|
||||
eq_(
|
||||
list(u.addresses.order_by(None).order_by(Address.email_address)),
|
||||
[Address(email_address=u'ed@bettyboop.com'), Address(email_address=u'ed@lala.com'), Address(email_address=u'ed@wood.com')]
|
||||
[
|
||||
Address(email_address=u'ed@bettyboop.com'),
|
||||
Address(email_address=u'ed@lala.com'),
|
||||
Address(email_address=u'ed@wood.com')
|
||||
]
|
||||
)
|
||||
|
||||
# test cancellation of None, replacement with nothing
|
||||
eq_(
|
||||
set(u.addresses.order_by(None)),
|
||||
set([Address(email_address=u'ed@bettyboop.com'), Address(email_address=u'ed@lala.com'), Address(email_address=u'ed@wood.com')])
|
||||
set([
|
||||
Address(email_address=u'ed@bettyboop.com'),
|
||||
Address(email_address=u'ed@lala.com'),
|
||||
Address(email_address=u'ed@wood.com')
|
||||
])
|
||||
)
|
||||
|
||||
def test_count(self):
|
||||
users, Address, addresses, User = (self.tables.users,
|
||||
self.classes.Address,
|
||||
self.tables.addresses,
|
||||
self.classes.User)
|
||||
|
||||
mapper(User, users, properties={
|
||||
'addresses':dynamic_loader(mapper(Address, addresses))
|
||||
})
|
||||
User, Address = self._user_address_fixture()
|
||||
sess = create_session()
|
||||
u = sess.query(User).first()
|
||||
eq_(u.addresses.count(), 1)
|
||||
|
||||
def test_backref(self):
|
||||
def test_dynamic_on_backref(self):
|
||||
users, Address, addresses, User = (self.tables.users,
|
||||
self.classes.Address,
|
||||
self.tables.addresses,
|
||||
self.classes.User)
|
||||
|
||||
mapper(Address, addresses, properties={
|
||||
'user':relationship(User, backref=backref('addresses', lazy='dynamic'))
|
||||
'user': relationship(User,
|
||||
backref=backref('addresses', lazy='dynamic'))
|
||||
})
|
||||
mapper(User, users)
|
||||
|
||||
@@ -183,14 +171,7 @@ class DynamicTest(_fixtures.FixtureTest, AssertsCompiledSQL):
|
||||
assert ad not in u.addresses
|
||||
|
||||
def test_no_count(self):
|
||||
users, Address, addresses, User = (self.tables.users,
|
||||
self.classes.Address,
|
||||
self.tables.addresses,
|
||||
self.classes.User)
|
||||
|
||||
mapper(User, users, properties={
|
||||
'addresses':dynamic_loader(mapper(Address, addresses))
|
||||
})
|
||||
User, Address = self._user_address_fixture()
|
||||
sess = create_session()
|
||||
q = sess.query(User)
|
||||
|
||||
@@ -198,21 +179,19 @@ class DynamicTest(_fixtures.FixtureTest, AssertsCompiledSQL):
|
||||
# returns a live database result), else additional count() queries are
|
||||
# issued when evaluating in a list context
|
||||
def go():
|
||||
eq_([User(id=7,
|
||||
addresses=[Address(id=1,
|
||||
email_address='jack@bean.com')])],
|
||||
q.filter(User.id==7).all())
|
||||
eq_(
|
||||
q.filter(User.id == 7).all(),
|
||||
[
|
||||
User(id=7,
|
||||
addresses=[
|
||||
Address(id=1, email_address='jack@bean.com')
|
||||
])
|
||||
]
|
||||
)
|
||||
self.assert_sql_count(testing.db, go, 2)
|
||||
|
||||
def test_no_populate(self):
|
||||
users, Address, addresses, User = (self.tables.users,
|
||||
self.classes.Address,
|
||||
self.tables.addresses,
|
||||
self.classes.User)
|
||||
|
||||
mapper(User, users, properties={
|
||||
'addresses':dynamic_loader(mapper(Address, addresses))
|
||||
})
|
||||
User, Address = self._user_address_fixture()
|
||||
u1 = User()
|
||||
assert_raises_message(
|
||||
NotImplementedError,
|
||||
@@ -228,8 +207,11 @@ class DynamicTest(_fixtures.FixtureTest, AssertsCompiledSQL):
|
||||
self.classes.Item)
|
||||
|
||||
mapper(Order, orders, properties={
|
||||
'items':relationship(Item, secondary=order_items, lazy="dynamic",
|
||||
backref=backref('orders', lazy="dynamic"))
|
||||
'items': relationship(Item,
|
||||
secondary=order_items,
|
||||
lazy="dynamic",
|
||||
backref=backref('orders', lazy="dynamic")
|
||||
)
|
||||
})
|
||||
mapper(Item, items)
|
||||
|
||||
@@ -244,7 +226,7 @@ class DynamicTest(_fixtures.FixtureTest, AssertsCompiledSQL):
|
||||
assert i1 in o1.items.all()
|
||||
|
||||
@testing.exclude('mysql', 'between',
|
||||
((5, 1,49), (5, 1, 52)),
|
||||
((5, 1, 49), (5, 1, 52)),
|
||||
'https://bugs.launchpad.net/ubuntu/+source/mysql-5.1/+bug/706988')
|
||||
def test_association_nonaliased(self):
|
||||
items, Order, orders, order_items, Item = (self.tables.items,
|
||||
@@ -254,7 +236,8 @@ class DynamicTest(_fixtures.FixtureTest, AssertsCompiledSQL):
|
||||
self.classes.Item)
|
||||
|
||||
mapper(Order, orders, properties={
|
||||
'items':relationship(Item, secondary=order_items,
|
||||
'items': relationship(Item,
|
||||
secondary=order_items,
|
||||
lazy="dynamic",
|
||||
order_by=order_items.c.item_id)
|
||||
})
|
||||
@@ -276,39 +259,28 @@ class DynamicTest(_fixtures.FixtureTest, AssertsCompiledSQL):
|
||||
# filter criterion against the secondary table
|
||||
# works
|
||||
eq_(
|
||||
o.items.filter(order_items.c.item_id==2).all(),
|
||||
o.items.filter(order_items.c.item_id == 2).all(),
|
||||
[Item(id=2)]
|
||||
)
|
||||
|
||||
|
||||
def test_transient_detached(self):
|
||||
users, Address, addresses, User = (self.tables.users,
|
||||
self.classes.Address,
|
||||
self.tables.addresses,
|
||||
self.classes.User)
|
||||
|
||||
mapper(User, users, properties={
|
||||
'addresses':dynamic_loader(mapper(Address, addresses))
|
||||
})
|
||||
sess = create_session()
|
||||
def test_transient_count(self):
|
||||
User, Address = self._user_address_fixture()
|
||||
u1 = User()
|
||||
u1.addresses.append(Address())
|
||||
eq_(u1.addresses.count(), 1)
|
||||
|
||||
def test_transient_access(self):
|
||||
User, Address = self._user_address_fixture()
|
||||
u1 = User()
|
||||
u1.addresses.append(Address())
|
||||
eq_(u1.addresses[0], Address())
|
||||
|
||||
def test_custom_query(self):
|
||||
users, Address, addresses, User = (self.tables.users,
|
||||
self.classes.Address,
|
||||
self.tables.addresses,
|
||||
self.classes.User)
|
||||
|
||||
class MyQuery(Query):
|
||||
pass
|
||||
User, Address = self._user_address_fixture(
|
||||
addresses_args={"query_class": MyQuery})
|
||||
|
||||
mapper(User, users, properties={
|
||||
'addresses':dynamic_loader(mapper(Address, addresses),
|
||||
query_class=MyQuery)
|
||||
})
|
||||
sess = create_session()
|
||||
u = User()
|
||||
sess.add(u)
|
||||
@@ -326,11 +298,6 @@ class DynamicTest(_fixtures.FixtureTest, AssertsCompiledSQL):
|
||||
eq_(type(q).__name__, 'MyQuery')
|
||||
|
||||
def test_custom_query_with_custom_mixin(self):
|
||||
users, Address, addresses, User = (self.tables.users,
|
||||
self.classes.Address,
|
||||
self.tables.addresses,
|
||||
self.classes.User)
|
||||
|
||||
class MyAppenderMixin(AppenderMixin):
|
||||
def add(self, items):
|
||||
if isinstance(items, list):
|
||||
@@ -345,10 +312,11 @@ class DynamicTest(_fixtures.FixtureTest, AssertsCompiledSQL):
|
||||
class MyAppenderQuery(MyAppenderMixin, MyQuery):
|
||||
query_class = MyQuery
|
||||
|
||||
mapper(User, users, properties={
|
||||
'addresses':dynamic_loader(mapper(Address, addresses),
|
||||
query_class=MyAppenderQuery)
|
||||
})
|
||||
User, Address = self._user_address_fixture(
|
||||
addresses_args={
|
||||
"query_class": MyAppenderQuery})
|
||||
|
||||
|
||||
sess = create_session()
|
||||
u = User()
|
||||
sess.add(u)
|
||||
@@ -368,58 +336,71 @@ class DynamicTest(_fixtures.FixtureTest, AssertsCompiledSQL):
|
||||
eq_(type(q).__name__, 'MyQuery')
|
||||
|
||||
|
||||
class SessionTest(_fixtures.FixtureTest):
|
||||
class UOWTest(_DynamicFixture, _fixtures.FixtureTest):
|
||||
run_inserts = None
|
||||
|
||||
def test_events(self):
|
||||
users, Address, addresses, User = (self.tables.users,
|
||||
self.classes.Address,
|
||||
self.tables.addresses,
|
||||
self.classes.User)
|
||||
def test_persistence(self):
|
||||
addresses = self.tables.addresses
|
||||
User, Address = self._user_address_fixture()
|
||||
|
||||
mapper(User, users, properties={
|
||||
'addresses':dynamic_loader(mapper(Address, addresses))
|
||||
})
|
||||
sess = create_session()
|
||||
u1 = User(name='jack')
|
||||
a1 = Address(email_address='foo')
|
||||
sess.add_all([u1, a1])
|
||||
sess.flush()
|
||||
|
||||
eq_(testing.db.scalar(select([func.count(1)]).where(addresses.c.user_id!=None)), 0)
|
||||
eq_(
|
||||
testing.db.scalar(
|
||||
select([func.count(1)]).where(addresses.c.user_id != None)
|
||||
),
|
||||
0
|
||||
)
|
||||
u1 = sess.query(User).get(u1.id)
|
||||
u1.addresses.append(a1)
|
||||
sess.flush()
|
||||
|
||||
eq_(testing.db.execute(select([addresses]).where(addresses.c.user_id!=None)).fetchall(),
|
||||
[(a1.id, u1.id, 'foo')])
|
||||
eq_(
|
||||
testing.db.execute(
|
||||
select([addresses]).where(addresses.c.user_id != None)
|
||||
).fetchall(),
|
||||
[(a1.id, u1.id, 'foo')]
|
||||
)
|
||||
|
||||
u1.addresses.remove(a1)
|
||||
sess.flush()
|
||||
eq_(testing.db.scalar(select([func.count(1)]).where(addresses.c.user_id!=None)), 0)
|
||||
eq_(
|
||||
testing.db.scalar(
|
||||
select([func.count(1)]).where(addresses.c.user_id != None)
|
||||
),
|
||||
0
|
||||
)
|
||||
|
||||
u1.addresses.append(a1)
|
||||
sess.flush()
|
||||
eq_(testing.db.execute(select([addresses]).where(addresses.c.user_id!=None)).fetchall(),
|
||||
[(a1.id, u1.id, 'foo')])
|
||||
eq_(
|
||||
testing.db.execute(
|
||||
select([addresses]).where(addresses.c.user_id != None)
|
||||
).fetchall(),
|
||||
[(a1.id, u1.id, 'foo')]
|
||||
)
|
||||
|
||||
a2 = Address(email_address='bar')
|
||||
u1.addresses.remove(a1)
|
||||
u1.addresses.append(a2)
|
||||
sess.flush()
|
||||
eq_(testing.db.execute(select([addresses]).where(addresses.c.user_id!=None)).fetchall(),
|
||||
[(a2.id, u1.id, 'bar')])
|
||||
eq_(
|
||||
testing.db.execute(
|
||||
select([addresses]).where(addresses.c.user_id != None)
|
||||
).fetchall(),
|
||||
[(a2.id, u1.id, 'bar')]
|
||||
)
|
||||
|
||||
|
||||
def test_merge(self):
|
||||
users, Address, addresses, User = (self.tables.users,
|
||||
self.classes.Address,
|
||||
self.tables.addresses,
|
||||
self.classes.User)
|
||||
|
||||
mapper(User, users, properties={
|
||||
'addresses':dynamic_loader(mapper(Address, addresses), order_by=addresses.c.email_address)
|
||||
})
|
||||
addresses = self.tables.addresses
|
||||
User, Address = self._user_address_fixture(
|
||||
addresses_args={
|
||||
"order_by": addresses.c.email_address})
|
||||
sess = create_session()
|
||||
u1 = User(name='jack')
|
||||
a1 = Address(email_address='a1')
|
||||
@@ -449,64 +430,20 @@ class SessionTest(_fixtures.FixtureTest):
|
||||
[a1, a3]
|
||||
)
|
||||
|
||||
def test_flush(self):
|
||||
users, Address, addresses, User = (self.tables.users,
|
||||
self.classes.Address,
|
||||
self.tables.addresses,
|
||||
self.classes.User)
|
||||
|
||||
mapper(User, users, properties={
|
||||
'addresses':dynamic_loader(mapper(Address, addresses))
|
||||
})
|
||||
sess = create_session()
|
||||
u1 = User(name='jack')
|
||||
u2 = User(name='ed')
|
||||
u2.addresses.append(Address(email_address='foo@bar.com'))
|
||||
u1.addresses.append(Address(email_address='lala@hoho.com'))
|
||||
sess.add_all((u1, u2))
|
||||
sess.flush()
|
||||
|
||||
from sqlalchemy.orm import attributes
|
||||
eq_(attributes.get_history(u1, 'addresses'), ([], [Address(email_address='lala@hoho.com')], []))
|
||||
|
||||
sess.expunge_all()
|
||||
|
||||
# test the test fixture a little bit
|
||||
ne_(User(name='jack', addresses=[Address(email_address='wrong')]),
|
||||
sess.query(User).first())
|
||||
eq_(User(name='jack', addresses=[Address(email_address='lala@hoho.com')]),
|
||||
sess.query(User).first())
|
||||
|
||||
eq_([
|
||||
User(name='jack', addresses=[Address(email_address='lala@hoho.com')]),
|
||||
User(name='ed', addresses=[Address(email_address='foo@bar.com')])
|
||||
],
|
||||
sess.query(User).all())
|
||||
|
||||
def test_hasattr(self):
|
||||
users, Address, addresses, User = (self.tables.users,
|
||||
self.classes.Address,
|
||||
self.tables.addresses,
|
||||
self.classes.User)
|
||||
User, Address = self._user_address_fixture()
|
||||
|
||||
mapper(User, users, properties={
|
||||
'addresses':dynamic_loader(mapper(Address, addresses))
|
||||
})
|
||||
u1 = User(name='jack')
|
||||
|
||||
assert 'addresses' not in u1.__dict__.keys()
|
||||
assert 'addresses' not in u1.__dict__
|
||||
u1.addresses = [Address(email_address='test')]
|
||||
assert 'addresses' in dir(u1)
|
||||
assert 'addresses' in u1.__dict__
|
||||
|
||||
def test_collection_set(self):
|
||||
users, Address, addresses, User = (self.tables.users,
|
||||
self.classes.Address,
|
||||
self.tables.addresses,
|
||||
self.classes.User)
|
||||
|
||||
mapper(User, users, properties={
|
||||
'addresses':dynamic_loader(mapper(Address, addresses), order_by=addresses.c.email_address)
|
||||
})
|
||||
addresses = self.tables.addresses
|
||||
User, Address = self._user_address_fixture(
|
||||
addresses_args={
|
||||
"order_by": addresses.c.email_address})
|
||||
sess = create_session(autoflush=True, autocommit=False)
|
||||
u1 = User(name='jack')
|
||||
a1 = Address(email_address='a1')
|
||||
@@ -525,161 +462,95 @@ class SessionTest(_fixtures.FixtureTest):
|
||||
eq_(list(u1.addresses), [])
|
||||
|
||||
def test_rollback(self):
|
||||
users, Address, addresses, User = (self.tables.users,
|
||||
self.classes.Address,
|
||||
self.tables.addresses,
|
||||
self.classes.User)
|
||||
|
||||
mapper(User, users, properties={
|
||||
'addresses':dynamic_loader(mapper(Address, addresses))
|
||||
})
|
||||
sess = create_session(expire_on_commit=False, autocommit=False, autoflush=True)
|
||||
User, Address = self._user_address_fixture()
|
||||
sess = create_session(
|
||||
expire_on_commit=False, autocommit=False, autoflush=True)
|
||||
u1 = User(name='jack')
|
||||
u1.addresses.append(Address(email_address='lala@hoho.com'))
|
||||
sess.add(u1)
|
||||
sess.flush()
|
||||
sess.commit()
|
||||
u1.addresses.append(Address(email_address='foo@bar.com'))
|
||||
eq_(u1.addresses.order_by(Address.id).all(),
|
||||
[Address(email_address='lala@hoho.com'), Address(email_address='foo@bar.com')])
|
||||
eq_(
|
||||
u1.addresses.order_by(Address.id).all(),
|
||||
[
|
||||
Address(email_address='lala@hoho.com'),
|
||||
Address(email_address='foo@bar.com')
|
||||
]
|
||||
)
|
||||
sess.rollback()
|
||||
eq_(u1.addresses.all(), [Address(email_address='lala@hoho.com')])
|
||||
|
||||
@testing.fails_on('maxdb', 'FIXME: unknown')
|
||||
def test_delete_nocascade(self):
|
||||
users, Address, addresses, User = (self.tables.users,
|
||||
self.classes.Address,
|
||||
self.tables.addresses,
|
||||
self.classes.User)
|
||||
|
||||
mapper(User, users, properties={
|
||||
'addresses':dynamic_loader(mapper(Address, addresses), order_by=Address.id,
|
||||
backref='user')
|
||||
})
|
||||
sess = create_session(autoflush=True, autocommit=False)
|
||||
u = User(name='ed')
|
||||
u.addresses.append(Address(email_address='a'))
|
||||
u.addresses.append(Address(email_address='b'))
|
||||
u.addresses.append(Address(email_address='c'))
|
||||
u.addresses.append(Address(email_address='d'))
|
||||
u.addresses.append(Address(email_address='e'))
|
||||
u.addresses.append(Address(email_address='f'))
|
||||
sess.add(u)
|
||||
|
||||
eq_(Address(email_address='c'), u.addresses[2])
|
||||
sess.delete(u.addresses[2])
|
||||
sess.delete(u.addresses[4])
|
||||
sess.delete(u.addresses[3])
|
||||
eq_([Address(email_address='a'), Address(email_address='b'), Address(email_address='d')],
|
||||
list(u.addresses))
|
||||
|
||||
sess.expunge_all()
|
||||
u = sess.query(User).get(u.id)
|
||||
|
||||
sess.delete(u)
|
||||
|
||||
# u.addresses relationship will have to force the load
|
||||
# of all addresses so that they can be updated
|
||||
sess.flush()
|
||||
sess.close()
|
||||
|
||||
eq_(testing.db.scalar(addresses.count(addresses.c.user_id != None)), 0)
|
||||
|
||||
@testing.fails_on('maxdb', 'FIXME: unknown')
|
||||
def test_delete_cascade(self):
|
||||
users, Address, addresses, User = (self.tables.users,
|
||||
self.classes.Address,
|
||||
self.tables.addresses,
|
||||
self.classes.User)
|
||||
|
||||
mapper(User, users, properties={
|
||||
'addresses':dynamic_loader(mapper(Address, addresses), order_by=Address.id,
|
||||
backref='user', cascade="all, delete-orphan")
|
||||
})
|
||||
sess = create_session(autoflush=True, autocommit=False)
|
||||
u = User(name='ed')
|
||||
u.addresses.append(Address(email_address='a'))
|
||||
u.addresses.append(Address(email_address='b'))
|
||||
u.addresses.append(Address(email_address='c'))
|
||||
u.addresses.append(Address(email_address='d'))
|
||||
u.addresses.append(Address(email_address='e'))
|
||||
u.addresses.append(Address(email_address='f'))
|
||||
sess.add(u)
|
||||
|
||||
eq_(Address(email_address='c'), u.addresses[2])
|
||||
sess.delete(u.addresses[2])
|
||||
sess.delete(u.addresses[4])
|
||||
sess.delete(u.addresses[3])
|
||||
eq_([Address(email_address='a'), Address(email_address='b'), Address(email_address='d')],
|
||||
list(u.addresses))
|
||||
|
||||
sess.expunge_all()
|
||||
u = sess.query(User).get(u.id)
|
||||
|
||||
sess.delete(u)
|
||||
|
||||
# u.addresses relationship will have to force the load
|
||||
# of all addresses so that they can be updated
|
||||
sess.flush()
|
||||
sess.close()
|
||||
|
||||
eq_(testing.db.scalar(addresses.count()), 0)
|
||||
|
||||
@testing.fails_on('maxdb', 'FIXME: unknown')
|
||||
def test_remove_orphans(self):
|
||||
users, Address, addresses, User = (self.tables.users,
|
||||
self.classes.Address,
|
||||
self.tables.addresses,
|
||||
self.classes.User)
|
||||
|
||||
mapper(User, users, properties={
|
||||
'addresses':dynamic_loader(mapper(Address, addresses), order_by=Address.id,
|
||||
cascade="all, delete-orphan", backref='user')
|
||||
})
|
||||
sess = create_session(autoflush=True, autocommit=False)
|
||||
u = User(name='ed')
|
||||
u.addresses.append(Address(email_address='a'))
|
||||
u.addresses.append(Address(email_address='b'))
|
||||
u.addresses.append(Address(email_address='c'))
|
||||
u.addresses.append(Address(email_address='d'))
|
||||
u.addresses.append(Address(email_address='e'))
|
||||
u.addresses.append(Address(email_address='f'))
|
||||
sess.add(u)
|
||||
|
||||
eq_([Address(email_address='a'), Address(email_address='b'), Address(email_address='c'),
|
||||
Address(email_address='d'), Address(email_address='e'), Address(email_address='f')],
|
||||
sess.query(Address).all())
|
||||
|
||||
eq_(Address(email_address='c'), u.addresses[2])
|
||||
|
||||
def go():
|
||||
del u.addresses[3]
|
||||
assert_raises(
|
||||
TypeError,
|
||||
go
|
||||
eq_(
|
||||
u1.addresses.all(),
|
||||
[Address(email_address='lala@hoho.com')]
|
||||
)
|
||||
|
||||
for a in u.addresses.filter(Address.email_address.in_(['c', 'e', 'f'])):
|
||||
u.addresses.remove(a)
|
||||
def _test_delete_cascade(self, expected):
|
||||
addresses = self.tables.addresses
|
||||
User, Address = self._user_address_fixture(addresses_args={
|
||||
"order_by": addresses.c.id,
|
||||
"backref": "user",
|
||||
"cascade": "save-update" if expected \
|
||||
else "all, delete"
|
||||
})
|
||||
|
||||
eq_([Address(email_address='a'), Address(email_address='b'), Address(email_address='d')],
|
||||
list(u.addresses))
|
||||
|
||||
eq_([Address(email_address='a'), Address(email_address='b'), Address(email_address='d')],
|
||||
sess.query(Address).all())
|
||||
sess = create_session(autoflush=True, autocommit=False)
|
||||
u = User(name='ed')
|
||||
u.addresses.extend(
|
||||
[Address(email_address=letter) for letter in 'abcdef']
|
||||
)
|
||||
sess.add(u)
|
||||
sess.commit()
|
||||
eq_(testing.db.scalar(
|
||||
addresses.count(addresses.c.user_id == None)), 0)
|
||||
eq_(testing.db.scalar(
|
||||
addresses.count(addresses.c.user_id != None)), 6)
|
||||
|
||||
sess.delete(u)
|
||||
sess.close()
|
||||
|
||||
sess.commit()
|
||||
|
||||
if expected:
|
||||
eq_(testing.db.scalar(
|
||||
addresses.count(addresses.c.user_id == None)), 6)
|
||||
eq_(testing.db.scalar(
|
||||
addresses.count(addresses.c.user_id != None)), 0)
|
||||
else:
|
||||
eq_(testing.db.scalar(addresses.count()), 0)
|
||||
|
||||
def test_delete_nocascade(self):
|
||||
self._test_delete_cascade(True)
|
||||
|
||||
def test_delete_cascade(self):
|
||||
self._test_delete_cascade(False)
|
||||
|
||||
def test_remove_orphans(self):
|
||||
addresses = self.tables.addresses
|
||||
User, Address = self._user_address_fixture(addresses_args={
|
||||
"order_by": addresses.c.id,
|
||||
"backref": "user",
|
||||
"cascade": "all, delete-orphan"
|
||||
})
|
||||
|
||||
sess = create_session(autoflush=True, autocommit=False)
|
||||
u = User(name='ed')
|
||||
u.addresses.extend(
|
||||
[Address(email_address=letter) for letter in 'abcdef']
|
||||
)
|
||||
sess.add(u)
|
||||
|
||||
for a in u.addresses.filter(
|
||||
Address.email_address.in_(['c', 'e', 'f'])):
|
||||
u.addresses.remove(a)
|
||||
|
||||
eq_(
|
||||
set(ad for ad, in sess.query(Address.email_address)),
|
||||
set(['a', 'b', 'd'])
|
||||
)
|
||||
|
||||
def _backref_test(self, autoflush, saveuser):
|
||||
users, Address, addresses, User = (self.tables.users,
|
||||
self.classes.Address,
|
||||
self.tables.addresses,
|
||||
self.classes.User)
|
||||
|
||||
mapper(User, users, properties={
|
||||
'addresses':dynamic_loader(mapper(Address, addresses), backref='user')
|
||||
})
|
||||
User, Address = self._user_address_fixture(addresses_args={
|
||||
"backref": "user",
|
||||
})
|
||||
sess = create_session(autoflush=autoflush, autocommit=False)
|
||||
|
||||
u = User(name='buffy')
|
||||
@@ -720,37 +591,10 @@ class SessionTest(_fixtures.FixtureTest):
|
||||
def test_backref_savead(self):
|
||||
self._backref_test(False, False)
|
||||
|
||||
class DontDereferenceTest(fixtures.MappedTest):
|
||||
@classmethod
|
||||
def define_tables(cls, metadata):
|
||||
Table('users', metadata,
|
||||
Column('id', Integer, primary_key=True, test_needs_autoincrement=True),
|
||||
Column('name', String(40)),
|
||||
Column('fullname', String(100)),
|
||||
Column('password', String(15)))
|
||||
|
||||
Table('addresses', metadata,
|
||||
Column('id', Integer, primary_key=True, test_needs_autoincrement=True),
|
||||
Column('email_address', String(100), nullable=False),
|
||||
Column('user_id', Integer, ForeignKey('users.id')))
|
||||
|
||||
@classmethod
|
||||
def setup_mappers(cls):
|
||||
users, addresses = cls.tables.users, cls.tables.addresses
|
||||
|
||||
class User(cls.Comparable):
|
||||
pass
|
||||
|
||||
class Address(cls.Comparable):
|
||||
pass
|
||||
|
||||
mapper(User, users, properties={
|
||||
'addresses': relationship(Address, backref='user', lazy='dynamic')
|
||||
})
|
||||
mapper(Address, addresses)
|
||||
|
||||
def test_no_deref(self):
|
||||
User, Address = self.classes.User, self.classes.Address
|
||||
User, Address = self._user_address_fixture(addresses_args={
|
||||
"backref": "user",
|
||||
})
|
||||
|
||||
session = create_session()
|
||||
user = User()
|
||||
|
||||
Reference in New Issue
Block a user