summaryrefslogtreecommitdiff
path: root/test/orm/eager_relations.py
diff options
context:
space:
mode:
authorJason Kirtland <jek@discorporate.us>2008-05-09 20:26:09 +0000
committerJason Kirtland <jek@discorporate.us>2008-05-09 20:26:09 +0000
commite41c0f4107a132b2feac83ba07a25a336e7eae0b (patch)
tree09c785fd5ef9557c3fc926afd7e0a78702dd8023 /test/orm/eager_relations.py
parenta2122a89f6d4f2d3ccc4ba7665cd588c2b0b93b0 (diff)
downloadsqlalchemy-e41c0f4107a132b2feac83ba07a25a336e7eae0b.tar.gz
Test suite modernization in progress. Big changes:
- @unsupported now only accepts a single target and demands a reason for not running the test. - @exclude also demands an exclusion reason - Greatly expanded @testing.requires.<feature>, eliminating many decorators in the suite and signficantly easing integration of multi-driver support. - New ORM test base class, and a featureful base for mapped tests - Usage of 'global' for shared setup going away, * imports as well
Diffstat (limited to 'test/orm/eager_relations.py')
-rw-r--r--test/orm/eager_relations.py598
1 files changed, 352 insertions, 246 deletions
diff --git a/test/orm/eager_relations.py b/test/orm/eager_relations.py
index 544e505f4..b78ba12e9 100644
--- a/test/orm/eager_relations.py
+++ b/test/orm/eager_relations.py
@@ -1,17 +1,18 @@
"""basic tests of eager loaded attributes"""
import testenv; testenv.configure_for_tests()
-from sqlalchemy import *
-from sqlalchemy.orm import *
-from testlib import *
-from testlib.fixtures import *
-from query import QueryTest
-from sqlalchemy.orm import attributes
-
-class EagerTest(FixtureTest):
- keep_mappers = False
- keep_data = True
-
+from testlib import sa, testing
+from sqlalchemy.orm import eagerload, deferred, undefer
+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 EagerTest(_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=False)
@@ -20,11 +21,11 @@ class EagerTest(FixtureTest):
q = sess.query(User)
assert [User(id=7, addresses=[Address(id=1, email_address='jack@bean.com')])] == q.filter(User.id==7).all()
- assert fixtures.user_address_result == q.all()
+ assert self.static.user_address_result == q.all()
+ @testing.resolve_artifact_names
def test_no_orphan(self):
- """test that an eagerly loaded child object is not marked as an orphan"""
-
+ """An eagerly loaded child object is not marked as an orphan"""
mapper(User, users, properties={
'addresses':relation(Address, cascade="all,delete-orphan", lazy=False)
})
@@ -32,9 +33,10 @@ class EagerTest(FixtureTest):
sess = create_session()
user = sess.query(User).get(7)
- assert getattr(User, 'addresses').hasparent(attributes.instance_state(user.addresses[0]), optimistic=True)
- assert not class_mapper(Address)._is_orphan(attributes.instance_state(user.addresses[0]))
+ assert getattr(User, 'addresses').hasparent(sa.orm.attributes.instance_state(user.addresses[0]), optimistic=True)
+ assert not sa.orm.class_mapper(Address)._is_orphan(sa.orm.attributes.instance_state(user.addresses[0]))
+ @testing.resolve_artifact_names
def test_orderby(self):
mapper(User, users, properties = {
'addresses':relation(mapper(Address, addresses), lazy=False, order_by=addresses.c.email_address),
@@ -55,6 +57,7 @@ class EagerTest(FixtureTest):
User(id=10, addresses=[])
] == q.all()
+ @testing.resolve_artifact_names
def test_orderby_multi(self):
mapper(User, users, properties = {
'addresses':relation(mapper(Address, addresses), lazy=False, order_by=[addresses.c.email_address, addresses.c.id]),
@@ -75,11 +78,10 @@ class EagerTest(FixtureTest):
User(id=10, addresses=[])
] == q.all()
+ @testing.resolve_artifact_names
def test_orderby_related(self):
- """tests that a regular mapper select on a single table can order by a relation to a second table"""
-
+ """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=False),
))
@@ -101,11 +103,12 @@ class EagerTest(FixtureTest):
]),
] == l
+ @testing.resolve_artifact_names
def test_orderby_desc(self):
mapper(Address, addresses)
-
mapper(User, users, properties = dict(
- addresses = relation(Address, lazy=False, order_by=[desc(addresses.c.email_address)]),
+ addresses = relation(Address, lazy=False,
+ order_by=[sa.desc(addresses.c.email_address)]),
))
sess = create_session()
assert [
@@ -123,7 +126,13 @@ class EagerTest(FixtureTest):
User(id=10, addresses=[])
] == sess.query(User).all()
+ @testing.resolve_artifact_names
def test_deferred_fk_col(self):
+ User, Address, Dingaling = self.classes.get_all(
+ 'User', 'Address', 'Dingaling')
+ users, addresses, dingalings = self.tables.get_all(
+ 'users', 'addresses', 'dingalings')
+
mapper(Address, addresses, properties={
'user_id':deferred(addresses.c.user_id),
'user':relation(User, lazy=False)
@@ -131,39 +140,43 @@ class EagerTest(FixtureTest):
mapper(User, users)
sess = create_session()
-
+
for q in [
sess.query(Address).filter(Address.id.in_([1, 4, 5])),
sess.query(Address).filter(Address.id.in_([1, 4, 5])).limit(3)
]:
sess.clear()
- self.assertEquals(q.all(),
- [Address(id=1, user=User(id=7)), Address(id=4, user=User(id=8)), Address(id=5, user=User(id=9))]
+ eq_(q.all(),
+ [Address(id=1, user=User(id=7)),
+ Address(id=4, user=User(id=8)),
+ Address(id=5, user=User(id=9))]
)
a = sess.query(Address).filter(Address.id==1).first()
def go():
- assert a.user_id==7
- # assert that the eager loader added 'user_id' to the row
- # and deferred loading of that col was disabled
+ eq_(a.user_id, 7)
+ # assert that the eager loader added 'user_id' to the row and deferred
+ # loading of that col was disabled
self.assert_sql_count(testing.db, go, 0)
# do the mapping in reverse
- # (we would have just used an "addresses" backref but the test fixtures then require the whole
- # backref to be set up, lazy loaders trigger, etc.)
- clear_mappers()
+ # (we would have just used an "addresses" backref but the test
+ # fixtures then require the whole backref to be set up, lazy loaders
+ # trigger, etc.)
+ sa.orm.clear_mappers()
mapper(Address, addresses, properties={
'user_id':deferred(addresses.c.user_id),
})
- mapper(User, users, properties={'addresses':relation(Address, lazy=False)})
-
+ mapper(User, users, properties={
+ 'addresses':relation(Address, lazy=False)})
+
for q in [
sess.query(User).filter(User.id==7),
sess.query(User).filter(User.id==7).limit(1)
]:
sess.clear()
- self.assertEquals(q.all(),
+ eq_(q.all(),
[User(id=7, addresses=[Address(id=1)])]
)
@@ -175,81 +188,115 @@ class EagerTest(FixtureTest):
# and that its still deferred
self.assert_sql_count(testing.db, go, 1)
- clear_mappers()
+ sa.orm.clear_mappers()
- mapper(User, users, properties={'addresses':relation(Address, lazy=False)})
+ mapper(User, users, properties={
+ 'addresses':relation(Address, lazy=False)})
mapper(Address, addresses, properties={
'user_id':deferred(addresses.c.user_id),
- 'dingalings':relation(Dingaling, lazy=False)
- })
+ 'dingalings':relation(Dingaling, lazy=False)})
mapper(Dingaling, dingalings, properties={
- 'address_id':deferred(dingalings.c.address_id)
- })
+ 'address_id':deferred(dingalings.c.address_id)})
sess.clear()
def go():
u = sess.query(User).get(8)
- assert User(id=8, addresses=[Address(id=2, dingalings=[Dingaling(id=1)]), Address(id=3), Address(id=4)]) == u
+ eq_(User(id=8,
+ addresses=[Address(id=2, dingalings=[Dingaling(id=1)]),
+ Address(id=3),
+ Address(id=4)]),
+ u)
self.assert_sql_count(testing.db, go, 1)
+ @testing.resolve_artifact_names
def test_many_to_many(self):
+ Keyword, Item = self.Keyword, self.Item
+ keywords, item_keywords, items = self.tables.get_all(
+ 'keywords', 'item_keywords', 'items')
mapper(Keyword, keywords)
mapper(Item, items, properties = dict(
- keywords = relation(Keyword, secondary=item_keywords, lazy=False, order_by=keywords.c.id),
- ))
+ keywords = relation(Keyword, secondary=item_keywords,
+ lazy=False, order_by=keywords.c.id)))
q = create_session().query(Item)
def go():
- assert fixtures.item_keyword_result == q.all()
+ assert self.static.item_keyword_result == q.all()
self.assert_sql_count(testing.db, go, 1)
def go():
- assert fixtures.item_keyword_result[0:2] == q.join('keywords').filter(Keyword.name == 'red').all()
+ eq_(self.static.item_keyword_result[0:2],
+ q.join('keywords').filter(Keyword.name == 'red').all())
self.assert_sql_count(testing.db, go, 1)
def go():
- assert fixtures.item_keyword_result[0:2] == q.join('keywords', aliased=True).filter(Keyword.name == 'red').all()
+ eq_(self.static.item_keyword_result[0:2],
+ (q.join('keywords', aliased=True).
+ filter(Keyword.name == 'red')).all())
self.assert_sql_count(testing.db, go, 1)
-
+ @testing.resolve_artifact_names
def test_eager_option(self):
+ Keyword, Item = self.Keyword, self.Item
+ keywords, item_keywords, items = self.tables.get_all(
+ 'keywords', 'item_keywords', 'items')
+
mapper(Keyword, keywords)
mapper(Item, items, properties = dict(
- keywords = relation(Keyword, secondary=item_keywords, lazy=True, order_by=keywords.c.id),
- ))
+ keywords = relation(Keyword, secondary=item_keywords, lazy=True,
+ order_by=keywords.c.id)))
q = create_session().query(Item)
def go():
- assert fixtures.item_keyword_result[0:2] == q.options(eagerload('keywords')).join('keywords').filter(keywords.c.name == 'red').all()
+ eq_(self.static.item_keyword_result[0:2],
+ (q.options(eagerload('keywords')).
+ join('keywords').filter(keywords.c.name == 'red')).all())
self.assert_sql_count(testing.db, go, 1)
+ @testing.resolve_artifact_names
def test_cyclical(self):
- """test that a circular eager relationship breaks the cycle with a lazy loader"""
+ """A circular eager relationship breaks the cycle with a lazy loader"""
+ User, Address = self.User, self.Address
+ users, addresses = self.tables.get_all('users', 'addresses')
mapper(Address, addresses)
mapper(User, users, properties = dict(
- addresses = relation(Address, lazy=False, backref=backref('user', lazy=False))
+ addresses = relation(Address, lazy=False,
+ backref=sa.orm.backref('user', lazy=False))
))
- assert class_mapper(User).get_property('addresses').lazy is False
- assert class_mapper(Address).get_property('user').lazy is False
+ assert sa.orm.class_mapper(User).get_property('addresses').lazy is False
+ assert sa.orm.class_mapper(Address).get_property('user').lazy is False
sess = create_session()
- assert fixtures.user_address_result == sess.query(User).all()
+ assert self.static.user_address_result == sess.query(User).all()
+ @testing.resolve_artifact_names
def test_double(self):
- """tests eager loading with two relations simulatneously, from the same table, using aliases. """
- openorders = alias(orders, 'openorders')
- closedorders = alias(orders, 'closedorders')
+ """Eager loading with two relations simultaneously, from the same table, using aliases."""
+ User, Address, Order = self.classes.get_all(
+ 'User', 'Address', 'Order')
+ users, addresses, orders = self.tables.get_all(
+ 'users', 'addresses', 'orders')
+
+ openorders = sa.alias(orders, 'openorders')
+ closedorders = sa.alias(orders, 'closedorders')
mapper(Address, addresses)
mapper(User, users, properties = dict(
addresses = relation(Address, lazy=False),
- open_orders = relation(mapper(Order, openorders, entity_name='open'), primaryjoin = and_(openorders.c.isopen == 1, users.c.id==openorders.c.user_id), lazy=False),
- closed_orders = relation(mapper(Order, closedorders,entity_name='closed'), primaryjoin = and_(closedorders.c.isopen == 0, users.c.id==closedorders.c.user_id), lazy=False)
- ))
+ open_orders = relation(
+ mapper(Order, openorders, entity_name='open'),
+ primaryjoin=sa.and_(openorders.c.isopen == 1,
+ users.c.id==openorders.c.user_id),
+ lazy=False),
+ closed_orders = relation(
+ mapper(Order, closedorders,entity_name='closed'),
+ primaryjoin=sa.and_(closedorders.c.isopen == 0,
+ users.c.id==closedorders.c.user_id),
+ lazy=False)))
+
q = create_session().query(User)
def go():
@@ -277,48 +324,83 @@ class EagerTest(FixtureTest):
] == q.all()
self.assert_sql_count(testing.db, go, 1)
+ @testing.resolve_artifact_names
def test_double_same_mappers(self):
- """tests eager loading with two relations simulatneously, from the same table, using aliases. """
+ """Eager loading with two relations simulatneously, from the same table, using aliases."""
+ User, Address, Order = self.classes.get_all(
+ 'User', 'Address', 'Order')
+ users, addresses, orders = self.tables.get_all(
+ 'users', 'addresses', 'orders')
mapper(Address, addresses)
mapper(Order, orders, properties={
- 'items':relation(Item, secondary=order_items, lazy=False, order_by=items.c.id),
- })
+ 'items': relation(Item, secondary=order_items, lazy=False,
+ order_by=items.c.id)})
mapper(Item, items)
- mapper(User, users, properties = dict(
- addresses = relation(Address, lazy=False),
- open_orders = relation(Order, primaryjoin = and_(orders.c.isopen == 1, users.c.id==orders.c.user_id), lazy=False),
- closed_orders = relation(Order, primaryjoin = and_(orders.c.isopen == 0, users.c.id==orders.c.user_id), lazy=False)
- ))
+ mapper(User, users, properties=dict(
+ addresses=relation(Address, lazy=False),
+ open_orders=relation(
+ Order,
+ primaryjoin=sa.and_(orders.c.isopen == 1,
+ users.c.id==orders.c.user_id),
+ lazy=False),
+ closed_orders=relation(
+ Order,
+ primaryjoin=sa.and_(orders.c.isopen == 0,
+ users.c.id==orders.c.user_id),
+ lazy=False)))
q = create_session().query(User)
def go():
assert [
- User(
- id=7,
- addresses=[Address(id=1)],
- open_orders = [Order(id=3, items=[Item(id=3), Item(id=4), Item(id=5)])],
- closed_orders = [Order(id=1, items=[Item(id=1), Item(id=2), Item(id=3)]), Order(id=5, items=[Item(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, items=[Item(id=1), Item(id=5)])],
- closed_orders = [Order(id=2, items=[Item(id=1), Item(id=2), Item(id=3)])]
- ),
+ User(id=7,
+ addresses=[
+ Address(id=1)],
+ open_orders=[Order(id=3,
+ items=[
+ Item(id=3),
+ Item(id=4),
+ Item(id=5)])],
+ closed_orders=[Order(id=1,
+ items=[
+ Item(id=1),
+ Item(id=2),
+ Item(id=3)]),
+ Order(id=5,
+ items=[
+ Item(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,
+ items=[
+ Item(id=1),
+ Item(id=5)])],
+ closed_orders=[
+ Order(id=2,
+ items=[
+ Item(id=1),
+ Item(id=2),
+ Item(id=3)])]),
User(id=10)
-
] == q.all()
self.assert_sql_count(testing.db, go, 1)
+ @testing.resolve_artifact_names
def test_no_false_hits(self):
- """test that eager loaders don't interpret main table columns as part of their eager load."""
+ """Eager loaders don't interpret main table columns as part of their eager load."""
+ User, Address, Order = self.classes.get_all(
+ 'User', 'Address', 'Order')
+ users, addresses, orders = self.tables.get_all(
+ 'users', 'addresses', 'orders')
mapper(User, users, properties={
'addresses':relation(Address, lazy=False),
@@ -329,16 +411,22 @@ class EagerTest(FixtureTest):
allusers = create_session().query(User).all()
- # using a textual select, the columns will be 'id' and 'name'.
- # the eager loaders have aliases which should not hit on those columns, they should
- # be required to locate only their aliased/fully table qualified column name.
+ # using a textual select, the columns will be 'id' and 'name'. the
+ # eager loaders have aliases which should not hit on those columns,
+ # they should be required to locate only their aliased/fully table
+ # qualified column name.
noeagers = create_session().query(User).from_statement("select * from users").all()
assert 'orders' not in noeagers[0].__dict__
assert 'addresses' not in noeagers[0].__dict__
@testing.fails_on('maxdb')
+ @testing.resolve_artifact_names
def test_limit(self):
- """test limit operations combined with lazy-load relationships."""
+ """Limit operations combined with lazy-load relationships."""
+ User, Item, Address, Order = self.classes.get_all(
+ 'User', 'Item', 'Address', 'Order')
+ users, items, order_items, orders, addresses = self.tables.get_all(
+ 'users', 'items', 'order_items', 'orders', 'addresses')
mapper(Item, items)
mapper(Order, orders, properties={
@@ -354,19 +442,20 @@ class EagerTest(FixtureTest):
if testing.against('mysql'):
l = q.limit(2).all()
- assert fixtures.user_all_result[:2] == l
+ assert self.static.user_all_result[:2] == l
else:
l = q.limit(2).offset(1).order_by(User.id).all()
- print fixtures.user_all_result[1:3]
+ print self.static.user_all_result[1:3]
print l
- assert fixtures.user_all_result[1:3] == l
+ assert self.static.user_all_result[1:3] == l
+ @testing.resolve_artifact_names
def test_distinct(self):
# this is an involved 3x union of the users table to get a lot of rows.
# then see if the "distinct" works its way out. you actually get the same
# result with or without the distinct, just via less or more rows.
u2 = users.alias('u2')
- s = union_all(u2.select(use_labels=True), u2.select(use_labels=True), u2.select(use_labels=True)).alias('u')
+ s = sa.union_all(u2.select(use_labels=True), u2.select(use_labels=True), u2.select(use_labels=True)).alias('u')
mapper(User, users, properties={
'addresses':relation(mapper(Address, addresses), lazy=False),
@@ -377,10 +466,11 @@ class EagerTest(FixtureTest):
def go():
l = q.filter(s.c.u2_id==User.id).distinct().all()
- assert fixtures.user_address_result == l
+ assert self.static.user_address_result == l
self.assert_sql_count(testing.db, go, 1)
@testing.fails_on('maxdb')
+ @testing.resolve_artifact_names
def test_limit_2(self):
mapper(Keyword, keywords)
mapper(Item, items, properties = dict(
@@ -392,9 +482,10 @@ class EagerTest(FixtureTest):
l = q.filter((Item.description=='item 2') | (Item.description=='item 5') | (Item.description=='item 3')).\
order_by(Item.id).limit(2).all()
- assert fixtures.item_keyword_result[1:3] == l
+ assert self.static.item_keyword_result[1:3] == l
@testing.fails_on('maxdb')
+ @testing.resolve_artifact_names
def test_limit_3(self):
"""test that the ORDER BY is propigated from the inner select to the outer select, when using the
'wrapped' select statement resulting from the combination of eager loading and limit/offset clauses."""
@@ -434,16 +525,17 @@ class EagerTest(FixtureTest):
)
] == l.all()
+ @testing.resolve_artifact_names
def test_limit_4(self):
# tests the LIMIT/OFFSET aliasing on a mapper against a select. original issue from ticket #904
- sel = select([users, addresses.c.email_address], users.c.id==addresses.c.user_id).alias('useralias')
+ sel = sa.select([users, addresses.c.email_address], users.c.id==addresses.c.user_id).alias('useralias')
mapper(User, sel, properties={
'orders':relation(Order, primaryjoin=sel.c.id==orders.c.user_id, lazy=False)
})
mapper(Order, orders)
sess = create_session()
- self.assertEquals(sess.query(User).first(),
+ eq_(sess.query(User).first(),
User(name=u'jack',orders=[
Order(address_id=1,description=u'order 1',isopen=0,user_id=7,id=1),
Order(address_id=1,description=u'order 3',isopen=1,user_id=7,id=3),
@@ -451,6 +543,7 @@ class EagerTest(FixtureTest):
email_address=u'jack@bean.com',id=7)
)
+ @testing.resolve_artifact_names
def test_one_to_many_scalar(self):
mapper(User, users, properties = dict(
address = relation(mapper(Address, addresses), lazy=False, uselist=False)
@@ -463,6 +556,7 @@ class EagerTest(FixtureTest):
self.assert_sql_count(testing.db, go, 1)
@testing.fails_on('maxdb')
+ @testing.resolve_artifact_names
def test_many_to_one(self):
mapper(Address, addresses, properties = dict(
user = relation(mapper(User, users), lazy=False)
@@ -477,7 +571,7 @@ class EagerTest(FixtureTest):
assert a.user is u1
self.assert_sql_count(testing.db, go, 1)
-
+ @testing.resolve_artifact_names
def test_one_and_many(self):
"""tests eager load for a parent object with a child object that
contains a many-to-many relationship to a third object."""
@@ -495,12 +589,12 @@ class EagerTest(FixtureTest):
l = q.filter("users.id in (7, 8, 9)")
def go():
- assert fixtures.user_order_result[0:3] == l.all()
+ assert self.static.user_order_result[0:3] == l.all()
self.assert_sql_count(testing.db, go, 1)
+ @testing.resolve_artifact_names
def test_double_with_aggregate(self):
-
- max_orders_by_user = select([func.max(orders.c.id).label('order_id')], group_by=[orders.c.user_id]).alias('max_orders_by_user')
+ max_orders_by_user = sa.select([sa.func.max(orders.c.id).label('order_id')], group_by=[orders.c.user_id]).alias('max_orders_by_user')
max_orders = orders.select(orders.c.id==max_orders_by_user.c.order_id).alias('max_orders')
@@ -528,6 +622,7 @@ class EagerTest(FixtureTest):
] == q.all()
self.assert_sql_count(testing.db, go, 1)
+ @testing.resolve_artifact_names
def test_wide(self):
mapper(Order, orders, properties={'items':relation(Item, secondary=order_items, lazy=False, order_by=items.c.id)})
mapper(Item, items)
@@ -537,12 +632,13 @@ class EagerTest(FixtureTest):
))
q = create_session().query(User)
l = q.all()
- assert fixtures.user_all_result == q.all()
+ assert self.static.user_all_result == q.all()
+ @testing.resolve_artifact_names
def test_against_select(self):
"""test eager loading of a mapper which is against a select"""
- s = select([orders], orders.c.isopen==1).alias('openorders')
+ s = sa.select([orders], orders.c.isopen==1).alias('openorders')
mapper(Order, s, properties={
'user':relation(User, lazy=False)
@@ -561,6 +657,7 @@ class EagerTest(FixtureTest):
Order(id=3, user=User(id=7)),
] == q.all()
+ @testing.resolve_artifact_names
def test_aliasing(self):
"""test that eager loading uses aliases to insulate the eager load from regular criterion against those tables."""
@@ -569,12 +666,13 @@ class EagerTest(FixtureTest):
))
q = create_session().query(User)
l = q.filter(addresses.c.email_address == 'ed@lala.com').filter(Address.user_id==User.id)
- assert fixtures.user_address_result[1:2] == l.all()
+ assert self.static.user_address_result[1:2] == l.all()
-class AddEntityTest(FixtureTest):
- keep_mappers = False
- keep_data = True
+class AddEntityTest(_fixtures.FixtureTest):
+ run_inserts = 'once'
+ run_deletes = None
+ @testing.resolve_artifact_names
def _assert_result(self):
return [
(
@@ -619,6 +717,7 @@ class AddEntityTest(FixtureTest):
)
]
+ @testing.resolve_artifact_names
def test_mapper_configured(self):
mapper(User, users, properties={
'addresses':relation(Address, lazy=False),
@@ -632,12 +731,13 @@ class AddEntityTest(FixtureTest):
sess = create_session()
- oalias = aliased(Order)
+ oalias = sa.orm.aliased(Order)
def go():
ret = sess.query(User, oalias).join(('orders', oalias)).order_by(User.id, oalias.id).all()
- self.assertEquals(ret, self._assert_result())
+ eq_(ret, self._assert_result())
self.assert_sql_count(testing.db, go, 1)
+ @testing.resolve_artifact_names
def test_options(self):
mapper(User, users, properties={
'addresses':relation(Address),
@@ -651,82 +751,78 @@ class AddEntityTest(FixtureTest):
sess = create_session()
- oalias = aliased(Order)
+ oalias = sa.orm.aliased(Order)
def go():
ret = sess.query(User, oalias).options(eagerload('addresses')).join(('orders', oalias)).order_by(User.id, oalias.id).all()
- self.assertEquals(ret, self._assert_result())
+ eq_(ret, self._assert_result())
self.assert_sql_count(testing.db, go, 6)
sess.clear()
def go():
ret = sess.query(User, oalias).options(eagerload('addresses'), eagerload(oalias.items)).join(('orders', oalias)).order_by(User.id, oalias.id).all()
- self.assertEquals(ret, self._assert_result())
+ eq_(ret, self._assert_result())
self.assert_sql_count(testing.db, go, 1)
-class OrderBySecondaryTest(ORMTest):
+class OrderBySecondaryTest(_base.MappedTest):
def define_tables(self, metadata):
- global a, b, m2m
- m2m = Table('mtom', metadata,
- Column('id', Integer, primary_key=True),
- Column('aid', Integer, ForeignKey('a.id')),
- Column('bid', Integer, ForeignKey('b.id')),
- )
-
- a = Table('a', metadata,
- Column('id', Integer, primary_key=True),
- Column('data', String(50)),
- )
- b = Table('b', metadata,
- Column('id', Integer, primary_key=True),
- Column('data', String(50)),
- )
-
- def insert_data(self):
- a.insert().execute([
- {'id':1, 'data':'a1'},
- {'id':2, 'data':'a2'}
- ])
-
- b.insert().execute([
- {'id':1, 'data':'b1'},
- {'id':2, 'data':'b2'},
- {'id':3, 'data':'b3'},
- {'id':4, 'data':'b4'},
- ])
-
- m2m.insert().execute([
- {'id':2, 'aid':1, 'bid':1},
- {'id':4, 'aid':2, 'bid':4},
- {'id':1, 'aid':1, 'bid':3},
- {'id':6, 'aid':2, 'bid':2},
- {'id':3, 'aid':1, 'bid':2},
- {'id':5, 'aid':2, 'bid':3},
- ])
-
+ Table('m2m', metadata,
+ Column('id', Integer, primary_key=True),
+ Column('aid', Integer, ForeignKey('a.id')),
+ Column('bid', Integer, ForeignKey('b.id')))
+
+ Table('a', metadata,
+ Column('id', Integer, primary_key=True),
+ Column('data', String(50)))
+ Table('b', metadata,
+ Column('id', Integer, primary_key=True),
+ Column('data', String(50)))
+
+ def fixtures(self):
+ return dict(
+ a=(('id', 'data'),
+ (1, 'a1'),
+ (2, 'a2')),
+
+ b=(('id', 'data'),
+ (1, 'b1'),
+ (2, 'b2'),
+ (3, 'b3'),
+ (4, 'b4')),
+
+ m2m=(('id', 'aid', 'bid'),
+ (2, 1, 1),
+ (4, 2, 4),
+ (1, 1, 3),
+ (6, 2, 2),
+ (3, 1, 2),
+ (5, 2, 3)))
+
+ @testing.resolve_artifact_names
def test_ordering(self):
- class A(Base):pass
- class B(Base):pass
-
+ class A(_base.ComparableEntity):pass
+ class B(_base.ComparableEntity):pass
+
mapper(A, a, properties={
'bs':relation(B, secondary=m2m, lazy=False, order_by=m2m.c.id)
})
mapper(B, b)
-
+
sess = create_session()
- self.assertEquals(sess.query(A).all(), [A(data='a1', bs=[B(data='b3'), B(data='b1'), B(data='b2')]), A(bs=[B(data='b4'), B(data='b3'), B(data='b2')])])
-
-
-class SelfReferentialEagerTest(ORMTest):
+ eq_(sess.query(A).all(), [A(data='a1', bs=[B(data='b3'), B(data='b1'), B(data='b2')]), A(bs=[B(data='b4'), B(data='b3'), B(data='b2')])])
+
+
+class SelfReferentialEagerTest(_base.MappedTest):
def define_tables(self, metadata):
- global nodes
- nodes = Table('nodes', metadata,
- Column('id', Integer, Sequence('node_id_seq', optional=True), primary_key=True),
+ Table('nodes', metadata,
+ Column('id', Integer, sa.Sequence('node_id_seq', optional=True),
+ primary_key=True),
Column('parent_id', Integer, ForeignKey('nodes.id')),
Column('data', String(30)))
@testing.fails_on('maxdb')
+ @testing.resolve_artifact_names
def test_basic(self):
- class Node(Base):
+ class Node(_base.ComparableEntity):
def append(self, node):
self.children.append(node)
@@ -758,8 +854,9 @@ class SelfReferentialEagerTest(ORMTest):
self.assert_sql_count(testing.db, go, 1)
+ @testing.resolve_artifact_names
def test_lazy_fallback_doesnt_affect_eager(self):
- class Node(Base):
+ class Node(_base.ComparableEntity):
def append(self, node):
self.children.append(node)
@@ -778,14 +875,14 @@ class SelfReferentialEagerTest(ORMTest):
sess.flush()
sess.clear()
- # eager load with join depth 1. when eager load of 'n1'
- # hits the children of 'n12', no columns are present, eager loader
- # degrades to lazy loader; fine. but then, 'n12' is *also* in the
- # first level of columns since we're loading the whole table.
- # when those rows arrive, now we *can* eager load its children and an
- # eager collection should be initialized. essentially the 'n12' instance
- # is present in not just two different rows but two distinct sets of columns
- # in this result set.
+ # eager load with join depth 1. when eager load of 'n1' hits the
+ # children of 'n12', no columns are present, eager loader degrades to
+ # lazy loader; fine. but then, 'n12' is *also* in the first level of
+ # columns since we're loading the whole table. when those rows
+ # arrive, now we *can* eager load its children and an eager collection
+ # should be initialized. essentially the 'n12' instance is present in
+ # not just two different rows but two distinct sets of columns in this
+ # result set.
def go():
allnodes = sess.query(Node).order_by(Node.data).all()
n12 = allnodes[2]
@@ -799,8 +896,9 @@ class SelfReferentialEagerTest(ORMTest):
] == list(n12.children)
self.assert_sql_count(testing.db, go, 1)
+ @testing.resolve_artifact_names
def test_with_deferred(self):
- class Node(Base):
+ class Node(_base.ComparableEntity):
def append(self, node):
self.children.append(node)
@@ -833,9 +931,9 @@ class SelfReferentialEagerTest(ORMTest):
self.assert_sql_count(testing.db, go, 1)
-
+ @testing.resolve_artifact_names
def test_options(self):
- class Node(Base):
+ class Node(_base.ComparableEntity):
def append(self, node):
self.children.append(node)
@@ -881,8 +979,9 @@ class SelfReferentialEagerTest(ORMTest):
])
@testing.fails_on('maxdb')
+ @testing.resolve_artifact_names
def test_no_depth(self):
- class Node(Base):
+ class Node(_base.ComparableEntity):
def append(self, node):
self.children.append(node)
@@ -913,22 +1012,22 @@ class SelfReferentialEagerTest(ORMTest):
]) == d
self.assert_sql_count(testing.db, go, 3)
-class SelfReferentialM2MEagerTest(ORMTest):
+class SelfReferentialM2MEagerTest(_base.MappedTest):
def define_tables(self, metadata):
- global widget, widget_rel
-
- widget = Table('widget', metadata,
+ Table('widget', metadata,
Column('id', Integer, primary_key=True),
- Column('name', Unicode(40), nullable=False, unique=True),
+ Column('name', sa.Unicode(40), nullable=False, unique=True),
)
- widget_rel = Table('widget_rel', metadata,
+ Table('widget_rel', metadata,
Column('parent_id', Integer, ForeignKey('widget.id')),
Column('child_id', Integer, ForeignKey('widget.id')),
- UniqueConstraint('parent_id', 'child_id'),
+ sa.UniqueConstraint('parent_id', 'child_id'),
)
+
+ @testing.resolve_artifact_names
def test_basic(self):
- class Widget(Base):
+ class Widget(_base.ComparableEntity):
pass
mapper(Widget, widget, properties={
@@ -949,10 +1048,12 @@ class SelfReferentialM2MEagerTest(ORMTest):
assert [Widget(name='w1', children=[Widget(name='w2')])] == sess.query(Widget).filter(Widget.name==u'w1').all()
-class MixedEntitiesTest(FixtureTest, AssertsCompiledSQL):
- keep_mappers = True
- keep_data = True
-
+class MixedEntitiesTest(_fixtures.FixtureTest, testing.AssertsCompiledSQL):
+ run_setup_mappers = 'once'
+ run_inserts = 'once'
+ run_deletes = None
+
+ @testing.resolve_artifact_names
def setup_mappers(self):
mapper(User, users, properties={
'addresses':relation(Address, backref='user'),
@@ -966,13 +1067,14 @@ class MixedEntitiesTest(FixtureTest, AssertsCompiledSQL):
'keywords':relation(Keyword, secondary=item_keywords) #m2m
})
mapper(Keyword, keywords)
-
+
+ @testing.resolve_artifact_names
def test_two_entities(self):
sess = create_session()
# two FROM clauses
def go():
- self.assertEquals(
+ eq_(
[
(User(id=9, addresses=[Address(id=5)]), Order(id=2, items=[Item(id=1), Item(id=2), Item(id=3)])),
(User(id=9, addresses=[Address(id=5)]), Order(id=4, items=[Item(id=1), Item(id=5)])),
@@ -985,7 +1087,7 @@ class MixedEntitiesTest(FixtureTest, AssertsCompiledSQL):
# one FROM clause
def go():
- self.assertEquals(
+ eq_(
[
(User(id=9, addresses=[Address(id=5)]), Order(id=2, items=[Item(id=1), Item(id=2), Item(id=3)])),
(User(id=9, addresses=[Address(id=5)]), Order(id=4, items=[Item(id=1), Item(id=5)])),
@@ -994,15 +1096,16 @@ class MixedEntitiesTest(FixtureTest, AssertsCompiledSQL):
order_by(User.id, Order.id).all(),
)
self.assert_sql_count(testing.db, go, 1)
-
+
+ @testing.resolve_artifact_names
def test_aliased_entity(self):
sess = create_session()
-
- oalias = aliased(Order)
-
+
+ oalias = sa.orm.aliased(Order)
+
# two FROM clauses
def go():
- self.assertEquals(
+ eq_(
[
(User(id=9, addresses=[Address(id=5)]), Order(id=2, items=[Item(id=1), Item(id=2), Item(id=3)])),
(User(id=9, addresses=[Address(id=5)]), Order(id=4, items=[Item(id=1), Item(id=5)])),
@@ -1015,7 +1118,7 @@ class MixedEntitiesTest(FixtureTest, AssertsCompiledSQL):
# one FROM clause
def go():
- self.assertEquals(
+ eq_(
[
(User(id=9, addresses=[Address(id=5)]), Order(id=2, items=[Item(id=1), Item(id=2), Item(id=3)])),
(User(id=9, addresses=[Address(id=5)]), Order(id=4, items=[Item(id=1), Item(id=5)])),
@@ -1024,12 +1127,13 @@ class MixedEntitiesTest(FixtureTest, AssertsCompiledSQL):
order_by(User.id, oalias.id).all(),
)
self.assert_sql_count(testing.db, go, 1)
-
+
from sqlalchemy.engine.default import DefaultDialect
-
- # improper setup: oalias in the columns clause but join to usual orders alias.
- # this should create two FROM clauses even though the query has a from_clause set up via the join
- self.assert_compile(sess.query(User, oalias).join(User.orders).options(eagerload(oalias.items)).with_labels().statement,
+
+ # improper setup: oalias in the columns clause but join to usual
+ # orders alias. this should create two FROM clauses even though the
+ # query has a from_clause set up via the join
+ self.assert_compile(sess.query(User, oalias).join(User.orders).options(eagerload(oalias.items)).with_labels().statement,
"SELECT users.id AS users_id, users.name AS users_name, orders_1.id AS orders_1_id, "\
"orders_1.user_id AS orders_1_user_id, orders_1.address_id AS orders_1_address_id, "\
"orders_1.description AS orders_1_description, orders_1.isopen AS orders_1_isopen, items_1.id AS items_1_id, "\
@@ -1038,23 +1142,23 @@ class MixedEntitiesTest(FixtureTest, AssertsCompiledSQL):
"LEFT OUTER JOIN items AS items_1 ON items_1.id = order_items_1.item_id ORDER BY users.id, items_1.id",
dialect=DefaultDialect()
)
-
-class CyclicalInheritingEagerTest(ORMTest):
+
+class CyclicalInheritingEagerTest(_base.MappedTest):
def define_tables(self, metadata):
- global t1, t2
- t1 = Table('t1', metadata,
+ Table('t1', metadata,
Column('c1', Integer, primary_key=True),
Column('c2', String(30)),
Column('type', String(30))
)
- t2 = Table('t2', metadata,
+ Table('t2', metadata,
Column('c1', Integer, primary_key=True),
Column('c2', String(30)),
Column('type', String(30)),
Column('t1.id', Integer, ForeignKey('t1.c1')))
+ @testing.resolve_artifact_names
def test_basic(self):
class T(object):
pass
@@ -1070,7 +1174,7 @@ class CyclicalInheritingEagerTest(ORMTest):
mapper(T, t1, polymorphic_on=t1.c.type, polymorphic_identity='t1')
mapper(SubT, None, inherits=T, polymorphic_identity='subt1', properties={
- 't2s':relation(SubT2, lazy=False, backref=backref('subt', lazy=False))
+ 't2s':relation(SubT2, lazy=False, backref=sa.orm.backref('subt', lazy=False))
})
mapper(T2, t2, polymorphic_on=t2.c.type, polymorphic_identity='t2')
mapper(SubT2, None, inherits=T2, polymorphic_identity='subt2')
@@ -1078,65 +1182,66 @@ class CyclicalInheritingEagerTest(ORMTest):
# testing a particular endless loop condition in eager join setup
create_session().query(SubT).all()
-class SubqueryTest(ORMTest):
+class SubqueryTest(_base.MappedTest):
def define_tables(self, metadata):
- global users_table, tags_table
-
- users_table = Table('users', metadata,
+ Table('users_table', metadata,
Column('id', Integer, primary_key=True),
Column('name', String(16))
)
- tags_table = Table('tags', metadata,
+ Table('tags_table', metadata,
Column('id', Integer, primary_key=True),
- Column('user_id', Integer, ForeignKey("users.id")),
- Column('score1', Float),
- Column('score2', Float),
+ Column('user_id', Integer, ForeignKey("users_table.id")),
+ Column('score1', sa.Float),
+ Column('score2', sa.Float),
)
+ @testing.resolve_artifact_names
def test_label_anonymizing(self):
- """test that eager loading works with subqueries with labels,
- even if an explicit labelname which conflicts with a label on the parent.
-
- There's not much reason a column_property() would ever need to have a label
- of a specific name (and they don't even need labels these days),
- unless you'd like the name to line up with a name
- that you may be using for a straight textual statement used for loading
- instances of that type.
-
+ """Eager loading works with subqueries with labels,
+
+ Even if an explicit labelname which conflicts with a label on the
+ parent.
+
+ There's not much reason a column_property() would ever need to have a
+ label of a specific name (and they don't even need labels these days),
+ unless you'd like the name to line up with a name that you may be
+ using for a straight textual statement used for loading instances of
+ that type.
+
"""
- class User(Base):
+ class User(_base.ComparableEntity):
@property
def prop_score(self):
return sum([tag.prop_score for tag in self.tags])
- class Tag(Base):
+ class Tag(_base.ComparableEntity):
@property
def prop_score(self):
return self.score1 * self.score2
-
+
for labeled, labelname in [(True, 'score'), (True, None), (False, None)]:
- clear_mappers()
-
+ sa.orm.clear_mappers()
+
tag_score = (tags_table.c.score1 * tags_table.c.score2)
- user_score = select([func.sum(tags_table.c.score1 *
- tags_table.c.score2)],
- tags_table.c.user_id == users_table.c.id)
-
+ user_score = sa.select([sa.func.sum(tags_table.c.score1 *
+ tags_table.c.score2)],
+ tags_table.c.user_id == users_table.c.id)
+
if labeled:
tag_score = tag_score.label(labelname)
user_score = user_score.label(labelname)
else:
user_score = user_score.as_scalar()
-
+
mapper(Tag, tags_table, properties={
- 'query_score': column_property(tag_score),
+ 'query_score': sa.orm.column_property(tag_score),
})
mapper(User, users_table, properties={
- 'tags': relation(Tag, backref='user', lazy=False),
- 'query_score': column_property(user_score),
+ 'tags': relation(Tag, backref='user', lazy=False),
+ 'query_score': sa.orm.column_property(user_score),
})
session = create_session()
@@ -1144,17 +1249,18 @@ class SubqueryTest(ORMTest):
session.save(User(name='bar', tags=[Tag(score1=5.0, score2=4.0), Tag(score1=50.0, score2=1.0), Tag(score1=15.0, score2=2.0)]))
session.flush()
session.clear()
-
+
for user in session.query(User).all():
- self.assertEquals(user.query_score, user.prop_score)
+ eq_(user.query_score, user.prop_score)
def go():
u = session.query(User).filter_by(name='joe').one()
- self.assertEquals(u.query_score, u.prop_score)
+ eq_(u.query_score, u.prop_score)
self.assert_sql_count(testing.db, go, 1)
-
+
for t in (tags_table, users_table):
t.delete().execute()
-
+
+
if __name__ == '__main__':
testenv.main()