diff options
| author | Mike Bayer <mike_mp@zzzcomputing.com> | 2009-06-10 21:18:24 +0000 |
|---|---|---|
| committer | Mike Bayer <mike_mp@zzzcomputing.com> | 2009-06-10 21:18:24 +0000 |
| commit | 45cec095b4904ba71425d2fe18c143982dd08f43 (patch) | |
| tree | af5e540fdcbf1cb2a3337157d69d4b40be010fa8 /test/orm/lazy_relations.py | |
| parent | 698a3c1ac665e7cd2ef8d5ad3ebf51b7fe6661f4 (diff) | |
| download | sqlalchemy-45cec095b4904ba71425d2fe18c143982dd08f43.tar.gz | |
- unit tests have been migrated from unittest to nose.
See README.unittests for information on how to run
the tests. [ticket:970]
Diffstat (limited to 'test/orm/lazy_relations.py')
| -rw-r--r-- | test/orm/lazy_relations.py | 416 |
1 files changed, 0 insertions, 416 deletions
diff --git a/test/orm/lazy_relations.py b/test/orm/lazy_relations.py deleted file mode 100644 index b5c3b3669..000000000 --- a/test/orm/lazy_relations.py +++ /dev/null @@ -1,416 +0,0 @@ -"""basic tests of lazy loaded attributes""" - -import testenv; testenv.configure_for_tests() -import datetime -from sqlalchemy import exc as sa_exc -from sqlalchemy.orm import attributes -from testlib import sa, testing -from testlib.sa import Table, Column, Integer, String, ForeignKey -from testlib.sa.orm import mapper, relation, create_session -from testlib.testing import eq_ -from orm import _base, _fixtures - - -class LazyTest(_fixtures.FixtureTest): - run_inserts = 'once' - run_deletes = None - - @testing.resolve_artifact_names - def test_basic(self): - mapper(User, users, properties={ - 'addresses':relation(mapper(Address, addresses), lazy=True) - }) - sess = create_session() - q = sess.query(User) - assert [User(id=7, addresses=[Address(id=1, email_address='jack@bean.com')])] == q.filter(users.c.id == 7).all() - - @testing.resolve_artifact_names - def test_needs_parent(self): - """test the error raised when parent object is not bound.""" - - mapper(User, users, properties={ - 'addresses':relation(mapper(Address, addresses), lazy=True) - }) - sess = create_session() - q = sess.query(User) - u = q.filter(users.c.id == 7).first() - sess.expunge(u) - self.assertRaises(sa_exc.InvalidRequestError, getattr, u, 'addresses') - - @testing.resolve_artifact_names - def test_orderby(self): - mapper(User, users, properties = { - 'addresses':relation(mapper(Address, addresses), lazy=True, order_by=addresses.c.email_address), - }) - q = create_session().query(User) - assert [ - User(id=7, addresses=[ - Address(id=1) - ]), - User(id=8, addresses=[ - Address(id=3, email_address='ed@bettyboop.com'), - Address(id=4, email_address='ed@lala.com'), - Address(id=2, email_address='ed@wood.com') - ]), - User(id=9, addresses=[ - Address(id=5) - ]), - User(id=10, addresses=[]) - ] == q.all() - - @testing.resolve_artifact_names - def test_orderby_secondary(self): - """tests that a regular mapper select on a single table can order by a relation to a second table""" - - mapper(Address, addresses) - - mapper(User, users, properties = dict( - addresses = relation(Address, lazy=True), - )) - q = create_session().query(User) - l = q.filter(users.c.id==addresses.c.user_id).order_by(addresses.c.email_address).all() - assert [ - User(id=8, addresses=[ - Address(id=2, email_address='ed@wood.com'), - Address(id=3, email_address='ed@bettyboop.com'), - Address(id=4, email_address='ed@lala.com'), - ]), - User(id=9, addresses=[ - Address(id=5) - ]), - User(id=7, addresses=[ - Address(id=1) - ]), - ] == l - - @testing.resolve_artifact_names - def test_orderby_desc(self): - mapper(Address, addresses) - - mapper(User, users, properties = dict( - addresses = relation(Address, lazy=True, order_by=[sa.desc(addresses.c.email_address)]), - )) - sess = create_session() - assert [ - User(id=7, addresses=[ - Address(id=1) - ]), - User(id=8, addresses=[ - Address(id=2, email_address='ed@wood.com'), - Address(id=4, email_address='ed@lala.com'), - Address(id=3, email_address='ed@bettyboop.com'), - ]), - User(id=9, addresses=[ - Address(id=5) - ]), - User(id=10, addresses=[]) - ] == sess.query(User).all() - - @testing.resolve_artifact_names - def test_no_orphan(self): - """test that a lazily loaded child object is not marked as an orphan""" - - mapper(User, users, properties={ - 'addresses':relation(Address, cascade="all,delete-orphan", lazy=True) - }) - mapper(Address, addresses) - - sess = create_session() - user = sess.query(User).get(7) - assert getattr(User, 'addresses').hasparent(attributes.instance_state(user.addresses[0]), optimistic=True) - assert not sa.orm.class_mapper(Address)._is_orphan(attributes.instance_state(user.addresses[0])) - - @testing.resolve_artifact_names - def test_limit(self): - """test limit operations combined with lazy-load relationships.""" - - mapper(Item, items) - mapper(Order, orders, properties={ - 'items':relation(Item, secondary=order_items, lazy=True) - }) - mapper(User, users, properties={ - 'addresses':relation(mapper(Address, addresses), lazy=True), - 'orders':relation(Order, lazy=True) - }) - - sess = create_session() - q = sess.query(User) - - if testing.against('maxdb', 'mssql'): - l = q.limit(2).all() - assert self.static.user_all_result[:2] == l - else: - l = q.limit(2).offset(1).all() - assert self.static.user_all_result[1:3] == l - - @testing.resolve_artifact_names - def test_distinct(self): - mapper(Item, items) - mapper(Order, orders, properties={ - 'items':relation(Item, secondary=order_items, lazy=True) - }) - mapper(User, users, properties={ - 'addresses':relation(mapper(Address, addresses), lazy=True), - 'orders':relation(Order, lazy=True) - }) - - sess = create_session() - q = sess.query(User) - - # use a union all to get a lot of rows to join against - u2 = users.alias('u2') - s = sa.union_all(u2.select(use_labels=True), u2.select(use_labels=True), u2.select(use_labels=True)).alias('u') - print [key for key in s.c.keys()] - l = q.filter(s.c.u2_id==User.id).distinct().all() - assert self.static.user_all_result == l - - @testing.resolve_artifact_names - def test_one_to_many_scalar(self): - mapper(User, users, properties = dict( - address = relation(mapper(Address, addresses), lazy=True, uselist=False) - )) - q = create_session().query(User) - l = q.filter(users.c.id == 7).all() - assert [User(id=7, address=Address(id=1))] == l - - @testing.resolve_artifact_names - def test_many_to_one_binds(self): - mapper(Address, addresses, primary_key=[addresses.c.user_id, addresses.c.email_address]) - - mapper(User, users, properties = dict( - address = relation(Address, uselist=False, - primaryjoin=sa.and_(users.c.id==addresses.c.user_id, addresses.c.email_address=='ed@bettyboop.com') - ) - )) - q = create_session().query(User) - eq_( - [ - User(id=7, address=None), - User(id=8, address=Address(id=3)), - User(id=9, address=None), - User(id=10, address=None), - ], - list(q) - ) - - - @testing.resolve_artifact_names - def test_double(self): - """tests lazy loading with two relations simulatneously, from the same table, using aliases. """ - - openorders = sa.alias(orders, 'openorders') - closedorders = sa.alias(orders, 'closedorders') - - mapper(Address, addresses) - - mapper(Order, orders) - - open_mapper = mapper(Order, openorders, non_primary=True) - closed_mapper = mapper(Order, closedorders, non_primary=True) - mapper(User, users, properties = dict( - addresses = relation(Address, lazy = True), - open_orders = relation(open_mapper, primaryjoin = sa.and_(openorders.c.isopen == 1, users.c.id==openorders.c.user_id), lazy=True), - closed_orders = relation(closed_mapper, primaryjoin = sa.and_(closedorders.c.isopen == 0, users.c.id==closedorders.c.user_id), lazy=True) - )) - q = create_session().query(User) - - assert [ - User( - id=7, - addresses=[Address(id=1)], - open_orders = [Order(id=3)], - closed_orders = [Order(id=1), Order(id=5)] - ), - User( - id=8, - addresses=[Address(id=2), Address(id=3), Address(id=4)], - open_orders = [], - closed_orders = [] - ), - User( - id=9, - addresses=[Address(id=5)], - open_orders = [Order(id=4)], - closed_orders = [Order(id=2)] - ), - User(id=10) - - ] == q.all() - - sess = create_session() - user = sess.query(User).get(7) - assert [Order(id=1), Order(id=5)] == create_session().query(closed_mapper).with_parent(user, property='closed_orders').all() - assert [Order(id=3)] == create_session().query(open_mapper).with_parent(user, property='open_orders').all() - - @testing.resolve_artifact_names - def test_many_to_many(self): - - mapper(Keyword, keywords) - mapper(Item, items, properties = dict( - keywords = relation(Keyword, secondary=item_keywords, lazy=True), - )) - - q = create_session().query(Item) - assert self.static.item_keyword_result == q.all() - - assert self.static.item_keyword_result[0:2] == q.join('keywords').filter(keywords.c.name == 'red').all() - - @testing.resolve_artifact_names - def test_uses_get(self): - """test that a simple many-to-one lazyload optimizes to use query.get().""" - - for pj in ( - None, - users.c.id==addresses.c.user_id, - addresses.c.user_id==users.c.id - ): - mapper(Address, addresses, properties = dict( - user = relation(mapper(User, users), lazy=True, primaryjoin=pj) - )) - - sess = create_session() - - # load address - a1 = sess.query(Address).filter_by(email_address="ed@wood.com").one() - - # load user that is attached to the address - u1 = sess.query(User).get(8) - - def go(): - # lazy load of a1.user should get it from the session - assert a1.user is u1 - self.assert_sql_count(testing.db, go, 0) - sa.orm.clear_mappers() - - @testing.resolve_artifact_names - def test_many_to_one(self): - mapper(Address, addresses, properties = dict( - user = relation(mapper(User, users), lazy=True) - )) - sess = create_session() - q = sess.query(Address) - a = q.filter(addresses.c.id==1).one() - - assert a.user is not None - - u1 = sess.query(User).get(7) - - assert a.user is u1 - - @testing.resolve_artifact_names - def test_backrefs_dont_lazyload(self): - mapper(User, users, properties={ - 'addresses':relation(Address, backref='user') - }) - mapper(Address, addresses) - sess = create_session() - ad = sess.query(Address).filter_by(id=1).one() - assert ad.user.id == 7 - def go(): - ad.user = None - assert ad.user is None - self.assert_sql_count(testing.db, go, 0) - - u1 = sess.query(User).filter_by(id=7).one() - def go(): - assert ad not in u1.addresses - self.assert_sql_count(testing.db, go, 1) - - sess.expire(u1, ['addresses']) - def go(): - assert ad in u1.addresses - self.assert_sql_count(testing.db, go, 1) - - sess.expire(u1, ['addresses']) - ad2 = Address() - def go(): - ad2.user = u1 - assert ad2.user is u1 - self.assert_sql_count(testing.db, go, 0) - - def go(): - assert ad2 in u1.addresses - self.assert_sql_count(testing.db, go, 1) - - -class M2OGetTest(_fixtures.FixtureTest): - run_inserts = 'once' - run_deletes = None - - @testing.resolve_artifact_names - def test_m2o_noload(self): - """test that a NULL foreign key doesn't trigger a lazy load""" - mapper(User, users) - - mapper(Address, addresses, properties={ - 'user':relation(User) - }) - - sess = create_session() - ad1 = Address(email_address='somenewaddress', id=12) - sess.add(ad1) - sess.flush() - sess.expunge_all() - - ad2 = sess.query(Address).get(1) - ad3 = sess.query(Address).get(ad1.id) - def go(): - # one lazy load - assert ad2.user.name == 'jack' - # no lazy load - assert ad3.user is None - self.assert_sql_count(testing.db, go, 1) - -class CorrelatedTest(_base.MappedTest): - - def define_tables(self, meta): - Table('user_t', meta, - Column('id', Integer, primary_key=True), - Column('name', String(50))) - - Table('stuff', meta, - Column('id', Integer, primary_key=True), - Column('date', sa.Date), - Column('user_id', Integer, ForeignKey('user_t.id'))) - - @testing.resolve_artifact_names - def insert_data(self): - user_t.insert().execute( - {'id':1, 'name':'user1'}, - {'id':2, 'name':'user2'}, - {'id':3, 'name':'user3'}) - - stuff.insert().execute( - {'id':1, 'user_id':1, 'date':datetime.date(2007, 10, 15)}, - {'id':2, 'user_id':1, 'date':datetime.date(2007, 12, 15)}, - {'id':3, 'user_id':1, 'date':datetime.date(2007, 11, 15)}, - {'id':4, 'user_id':2, 'date':datetime.date(2008, 1, 15)}, - {'id':5, 'user_id':3, 'date':datetime.date(2007, 6, 15)}) - - @testing.resolve_artifact_names - def test_correlated_lazyload(self): - class User(_base.ComparableEntity): - pass - - class Stuff(_base.ComparableEntity): - pass - - mapper(Stuff, stuff) - - stuff_view = sa.select([stuff.c.id]).where(stuff.c.user_id==user_t.c.id).correlate(user_t).order_by(sa.desc(stuff.c.date)).limit(1) - - mapper(User, user_t, properties={ - 'stuff':relation(Stuff, primaryjoin=sa.and_(user_t.c.id==stuff.c.user_id, stuff.c.id==(stuff_view.as_scalar()))) - }) - - sess = create_session() - - eq_(sess.query(User).all(), [ - User(name='user1', stuff=[Stuff(date=datetime.date(2007, 12, 15), id=2)]), - User(name='user2', stuff=[Stuff(id=4, date=datetime.date(2008, 1 , 15))]), - User(name='user3', stuff=[Stuff(id=5, date=datetime.date(2007, 6, 15))]) - ]) - - -if __name__ == '__main__': - testenv.main() |
